"""Spotify Marketshare Week Ending Data Workflow.""" 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.base import FlowBase from feed_ingestion.flows.base import FlowConfigMixin from feed_ingestion.flows.base import FlowLoadMarketshareMixinSF from feed_ingestion.flows.base import FlowLoadRawMixinSF from feed_ingestion.flows.spotify_marketshare_week_ending import config from feed_ingestion.flows.spotify_marketshare_week_ending import tasks from feed_ingestion.tasks import load_raw_table_tasks_sf from feed_ingestion.tasks import overall_status_tasks from feed_ingestion.tasks import s3_tasks class Flow( FlowBase, FlowConfigMixin, FlowLoadRawMixinSF, FlowLoadMarketshareMixinSF): """Spotify Marketshare Flow class. This flow downloads Spotify marketshare data and loads to a main_market_share. """ def __init__(self): """Initialize an Spotify Marketshare workflow.""" super(Flow, self).__init__(config.feed_name, config.feed_version) def decider(self, schedule): """Orchestrate a Spotify Marketshare workflow. Args: schedule (callable): The scheduler method. """ bootstrap = schedule( 'bootstrap', self.bootstrap) fetch_from_drop_location = schedule( 'fetch_from_drop_location', self.fetch_from_drop_location, requires=[bootstrap]) # Stop flow if files aren't available if fetch_from_drop_location.result.get( 'fetch_from_drop_location.stop'): return fetch_from_drop_location.\ result.get('fetch_from_drop_location.' 'download_result') set_status_to_downloaded = schedule( 'set_status_to_downloaded', self.set_status_to_downloaded, requires=[fetch_from_drop_location]) create_temp_staging_raw_table = schedule( 'create_temp_staging_raw_table', self.create_temp_staging_raw_table, requires=[set_status_to_downloaded]) 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]) schedule( 'set_status_to_ingested', self.set_status_to_ingested, requires=[load_staging_raw_table]) @property def bootstrap(self): """Bootstrap initial configuration.""" return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap', date='context_date', reload='reload'))) @property def fetch_from_drop_location(self): """Move file from drop (downloads) folder to archive folder on S3.""" return self.create( name='fetch_from_drop_location', tasks=base.SyncRunner( s3_tasks.copy_files.fill( namespace='fetch_from_drop_location', s3_archive_path='bootstrap.s3_archive_path', s3_download_path='bootstrap.s3_download_path', source_files_dict='bootstrap.source_files_dict', need_all_files=StaticParam(True)))) @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'))) @property def load_temp_staging_raw_table(self): """Load temp staging raw table.""" return self.create( name='load_temp_staging_raw_table', tasks=base.SyncRunner( load_raw_table_tasks_sf.load_temp_staging_raw_table.fill( namespace='load_temp_staging_raw_table', date='bootstrap.date', sfdb_params='sfdb_params', feed_name='bootstrap.feed_name', secrets_path='bootstrap.secrets_path', kwargs='kwargs', key_dir='bootstrap.s3_archive_path', temp_staging_raw_table='bootstrap.' 'temp_staging_raw_table')))