""" Amazon Prime Marketshare Workflow. Ingest Amazon Prime marketshare data into marketshare table in Snowflake. """ 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.amazon_prime_marketshare import config from feed_ingestion.flows.amazon_prime_marketshare import tasks from feed_ingestion.tasks import date_tasks from feed_ingestion.tasks import feed_status_tasks from feed_ingestion.tasks import ftp_amazon_marketshare_tasks from feed_ingestion.tasks import load_raw_table_tasks_sf from feed_ingestion.tasks import overall_status_tasks class Flow( base.FlowBase, base.FlowConfigMixin, base.FlowLoadRawMixinSF, base.FlowLoadMarketshareMixinSF): """Amazon Prime Marketshare Flow class. This flow downloads Amazon Prime marketshare data and loads to a main_market_share. """ def __init__(self): """Initialize an Amazon Prime Marketshare workflow.""" super(Flow, self).__init__(config.feed_name, config.feed_version) def decider(self, schedule): """Orchestrate an Amazon Prime Marketshare workflow. Args: schedule (callable): The scheduler method. """ get_first_day_of_month = schedule( 'get_first_day_of_month', self.get_first_day_of_month) bootstrap = schedule( 'bootstrap', self.bootstrap, requires=[get_first_day_of_month]) check_for_new_files = schedule( 'check_for_new_files', self.check_for_new_files, requires=[bootstrap]) if check_for_new_files.result.get('check_for_new_files.stop'): return fetch_from_drop_location = schedule( 'fetch_from_drop_location', self.fetch_from_drop_location, requires=[check_for_new_files]) if fetch_from_drop_location.result.get( 'fetch_from_drop_location.stop'): return # Status updated to DOWNLOADED if files are present, else NOT_AVAILABLE set_status_to_downloaded = schedule( 'set_status_to_downloaded', self.set_status_to_downloaded, requires=[fetch_from_drop_location]) reset_dynamo_db_status = schedule( 'reset_dynamo_db_status', self.reset_dynamo_db_status, requires=[set_status_to_downloaded]) process_drop_files = schedule( 'process_drop_files', self.process_drop_files, requires=[reset_dynamo_db_status]) create_temp_staging_raw_table = schedule( 'create_temp_staging_raw_table', self.create_temp_staging_raw_table, requires=[process_drop_files]) load_temp_staging_raw_table = schedule( 'load_temp_staging_raw_table', self.load_temp_staging_raw_table, requires=[create_temp_staging_raw_table]) load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[load_temp_staging_raw_table]) set_status_to_populated_raw_table = schedule( 'set_status_to_populated_raw_table', self.set_status_to_populated_raw_table, requires=[load_staging_raw_table]) load_marketshare_table = schedule( 'load_market_share_table', self.load_marketshare_table, requires=[set_status_to_populated_raw_table]) mark_ingested_files = schedule( 'mark_ingested_files', self.mark_ingested_files, requires=[load_marketshare_table], ) schedule( 'set_status_to_ingested', self.set_status_to_ingested, requires=[mark_ingested_files], ) @property def get_first_day_of_month(self): """Get the first day of the month of date.""" return self.create( name='get_first_day_of_month', tasks=base.SyncRunner( date_tasks.get_first_day_of_month.fill( namespace='get_first_day_of_month', date='context_date'))) @property def bootstrap(self): """Bootstrap initial configuration.""" return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap', date='get_first_day_of_month.date', reload='reload'))) @property def check_for_new_files(self): """Check if there are some new files on FTP.""" return self.create( name='check_for_new_files', tasks=base.SyncRunner( ftp_amazon_marketshare_tasks.check_for_new_files.fill( namespace='check_for_new_files', feed_name='bootstrap.feed_name', date='bootstrap.date', file_pattern='bootstrap.file_pattern', ftp_creds=StaticParam(config.sftp), ftp_path_template=StaticParam(config.sftp_path_template)))) @property def fetch_from_drop_location(self): """Move files from FTP to S3 archive bucket.""" return self.create( name='fetch_from_drop_location', tasks=base.SyncRunner( ftp_amazon_marketshare_tasks.fetch_from_drop_location.fill( namespace='fetch_from_drop_location', date='bootstrap.date', s3_archive_path='bootstrap.s3_archive_path', files_on_ftp='check_for_new_files.files_on_ftp', ftp_creds=StaticParam(config.sftp)))) @property def process_drop_files(self): """Process and archive drop files on s3.""" return self.create( name='process_drop_files', tasks=base.SyncRunner( tasks.process_drop_files.fill( namespace='process_drop_files', date='bootstrap.date', s3_archive_path='bootstrap.s3_archive_path', s3_preprocessed_path='bootstrap.s3_preprocessed_path', processed_filename='bootstrap.processed_filename', source_files_dict=( 'fetch_from_drop_location.source_files_dict')))) @property def set_status_to_downloaded(self): """Set overall feed status to DOWNLOADED.""" return self.create( name='set_status_to_downloaded', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_downloaded', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam( garcon_feed_status.STATUS_DOWNLOADED)))) @property def create_temp_staging_raw_table(self): """Create temp staging raw table.""" return self.create( name='create_temp_staging_raw_table', tasks=base.SyncRunner( load_raw_table_tasks_sf.create_temp_staging_raw_table.fill( namespace='create_temp_staging_raw_table', date='bootstrap.date', sfdb_params='sfdb_params', feed_name='bootstrap.feed_name', secrets_path='bootstrap.secrets_path', temp_staging_raw_table='bootstrap.temp_staging_raw_table', kwargs='bootstrap.kwargs'))) @property def set_status_to_populated_raw_table(self): """Set overall feed status to POPULATED_RAW_TABLE.""" return self.create( name='set_status_to_downloaded', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_downloaded', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam( garcon_feed_status.STATUS_POPULATED_RAW_TABLE)))) @property def mark_ingested_files(self): """Put list of injected files in DynamoD.""" return self.create( name='mark_ingested_files', tasks=base.SyncRunner( feed_status_tasks.mark_ingested_files.fill( namespace='mark_ingested_files', feed_name='bootstrap.feed_name', date='bootstrap.date', source_files_dict=( 'fetch_from_drop_location.source_files_dict')))) @property def set_status_to_ingested(self): """Set overall feed status to INGESTED.""" return self.create( name='set_status_to_ingested', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_ingested', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam( garcon_feed_status.STATUS_INGESTED, ), ), ), )