"""Sme latam Ingestion Workflow.""" from snowflake_connector.etl_connector import SQLLoader from feed_ingestion.common.staging_raw_sf.snowflake_stage_loader \ import StageLoader from feed_ingestion.flows import base from feed_ingestion.flows.sme_latam import config from feed_ingestion.flows.sme_latam import 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') 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_drop_location = schedule( 'fetch_from_drop_location', self.fetch_from_drop_location, requires=[bootstrap]) # Stop flow if files aren't available if fetch_from_drop_location.result.get( 'fetch_from_drop_location.stop'): return process_drop_files = schedule( 'process_drop_files', self.process_drop_files, requires=[fetch_from_drop_location]) load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[process_drop_files]) 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', reload='reload', snowflake_error_limit='snowflake_error_limit'))) @property def fetch_from_drop_location(self): """Move file from S3 SME location 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', drop_path='bootstrap.drop_path', archive_path='bootstrap.archive_path'))) @property def process_drop_files(self): """Process xlsx file to csv.""" return self.create( name='process_drop_files', tasks=base.SyncRunner( tasks.process_drop_files.fill( namespace='process_drop_files', feed_name='bootstrap.feed_name', date='bootstrap.date', archive_path='bootstrap.archive_path', processed_path='bootstrap.processed_path'))) load_staging_raw_table = StageLoader.load_activity( feed_name=config.feed_name, sql_loader=SQLLoader(__file__), secrets_path=config.secrets_path, requirements=dict( date='bootstrap.date', s3_dir_path='process_drop_files.s3_dir_path', staging_raw_table_name='bootstrap.staging_raw_table', source_files_dict=( 'process_drop_files.source_files_dict'), sfdb_params='sfdb_params' ))