"""YouTube Bulk Reports Workflow.""" import datetime from garcon import param 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.flows import base from feed_ingestion.flows.youtube_bulk_reports import config, tasks from feed_ingestion.flows.youtube_bulk_reports.stage_loader import \ YouTubeBulkReportsSL from feed_ingestion.tasks import overall_status_tasks, youtube_tasks sql_loader = SQLLoader(__file__) class Flow(base.FlowBase, base.FlowConfigMixin): """Class representing the workflow.""" def __init__(self): """Initialize flow object.""" super(Flow, self).__init__(config.feed_name, config.feed_version) self.timeout = 60 * 60 * 10 # 10 hours def decider(self, schedule): """Activity decider. Args: schedule (callable): The scheduler method. """ bootstrap = schedule('bootstrap', self.bootstrap) if bootstrap.result.get('bootstrap.stop') is True: return licensor = bootstrap.result.get('bootstrap.licensor') if licensor in ['theorchard']: grab_reports_files = schedule( 'grab_reports_files', self.grab_reports_files, requires=[bootstrap]) next_task_requires = grab_reports_files if grab_reports_files.result.get('grab_reports_files.stop'): return else: assert licensor == 'sme' sme_copy_from_athena_to_s3 = schedule( 'sme_copy_from_athena_to_s3', self.sme_copy_from_athena_to_s3, requires=[bootstrap]) next_task_requires = sme_copy_from_athena_to_s3 if sme_copy_from_athena_to_s3.result.get( 'sme_copy_from_athena_to_s3.stop'): return if bootstrap.result.get('bootstrap.only_download') == 'True': return load_staging_raw_table = schedule( 'load_staging_raw_table', self.load_staging_raw_table, requires=[next_task_requires]) if load_staging_raw_table.result.get('stop'): return schedule( 'set_status_to_ingested', self.set_status_to_ingested, requires=[load_staging_raw_table]) return True 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_name = initial_context['report_name'] flow_name = '_'.join([self.name, report_name]) licensor = initial_context['licensor'] if 'context_date' not in initial_context: date = datetime.datetime.today().strftime('%Y-%m-%d') else: date = initial_context['context_date'] return '{flow_name}-{licensor}-{date}'.format( flow_name=flow_name, licensor=licensor, date=date) def contextified_feed_name(self, context): """Get feed_name in context. Args: context (dict): The context of the flow. Returns: str: Contextified feed name. """ assert 'report_name' in context, 'There is no report_name in context' assert 'licensor' in context, 'There is no licensor in context' return '_'.join([self.feed_name, context['licensor'], context['report_name']]) @property def bootstrap(self): """Bootstrap initial configuration.""" return self.create( name='bootstrap', tasks=base.SyncRunner( tasks.bootstrap.fill( namespace='bootstrap', date='context_date', licensor='licensor', report_name='report_name', selected_owner='selected_owner', reload='reload'))) @property def grab_reports_files(self): """Download report files into archive location.""" return self.create( name='grab_reports_files', schedule_to_start=48000, tasks=base.SyncRunner( youtube_tasks.grab_reports_files.fill( namespace='grab_reports_files', report_name='bootstrap.report_name', report_status_name='bootstrap.report_status_name', date='bootstrap.date', archive_path='bootstrap.s3_dir_path', credentials_path='bootstrap.credentials_path', api_service_name=StaticParam( config.youtube_reporting_api_service_name), api_version=StaticParam( config.youtube_reporting_api_version), jobs_meta_path=StaticParam(config.jobs_meta_path), cms_dict=StaticParam(config.orchard_content_owners_map), gz=param.StaticParam(True), selected_owner='bootstrap.selected_owner'), overall_status_tasks.set_overall_status.fill( feed_name='bootstrap.feed_name', status=StaticParam(garcon_feed_status.STATUS_DOWNLOADED), date='bootstrap.date') )) @property def sme_copy_from_athena_to_s3(self): """Bootstrap initial configuration.""" return self.create( name='sme_copy_from_athena_to_s3', tasks=base.SyncRunner( youtube_tasks.sme_copy_from_athena_to_s3.fill( namespace='sme_copy_from_athena_to_s3', date='bootstrap.date', report_status_name='bootstrap.report_status_name', sme_athena_database=param.StaticParam( config.sme_athena_database), sme_athena_temp_database=param.StaticParam( config.sme_athena_temp_database), sme_athena_source_table='bootstrap.youtube_report_id', destination_s3_bucket='bootstrap.s3_bucket', destination_s3_path='bootstrap.archive_path', athena_workgroup=param.StaticParam( config.athena_workgroup), aws_region=param.StaticParam(config.athena_aws_region)), )) load_staging_raw_table = YouTubeBulkReportsSL.load_activity( feed_name=config.feed_name, secrets_path=config.secrets_path, sql_loader=sql_loader, requirements=dict( date='bootstrap.date', report_name='bootstrap.report_name', staging_raw_table_name='bootstrap.staging_raw_table', source_files_dict='source_files.source_files_dict', report_status_name='bootstrap.report_status_name', s3_bucket='bootstrap.s3_bucket', s3_path='bootstrap.archive_path', file_pattern='bootstrap.file_pattern', licensor='bootstrap.licensor', selected_owner='bootstrap.selected_owner', sfdb_params='sfdb_params')) @property def set_status_to_ingested(self): """Set overall feed status to INGESTED.""" return self.create( name='set_status_to_ingested', schedule_to_start=48000, tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_ingested', feed_name='bootstrap.report_status_name', date='bootstrap.date', status=param.StaticParam( garcon_feed_status.STATUS_INGESTED)))) class GroupFlow(Flow): """Class representing the workflow that ingests a group of reports.""" 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. """ licensor = initial_context['licensor'] if 'context_date' not in initial_context: date = datetime.datetime.today().strftime('%Y-%m-%d') else: date = initial_context['context_date'] return '{flow_name}-{licensor}-{date}'.format( flow_name=self.name, licensor=licensor, date=date) @property def bootstrap(self): """Bootstrap initial configuration. The implementation should return 'reports' with list of dict report-dependant bootstrap context data. """ raise NotImplementedError() def report_generator(self, context): """Generate parameters for grab_drop_files activity. Args: context (dict): The current context. Yields: dict: Dictionary with report_name and feed name, which will be used for writing statuses in DynamoDB. """ for report in context['bootstrap.reports']: yield report @property def grab_reports_files(self): """Download report files into archive location.""" return self.create( name='grab_reports_files', schedule_to_start=48000, generators=[self.report_generator], tasks=base.SyncRunner( youtube_tasks.grab_reports_files.fill( namespace='grab_reports_files', report_name='report_name', report_status_name='report_status_name', date='bootstrap.date', archive_path='s3_dir_path', credentials_path='bootstrap.credentials_path', api_service_name=StaticParam( config.youtube_reporting_api_service_name), api_version=StaticParam( config.youtube_reporting_api_version), jobs_meta_path=StaticParam(config.jobs_meta_path), cms_dict='bootstrap.cms_dict', gz=param.StaticParam(True)))) @property def sme_copy_from_athena_to_s3(self): """Bootstrap initial configuration.""" return self.create( name='sme_copy_from_athena_to_s3', generators=[self.report_generator], tasks=base.SyncRunner( youtube_tasks.sme_copy_from_athena_to_s3.fill( namespace='sme_copy_from_athena_to_s3', date='bootstrap.date', report_status_name='report_status_name', sme_athena_database=param.StaticParam( config.sme_athena_database), sme_athena_temp_database=param.StaticParam( config.sme_athena_temp_database), sme_athena_source_table='youtube_report_id', destination_s3_bucket='s3_bucket', destination_s3_path='archive_path', athena_workgroup=param.StaticParam( config.athena_workgroup), aws_region=param.StaticParam(config.athena_aws_region), ))) load_staging_raw_table = YouTubeBulkReportsSL.load_activity( feed_name=config.feed_name, secrets_path=config.secrets_path, sql_loader=sql_loader, generators=[report_generator], requirements=dict( date='bootstrap.date', report_name='report_name', staging_raw_table_name='staging_raw_table', report_status_name='report_status_name', s3_bucket='s3_bucket', s3_path='archive_path', licensor='bootstrap.licensor')) @property def set_status_to_ingested(self): """Set overall feed status to INGESTED.""" return self.create( name='set_status_to_ingested', schedule_to_start=48000, generators=[self.report_generator], tasks=base.SyncRunner( overall_status_tasks.set_overall_status.fill( namespace='set_status_to_ingested', feed_name='report_status_name', date='bootstrap.date', status=param.StaticParam( garcon_feed_status.STATUS_INGESTED))))