"""Snowflake connector for the YouTube Monthly workflow.""" from datetime import datetime from snowflake_connector.etl_connector import SQLLoader from feed_ingestion.common.staging_raw_sf.base_executor \ import (SnowflakeSQLExecutorSR) sql_loader = SQLLoader(__file__) class DelphiCrmSnowflakeExecutor(SnowflakeSQLExecutorSR): """Snowflake executor for youtube_monthly.""" def clean_staging_raw_table( self, staging_raw_table, date): """Delete rows from previous unsuccessful workflow run. Args: staging_raw_table (str): A table name in Snowflake. download_date (str): Date of the data being process (YYYY-MM-DD). mcn_account (str): youtube MCN account. """ self.execute_query( sql_loader, query_name='delete_from_staging_raw', params=dict( db=self.sf_config['db'], schema=self.sf_config['schema'], staging_raw_table=staging_raw_table, date=date, ) ) def create_temp_staging_raw_table(self, report_name, temp_staging_raw_table): """Delete rows from previous unsuccessful workflow run. Args: report_name (str): Report name from config.reports. temp_staging_raw_table (str): A table name in Snowflake. """ sql_filename = f'create_temp_staging_raw_{report_name}' self.execute_query( sql_loader, query_name=sql_filename, params=dict( db=self.sf_config['db'], schema=self.sf_config['schema'], temp_staging_raw_table=temp_staging_raw_table, ) ) def load_temp_staging_raw_table(self, temp_staging_raw_table, file_name, s3_dir_path): """Load the temp_staging_raw data to the feed's staging_raw table. Args: temp_staging_raw_table (str): A table name in Snowflake. file_name (list): a list of filenames to load data from. lines_to_skip (int): number of first n lines to skip from CSV file """ sql_template = 'load_temp_staging_raw' aws_params = self.get_aws_params() self.execute_query( sql_loader, query_name=sql_template, params=dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=temp_staging_raw_table, file_name=file_name, s3_dir_path=s3_dir_path, **aws_params ) ) def load_staging_raw_table( self, staging_raw_table, temp_staging_raw_table, date, report_name, file_name): """Load the staging_raw data from temp_staging_raw table. Args: staging_raw_table (str): the temp table name in Snowflake. temp_staging_raw_table (str): the staging raw table name. date (str): Date in format YYYY-MM-DD. report_name (str): Report name from config.reports. mcn_account (str): Youtube MCN account code. """ ingestion_time = datetime.now() sql_template = 'load_staging_raw' self.execute_query( sql_loader, query_name=sql_template, params=dict( db=self.sf_config['db'], schema=self.sf_config['schema'], staging_raw_table=staging_raw_table, temp_staging_raw_table=temp_staging_raw_table, date=date, report_name=report_name, ingestion_time=ingestion_time, file_name=file_name, licensor='sme', ) )