"""GFK Ingestion Workflow.""" from garcon.param import StaticParam from garcon_contrib.aws import garcon_s3 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.gfk import config, tasks from feed_ingestion.flows.gfk.stage_loader import GFKSL from feed_ingestion.tasks import overall_status_tasks class Flow(base.FlowBase, base.FlowConfigMixin, base.FlowLoadFactMixinSF): """Class representing the workflow.""" def __init__(self): """Initialize flow object.""" super(Flow, self).__init__(feed_name=config.feed_name, version='2.0') def decider(self, schedule, context): """Activity decider. Args: schedule (callable): The scheduler method. """ bootstrap = schedule('bootstrap', self.bootstrap) # Stop flow if feed already ingested if bootstrap.result.get('bootstrap.stop'): return cleanup_s3 = schedule( 'cleanup_s3', self.cleanup_s3, requires=[bootstrap]) fetch_from_drop_location = schedule( 'fetch_from_drop_location', self.fetch_from_drop_location, requires=[cleanup_s3]) # Stop flow if files aren't available 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]) load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[set_status_to_downloaded]) 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_staging_fact_table = schedule( 'load_staging_fact_table', self.load_staging_fact_table, requires=[set_status_to_populated_raw_table]) schedule( 'load_fact_tables', self.load_fact_tables, requires=[load_staging_fact_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 cleanup_s3(self): """Cleanup S3 in order to achieve idempotency.""" return self.create( name='cleanup_s3', tasks=base.SyncRunner( garcon_s3.remove_files_from_path.fill( path='bootstrap.s3_archive_path', return_deleted_files=StaticParam(False)))) @property def fetch_from_drop_location(self): """Move file from FTP to S3 archive bucket.""" return self.create( name='fetch_from_drop_location', tasks=base.SyncRunner( tasks.fetch_from_drop_location.fill( namespace='fetch_from_drop_location', feed_name='bootstrap.feed_name', date='bootstrap.date', s3_archive_path='bootstrap.s3_archive_path', ftp_creds=StaticParam(config.ftp)))) load_staging_raw_table = GFKSL.load_activity( feed_name=config.feed_name, secrets_path=config.secrets_path, sql_loader=SQLLoader(__file__), requirements=dict( date='bootstrap.date', source_files_dict='fetch_from_drop_location.source_files_dict', s3_dir_path='bootstrap.s3_archive_path', staging_raw_table_name=StaticParam( config.snowflake_table_names['staging_raw']))) @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 set_status_to_populated_raw_table(self): """Set overall feed status to POPULATED_RAW_TABLE.""" return self.create( name='set_status_to_populated_raw_table', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_populated_raw_table', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam( garcon_feed_status.STATUS_POPULATED_RAW_TABLE))))