""" Peloton Monthly Ingestion Workflow. Ingest Peloton monthly usage files delivered per country to the sme-ca-prod-partners S3 bucket. """ from garcon.param import StaticParam from garcon_contrib.dynamo_feed_status import garcon_feed_status from snowflake_connector.etl_connector import SQLLoader from feed_ingestion.flows import base from feed_ingestion.flows.peloton import config from feed_ingestion.flows.peloton import tasks from feed_ingestion.flows.peloton.stage_loader import PelotonSL from feed_ingestion.tasks import date_tasks from feed_ingestion.tasks import overall_status_tasks class Flow(base.FlowBase, base.FlowConfigMixin): """Class representing the Peloton monthly workflow.""" def __init__(self): """Initialize flow object.""" super(Flow, self).__init__(config.feed_name, version='1.0') def decider(self, schedule): """Activity decider. 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_activity, requires=[get_first_day_of_month]) grab_drop_files = schedule( 'grab_drop_files', self.grab_drop_files, requires=[bootstrap]) # If the file has not landed yet, stop here. if grab_drop_files.result.get('grab_drop_files.stop'): return # Status updated to DOWNLOADED if files are present set_status_to_downloaded = schedule( 'set_status_to_downloaded', self.set_status_to_downloaded, requires=[grab_drop_files]) load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[set_status_to_downloaded]) schedule( 'set_status_to_ingested', self.set_status_to_ingested, requires=[load_staging_raw_table]) @property def get_first_day_of_month(self): """Get the first day of the month of context_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_activity(self): """First activity for the workflow.""" return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap', date='get_first_day_of_month.date', reload='reload'))) @property def grab_drop_files(self): """Copy Peloton files from the SME drop bucket to the archive.""" return self.create( name='grab_drop_files', tasks=base.SyncRunner( tasks.grab_drop_files.fill( namespace='grab_drop_files', feed_name='bootstrap.feed_name', date='bootstrap.date', source_files='bootstrap.source_files'))) @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)))) load_staging_raw_table = PelotonSL.load_activity( secrets_path=config.secrets_path, sql_loader=SQLLoader(__file__), requirements=dict( date='bootstrap.date', feed_name='bootstrap.feed_name', staging_raw_table_name='bootstrap.staging_raw_table', s3_dir_path='bootstrap.s3_dir_path', source_files_dict='grab_drop_files.source_files_dict', sfdb_params='sfdb_params')) @property def set_status_to_ingested(self): """Set the 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))))