"""Snowflake connector class for YouTube Asset Conflict tasks.""" from snowflake_connector.etl_connector import SQLLoader from feed_ingestion.common.staging_raw_sf.base_executor \ import SnowflakeSQLExecutorSR # Load SQL templates sql_loader = SQLLoader(__file__) class YouTubeAssetConflictSFExecutor(SnowflakeSQLExecutorSR): """Helper class to abstract Snowflake operations.""" def create_staging_raw_temp_table(self, staging_raw_temp_table): """Create temp staging raw table.""" sql_template = sql_loader.load_query('create_staging_raw_temp') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], staging_raw_temp_table=staging_raw_temp_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def load_staging_raw_temp_table( self, staging_raw_temp_table, key_dir): """Load temp staging raw table with YouTube conflict report from S3. Args: staging_raw_temp_table (str): A table name in Snowflake. key_dir (str): A S3 path to load files from. """ aws_params = self.get_aws_params() sql_template = sql_loader.load_query('load_staging_raw_temp') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], staging_raw_temp_table=staging_raw_temp_table, s3_path=key_dir, **aws_params) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def truncate_table(self, table_name): """Truncate by table name. Args: table_name (str): A table name in Snowflake. """ sql_template = sql_loader.load_query( 'truncate_table') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=table_name) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def insert_into_staging_raw_table( self, staging_raw_temp_table, staging_raw_table, file_name, file_size, download_date, content_owner): """Insert into staging_raw_table from staging_raw_temp_table. Args: staging_raw_temp_table (str): table inserting from. staging_raw_table (str): table inserting into. file_name (str): file_name used to createstaging_raw_temp_table. file_size (str): file_size used to createstaging_raw_temp_table. download_date (str): date file was downloaded. content_owner (str): the content owner of the file. """ sql_template = sql_loader.load_query( 'insert_into_staging_raw') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], staging_raw_table=staging_raw_table, staging_raw_temp_table=staging_raw_temp_table, file_name=file_name, file_size=file_size, download_date=download_date, content_owner=content_owner) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def create_territories_temp_table(self, territories_temp_table): """Сreate temp territories table. Args: territories_temp_table (str): A table name in Snowflake. """ sql_template = sql_loader.load_query( 'create_territories_temp_table') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], territories_temp_table=territories_temp_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def fill_territories_temp_table( self, territories_temp_table, territories): """Fill temp territories table with territories. Args: territories_temp_table (str): A table name in Snowflake. territories (list): List of territories from ows-territories. """ sql_template = sql_loader.load_query( 'fill_territories_temp_table') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], territories_temp_table=territories_temp_table) territory_values = [ { 'territory': item['territory_code_a2'], 'standard': item['standard'] } for item in territories ] sql, _ = self.validator.format_identifiers(sql_template, params) self.executemany(sql, params_list=territory_values) def create_fact_conflict_temp_table(self, fact_conflict_temp_table): """Сreate fact_conflict_temp table. Args: fact_conflict_temp_table (str): A table name in Snowflake. """ sql_template = sql_loader.load_query( 'create_fact_conflict_temp_table') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_temp_table=fact_conflict_temp_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def insert_overwrite_into_fact_conflict_temp_table( self, art_relations_db, art_relations_schema, asset_report_schema, registry_schema, fact_conflict_temp_table, yact_table, asset_report_table, registry_table, track_table, releases_table, artist_info_table, territory_standard, asset_type, time_zone, orchard_account): """Insert overwrite into fact_conflict_temp_table. Args: art_relations_db (str): Name of art_relations database in Snowflake. art_relations_schema (str): Schema name used in art_relations database. asset_report_schema (str): Schema name used for asset report table. registry_schema (str): Schema name used for registry. fact_conflict_temp_table (str): Name of fact_conflict_temp table. yact_table (str): Name of youtube_asset_conflict_by_territory_temp table. asset_report_table (str): Name of staging_raw_youtube_asset_report table. registry_table (str): Name of registry table. track_table (str): Name of track table. releases_table (str): Name of releases table. artist_info_table (str): Name of artist info table territory_standard (str): Territory standard that the conflict is in. asset_type (str): YouTube asset type. time_zone (str): System specified time zone. orchard_account (str): YT CMS account name for The Orchard Music. """ sql_template = sql_loader.load_query( 'insert_overwrite_into_fact_conflict_temp_with_yt_conflicts') params = dict( facts_db=self.sf_config['db'], facts_schema=self.sf_config['schema'], art_relations_db=art_relations_db, art_relations_schema=art_relations_schema, asset_report_schema=asset_report_schema, registry_schema=registry_schema, fact_conflict_temp_table=fact_conflict_temp_table, youtube_asset_conflict_by_territory_temp_table=yact_table, staging_raw_youtube_asset_report_table=asset_report_table, registry_table=registry_table, track_table=track_table, releases_table=releases_table, artist_info_table=artist_info_table, territory_standard=territory_standard, asset_type=asset_type, time_zone=time_zone, orchard_account=orchard_account) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def check_size_of_fact_conflict_temp_table(self, fact_conflict_temp_table): """Check if the size of the fact conflict temp table seems correct. Args: fact_conflict_temp_table (str): fact_conflict_temp table. """ sql_template = sql_loader.load_query( 'check_size_of_fact_conflict_temp_table') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_temp_table=fact_conflict_temp_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) return self.fetchone(sql, params=non_identifier_params)[0] def check_number_of_unresolved_conflicts(self, fact_conflict_table): """Check number of unresolved conflicts. Args: fact_conflict_table (str): fact_conflict table. """ sql_template = sql_loader.load_query( 'check_number_of_unresolved_conflicts') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_table=fact_conflict_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) return self.fetchone(sql, params=non_identifier_params)[0] def create_youtube_asset_conflict_by_territory_table(self, yact_table): """Create youtube_asset_conflict_by_territory_temp table. Args: yact_table (str): youtube_asset_conflict_by_territory_table """ sql_template = sql_loader.load_query( 'create_youtube_asset_conflict_by_territory_temp') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], yact_table=yact_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def insert_into_fact_conflict_table( self, fact_conflict_table, fact_conflict_temp_table): """Insert into fact_conflict_table. Args: fact_conflict_table (str): fact_conflict table. fact_conflict_temp_table (str): fact_conflict_temp table. """ sql_template = sql_loader.load_query( 'insert_into_fact_conflict') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_table=fact_conflict_table, fact_conflict_temp_table=fact_conflict_temp_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def update_fact_conflict_yt_recent_daily_average( self, fact_conflict_table, fact_conflict_temp_table): """Update fact_conflict_table yt_recent_daily_average column. Args: fact_conflict_table (str): fact_conflict table. fact_conflict_temp_table (str): fact_conflict_temp table. """ sql_template = sql_loader.load_query( 'update_fact_conflict_yt_recent_daily_average') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_table=fact_conflict_table, fact_conflict_temp_table=fact_conflict_temp_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def update_fact_conflict_views_in_conflict( self, fact_conflict_table, fact_conflict_temp_table): """Update fact_conflict_table views_in_conflict column. Args: fact_conflict_table (str): fact_conflict table. fact_conflict_temp_table (str): fact_conflict_temp table. """ sql_template = sql_loader.load_query( 'update_fact_conflict_views_in_conflict') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_table=fact_conflict_table, fact_conflict_temp_table=fact_conflict_temp_table) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def update_fact_conflict_resolved_datetime( self, fact_conflict_table, fact_conflict_temp_table, time_zone): """Update fact_conflict_table resolved_datetime column. Args: fact_conflict_table (str): fact_conflict table. fact_conflict_temp_table (str): fact_conflict_temp table. time_zone (str): System specified time zone. """ sql_template = sql_loader.load_query( 'update_fact_conflict_resolved_datetime') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_table=fact_conflict_table, fact_conflict_temp_table=fact_conflict_temp_table, time_zone=time_zone) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def reset_es_indexed_for_partially_resolved_conflicts( self, fact_conflict_table, time_zone): """Update fact_conflict_table es_indexed column. Args: fact_conflict_table (str): fact_conflict table. fact_conflict_temp_table (str): fact_conflict_temp table. time_zone (str): System specified time zone. """ sql_template = sql_loader.load_query( 'reset_es_indexed_for_partially_resolved_conflicts') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], fact_conflict_table=fact_conflict_table, time_zone=time_zone) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params) def fill_youtube_asset_conflict_by_territory_table( self, yact_table, staging_raw_table, territories_temp_table): """Fill youtube_asset_conflict_by_territory_temp table. Args: yact_table (str): youtube_asset_conflict_by_territory_table staging_raw_table (str): staging raw table name. territories_temp_table (str): territories_temp table name """ sql_template = sql_loader.load_query( 'load_youtube_asset_conflict_by_territory') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], yact_table=yact_table, staging_raw_table=staging_raw_table, territories_temp_table=territories_temp_table ) sql, non_identifier_params = self.validator.format_identifiers( sql_template, params) self.execute(sql, params=non_identifier_params)