"""Ingestion 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.delphi_crm import config from feed_ingestion.flows.delphi_crm import tasks from feed_ingestion.tasks import feed_status_tasks, 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 = 1 * 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. """ bootstrap = schedule( 'bootstrap', self.bootstrap) if bootstrap.result.get('bootstrap.stop'): return check_files_on_s3 = schedule( 'check_files_on_s3', self.check_files_on_s3, requires=[bootstrap]) if check_files_on_s3.result.get('check_files_on_s3.stop'): return load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[check_files_on_s3]) 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 check_files_on_s3(self): """Check if there are some new files in s3_download_path.""" return self.create( name='check_files_on_s3', tasks=base.SyncRunner( feed_status_tasks.check_files_on_s3.fill( namespace='check_files_on_s3', feed_name='bootstrap.feed_name', date='bootstrap.date', file_pattern='bootstrap.file_pattern', s3_download_path='bootstrap.s3_dir_path'))) @property def load_staging_raw_table(self): """Create temp staging raw table.""" return self.create( name='load_staging_raw_table_reports', tasks=base.SyncRunner( tasks.load_staging_raw_table_reports.fill( namespace='load_staging_raw_table_reports', date='bootstrap.date', feed_name='bootstrap.feed_name', report_name='bootstrap.report_name', staging_raw_table_name='bootstrap.staging_raw_table', s3_dir_path='bootstrap.s3_dir_path', file_name='bootstrap.source_file_name'))) @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))))