"""GfK Physical Ingestion Workflow.""" from snowflake_connector.etl_connector import SQLLoader from feed_ingestion.flows import base from feed_ingestion.flows.gfk_physical import config, tasks from feed_ingestion.flows.gfk_physical.snowflake_executor import ( GfKPhysicalSFExecutor, ) from feed_ingestion.flows.gfk_physical.stage_loader import GfKPhysicalSL sql_loader = SQLLoader(__file__) 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_sftp = schedule( 'fetch_from_sftp', self.fetch_from_sftp, requires=[bootstrap] ) if fetch_from_sftp.result.get('fetch_from_sftp.stop'): return load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[fetch_from_sftp], ) 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', ) ), ) @property def fetch_from_sftp(self): """Download ZIP from SFTP, extract CSV, upload to S3.""" return self.create( name='fetch_from_sftp', tasks=base.SyncRunner( tasks.fetch_from_sftp.fill( namespace='fetch_from_sftp', feed_name='bootstrap.feed_name', date='bootstrap.date', filename='bootstrap.filename', archive_path='bootstrap.archive_path', processed_path='bootstrap.processed_path', ) ), ) load_staging_raw_table = GfKPhysicalSL.load_activity( secrets_path=config.secrets_path, sql_loader=sql_loader, executor_class=GfKPhysicalSFExecutor, requirements=dict( date='bootstrap.date', feed_name='bootstrap.feed_name', staging_raw_table_name='bootstrap.staging_raw_table', s3_dir_path='bootstrap.s3_dir_path', ), )