"""Ingestion Workflow.""" 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.common.staging_raw_sf.snowflake_stage_loader \ import LicensorSL from feed_ingestion.flows import base from feed_ingestion.flows.insta_reels import config from feed_ingestion.flows.insta_reels import tasks from feed_ingestion.tasks import overall_status_tasks class Flow(base.FlowBase, base.FlowConfigMixin): """Class representing the workflow.""" def __init__(self): """Initialize flow object.""" super(Flow, self).__init__(feed_name=config.feed_name, version='1.0') self.timeout = 4 * 60 * 60 def workflow_id(self, initial_context): """Generate workflow id. Args: initial_context (dict): The initial context for the flow. Returns: str: A unique identifier for a workflow being executed. """ report = initial_context['report'] flow_name = '_'.join([self.name, report]) date = initial_context['context_date'] return '{flow_name}-{date}'.format( flow_name=flow_name, date=date) def contextified_feed_name(self, context): """Get feed_name in context. Args: context (dict): The context of the flow. Returns: str: Contextified feed name. """ report = context['report'] assert report in config.reports return '_'.join([self.feed_name, report]) def decider(self, schedule): """Activity decider. Args: schedule (callable): The scheduler method. """ bootstrap = schedule( 'bootstrap', self.bootstrap) if bootstrap.result.get('bootstrap.stop'): return fetch_from_athena = schedule( 'fetch_from_athena', self.fetch_from_athena, requires=[bootstrap]) load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[fetch_from_athena]) schedule( 'set_overall_status_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', report='report', reload='reload', ))) @property def fetch_from_athena(self): """Move file from S3 SME location to S3 archive bucket.""" return self.create( name='fetch_from_athena', tasks=base.SyncRunner( tasks.fetch_from_athena.fill( namespace='fetch_from_athena', feed_name='bootstrap.feed_name', date='bootstrap.date', report='bootstrap.report', destination_s3_bucket='bootstrap.archive_bucket', destination_s3_path='bootstrap.archive_path'))) load_staging_raw_table = LicensorSL.load_activity( feed_name=config.feed_name, sql_loader=SQLLoader(__file__), secrets_path=config.secrets_path, requirements=dict( date='bootstrap.date', feed_name='bootstrap.feed_name', s3_dir_path='bootstrap.s3_dir_path', staging_raw_table_name='bootstrap.staging_raw_table', report='bootstrap.report', source_files_dict=( 'fetch_from_athena.source_files_dict'), sfdb_params='sfdb_params')) @property def set_status_to_ingested(self): """Set overall feed status to STATUS_INGESTED.""" return self.create( name='set_status_ingested', 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_INGESTED))))