"""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.physical_reporting import config from feed_ingestion.flows.physical_reporting 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 = 3 * 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'] date = initial_context['context_date'] workflow_id = '-'.join([self.name, report, date]) return workflow_id 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'] return '_'.join([self.feed_name, report]) def decider(self, schedule): """Activity decider. Args: schedule (callable): The scheduler method. """ check_feed_status = schedule( 'check_feed_status', self.check_feed_status) if check_feed_status.result.get('check_feed_status.stop'): return bootstrap = schedule( 'bootstrap', self.bootstrap, requires=[check_feed_status]) grab_drop_files = schedule( 'grab_drop_files', self.grab_drop_files, requires=[bootstrap]) if grab_drop_files.result.get('grab_drop_files.stop'): return load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[grab_drop_files]) schedule( 'set_status_to_ingested', self.set_status_to_ingested, requires=[load_staging_raw_table]) @property def check_feed_status(self): """Check and reset feed status if it is needed.""" return self.create( name='check_feed_status', tasks=base.SyncRunner( overall_status_tasks.check_feed_status.fill( namespace='check_feed_status', date='context_date', feed_name=StaticParam(config.feed_name), report='report', reload='reload'))) @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 grab_drop_files(self): """Grab files from the drop location.""" 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', s3_archive_path='bootstrap.s3_archive_path', s3_download_path='bootstrap.s3_download_path', drop_file_name='bootstrap.drop_file_name'))) 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_archive_path', staging_raw_table_name='bootstrap.staging_raw_table_name', report='bootstrap.report_name', source_files_dict=( 'grab_drop_files.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_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))))