""" NetEase Ingestion Workflow. Ingest data from NetEase through sme-ca-prod-partners s3 bucket. """ 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.netease import config from feed_ingestion.flows.netease import tasks from feed_ingestion.tasks import load_raw_table_tasks_sf from feed_ingestion.tasks import overall_status_tasks from feed_ingestion.tasks import validate_raw_data_tasks_sf class Flow(base.FlowBase, base.FlowConfigMixin, base.FlowLoadRawMixinSF, base.FlowLoadFactMixinSF): """Class representing the workflow.""" def __init__(self): """Initialize flow object.""" super(Flow, self).__init__(config.feed_name, version='1.0') def decider(self, schedule): """Activity decider. Args: schedule (callable): The scheduler method. """ bootstrap = schedule('bootstrap', self.bootstrap_activity) grab_drop_files = schedule( 'grab_drop_files', self.grab_drop_files, requires=[bootstrap] ) # if files unavailable, let's stop here if grab_drop_files.result.get('grab_drop_files.stop'): return set_status_to_downloaded = schedule( 'set_status_to_downloaded', self.set_status_to_downloaded, requires=[grab_drop_files]) create_temp_staging_raw_table = schedule( 'create_temp_staging_raw_table', self.create_temp_staging_raw_table, requires=[set_status_to_downloaded]) load_temp_staging_raw_table = schedule( 'load_temp_staging_raw_table', self.load_temp_staging_raw_table_, requires=[create_temp_staging_raw_table]) load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[load_temp_staging_raw_table]) mark_staging_raw_table_tasks_complete = schedule( 'mark_staging_raw_table_tasks_complete', self.mark_staging_raw_table_tasks_complete, requires=[load_staging_raw_table]) set_status_populated_raw_table = schedule( 'set_status_populated_raw_table', self.set_status_populated_raw_table, requires=[mark_staging_raw_table_tasks_complete]) schedule( 'set_overall_status_ingested', self.set_overall_status_ingested, requires=[set_status_populated_raw_table] ) @property def bootstrap_activity(self): """First activity for the workflow.""" return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap', date='context_date', reload='reload'))) @property def grab_drop_files(self): """Grab files and copy to archive bucket on S3.""" 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', source_key_name='bootstrap.source_key_name', destination_key_name='bootstrap.destination_key_name'))) @property def set_status_to_downloaded(self): """Set overall feed status to DOWNLOADED.""" return self.create( name='set_status_to_downloaded', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_downloaded', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam( garcon_feed_status.STATUS_DOWNLOADED)))) @property def load_temp_staging_raw_table_(self): """Load temp staging raw table.""" return self.create( name='load_temp_staging_raw_tables', tasks=base.SyncRunner( validate_raw_data_tasks_sf.load_temp_staging_raw_table.fill( namespace='load_temp_staging_raw_tables', feed_name='bootstrap.feed_name', sfdb_params='sfdb_params', date='bootstrap.date', key_dir='bootstrap.key_dir', temp_staging_raw_table='bootstrap.temp_staging_raw_table', secrets_path='bootstrap.secrets_path', error_limit=( StaticParam(config.snowflake_error_limit)), kwargs='kwargs'))) @property def mark_staging_raw_table_tasks_complete(self): """Set the complete staging_raw_table_tasks to feed.""" return self.create( name='mark_staging_raw_table_tasks_complete', tasks=base.SyncRunner( (load_raw_table_tasks_sf.mark_staging_raw_table_tasks_complete. fill(namespace='mark_staging_raw_table_tasks_complete', date='bootstrap.date', feed_name='bootstrap.feed_name')))) @property def set_status_populated_raw_table(self): """Set the overall feed status to populated_raw_table.""" return self.create( name='set_status_populated_raw_table', tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_populated_raw_table', feed_name='bootstrap.feed_name', date='bootstrap.date', status=StaticParam( garcon_feed_status.STATUS_POPULATED_RAW_TABLE)))) @property def set_overall_status_ingested(self): """Set overall feed status INGESTED if all reports are completed.""" return self.create( name='set_overall_status_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))))