""" Proper Music Distribution Ingestion Workflow. Ingest data from Proper v1 Feed. """ import datetime from garcon.param import StaticParam from garcon_contrib.dynamo_feed_status import \ garcon_feed_status from feed_ingestion.flows import base from feed_ingestion.flows.proper_incoming import config from feed_ingestion.flows.proper_incoming import tasks from feed_ingestion.tasks import overall_status_tasks class Flow(base.FlowBase, base.FlowConfigMixin): """Proper Music Distribution Streams Flow class. This flow download the Proper report data from FTP, loads to a S3 and ingests Stock_ESSN file to MySQL table, and Shortages and GoodsIn to Snowflake. (No data is loaded into fact_analytics). """ def __init__(self): """Initialize flow object.""" super(Flow, self).__init__(feed_name=config.feed_name, version='1.0') self.timeout = 21600 # override default to 6 hours (60 * 60 * 6) def _get_proper_run_params_from_context(self, context): """Handle date and proper_feed_type parameters from context. Context must contains 'proper_feed_type' parameter, otherwise KeyError exception will ber raised. The 'proper_feed_type' parameter value can be one of ['goodsin', 'sales', 'shortages', 'stock'], otherwise ValueError exception will be raised. If context contains 'context_date' parameter it will be returned as is, otherwise the value will be set to current date in YYYY-MM-DD format. Args: context (dict): The initial context for the flow. Returns: (str, str): Tuple of date and proper_feed_type in str type. Raises: KeyError: If context doesn't contain 'proper_feed_type' parameter. ValueError: Value of 'proper_feed_type' parameter is incorrect. """ if 'context_date' not in context: date = datetime.datetime.today().strftime('%Y-%m-%d') else: date = context['context_date'] if 'proper_feed_type' in context: proper_feed_type = context['proper_feed_type'] else: raise KeyError('Missing proper_feed_type parameter') if proper_feed_type not in config.proper_feeds.keys(): raise ValueError( 'Incorrect feed type. Must be on of {feed_types}'.format( feed_types=config.proper_feeds.keys())) return date, proper_feed_type def workflow_id(self, initial_context): """Generate workflow id. This method is overridden from FlowBase class. The only difference is using passed to workflow context 'proper_feed_type' param instead of self.name for generation workflow_id value. Args: initial_context (dict): the initial context for the flow. Returns: str: a unique identifier for a workflow being executed In the forms of '-YYYY-MM-DD', where proper_feed_type is passed by context value and YYYY-MM-DD is the context date or, if none passed the current date. """ date, proper_feed_type = self._get_proper_run_params_from_context( initial_context) workflow_id = '{flow_type}-{date}'.format( flow_type=proper_feed_type, date=date) return workflow_id def decider(self, schedule, context): """Activity decider. Args: schedule (callable): The scheduler method. context (dict): Initial context of workflow. """ # handle run params for detection of flow type date, proper_feed_type = self._get_proper_run_params_from_context( context) bootstrap = schedule('bootstrap', self.bootstrap) fetch_from_drop_location = schedule( 'fetch_from_drop_location', self.fetch_from_drop_location, requires=[bootstrap]) # stops the flow if it has INGESTED status in DynamoDB if fetch_from_drop_location.result.get( 'fetch_from_drop_location.stop'): return # file header and rows types validation validate_csv_file = schedule( 'validate_csv_file', self.validate_csv_file, requires=[fetch_from_drop_location]) # processing Stock_ESSN file if proper_feed_type == config.FEED_TYPE_STOCK: # uploading Stock_ESSN to MySQL transform_csv_file_stock = schedule( 'transform_csv_file_stock', self.transform_csv_file_stock, requires=[validate_csv_file]) ingest_stock_data_into_mysql_table = schedule( 'ingest_stock_data_into_mysql_table', self.ingest_stock_data_into_mysql_table, requires=[transform_csv_file_stock]) # update proper_stock_essn MySQL table with release_id values add_release_id_to_proper_stock_essn_table = schedule( 'add_release_id_to_proper_stock_essn_table', self.add_release_id_to_proper_stock_essn_table, requires=[ingest_stock_data_into_mysql_table]) # set INGESTED_TO_MYSQL status for run schedule( 'set_overall_status_to_ingested', self.set_overall_status_to_ingested, requires=[add_release_id_to_proper_stock_essn_table]) return if proper_feed_type in [config.FEED_TYPE_GOODSIN, config.FEED_TYPE_SHORTAGES]: transform_csv_file = schedule( 'transform_csv_file', self.transform_csv_file, requires=[validate_csv_file]) purge_old_data_from_snowflake = schedule( 'purge_old_data_from_snowflake', self.purge_old_data_from_snowflake, requires=[transform_csv_file]) ingest_data_into_snowflake = schedule( 'ingest_data_into_snowflake', self.ingest_data_into_snowflake, requires=[purge_old_data_from_snowflake]) schedule( 'set_overall_status_to_ingested', self.set_overall_status_to_ingested, requires=[ingest_data_into_snowflake]) return if proper_feed_type == config.FEED_TYPE_SALES: # set INGESTED status for success run schedule( 'set_overall_status_to_ingested', self.set_overall_status_to_ingested, requires=[validate_csv_file]) return @property def bootstrap(self): """Bootstrap initial configuration.""" return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap_feed', date='context_date', proper_feed_type='proper_feed_type'))) @property def fetch_from_drop_location(self): """Move file from FTP to an archive folder in a S3 bucket.""" return self.create( name='fetch_from_drop_location', tasks=base.SyncRunner( tasks.fetch_from_drop_location.fill( namespace='fetch_from_drop_location', date='bootstrap_feed.date', feed_name='bootstrap_feed.feed_name', ftp_path=StaticParam(config.ftp.get('path')), ftp_file_name='bootstrap_feed.ftp_file_name', s3_path='bootstrap_feed.s3_path', s3_file_name='bootstrap_feed.s3_file_name'))) @property def validate_csv_file(self): """Validate CSV file against schema.""" return self.create( name='validate_csv_file', tasks=base.SyncRunner( tasks.validate_csv_file.fill( namespace='validate_csv_file', date='bootstrap_feed.date', feed_name='bootstrap_feed.feed_name', proper_feed_type='bootstrap_feed.proper_feed_type', s3_file_full_path='bootstrap_feed.s3_file_full_path', expected_bucket_owner=StaticParam( config.expected_bucket_owner)))) @property def ingest_stock_data_into_mysql_table(self): """Ingest CSV files to the corresponding MySQL tables. (For now it ingests only Stock_ESSN file). """ return self.create( name='ingest_stock_data_into_mysql_table', tasks=base.SyncRunner( tasks.ingest_stock_data_into_mysql_table.fill( namespace='ingest_stock_data_into_mysql_table', s3_file_full_path=( 'transform_csv_file_stock.s3_file_full_path'), feed_name='bootstrap_feed.feed_name', proper_feed_type='bootstrap_feed.proper_feed_type', date='bootstrap_feed.date'))) @property def add_release_id_to_proper_stock_essn_table(self): """Update art_relations.proper_stock_essn.release_id.""" return self.create( name='add_release_id_to_proper_stock_essn_table', tasks=base.SyncRunner( tasks.add_release_id_to_proper_stock_essn_table.fill( namespace='add_release_id_to_proper_stock_essn_table', feed_name='bootstrap_feed.feed_name', date='bootstrap_feed.date'))) @property def transform_csv_file(self): """Extend passed CSV file with 3 additional columns. These columns are: file_date, file_name, ingestion_timestamp. """ return self.create( name='transform_csv_file', tasks=base.SyncRunner( tasks.transform_csv_file.fill( namespace='transform_csv_file', proper_feed_type='bootstrap_feed.proper_feed_type', feed_name='bootstrap_feed.feed_name', date='bootstrap_feed.date', file_to_transform='bootstrap_feed.s3_file_full_path'))) @property def transform_csv_file_stock(self): """Create a new csv file without rows with incorrect label code.""" return self.create( name='transform_csv_file_stock', tasks=base.SyncRunner( tasks.transform_csv_file_stock.fill( namespace='transform_csv_file_stock', proper_feed_type='bootstrap_feed.proper_feed_type', feed_name='bootstrap_feed.feed_name', date='bootstrap_feed.date', s3_file_full_path='bootstrap_feed.s3_file_full_path'))) @property def purge_old_data_from_snowflake(self): """Cleanup Snowflake table before data upload.""" return self.create( name='purge_old_data_from_snowflake', tasks=base.SyncRunner( tasks.purge_old_data_from_snowflake.fill( namespace='purge_old_data_from_snowflake', proper_feed_type='bootstrap_feed.proper_feed_type', feed_name='bootstrap_feed.feed_name', date='bootstrap_feed.date', file_to_ingest_to_snowflake=( 'bootstrap_feed.s3_file_name')))) @property def ingest_data_into_snowflake(self): """Copy passed CSV to the Snowflake.""" return self.create( name='ingest_data_into_snowflake', tasks=base.SyncRunner( tasks.load_data_into_snowflake.fill( namespace='ingest_data_into_snowflake', feed_name='bootstrap_feed.feed_name', date='bootstrap_feed.date', file_on_s3='transform_csv_file.s3_file_path', proper_feed_type='bootstrap_feed.proper_feed_type'))) @property def set_overall_status_to_ingested(self): """Set overall status to INGESTED.""" return self.create( name='set_overall_status_to_ingested', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_overall_status_to_ingested', feed_name='bootstrap_feed.feed_name', date='bootstrap_feed.date', status=StaticParam(garcon_feed_status.STATUS_INGESTED))))