"""Snowflake connector class for Shazam tasks.""" import datetime import config from snowflake_connector.etl_connector import SQLLoader from snowflake_connector.etl_connector import SnowflakeSQLExecutor # Load SQL templates sql_loader = SQLLoader(__file__) looker_sql_loader = SQLLoader(__file__, folder='/looker_queries') class YouTubeStatsSFExecutor(SnowflakeSQLExecutor): """Helper class to abstract Snowflake operations.""" @property def stats_name(self): """Name of the stats. Should match dir name of this stats, stats_name in config.py of a statistics. Returns: str: Stats name, e.g. 'youtube_statistics'. """ return config.STATS_NAME @property def artist_to_track_table(self): """Name of the artist_to_track table for the statistics. Returns: str: {stats}_artist_to_track table name. """ return config.snowflake_table_names['youtube_artist_to_track'] @property def youtube_statistics_table(self): """Name of the artist_to_track table for the statistics. Returns: str: {stats}_artist_to_track table name. """ return config.snowflake_table_names['youtube_statistics'] @property def staging_raw_table(self): """Name of the staging raw table for the statistics. Returns: str: staging raw table name. """ return config.snowflake_table_names['staging_raw'] def create_youtube_statistics(self): """Create youtube statistics historical table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.youtube_statistics_table) return self.fetchone_query( sql_loader, 'create_youtube_statistics', params) def create_staging_raw_youtube_statistics(self): """Create staging raw youtube statistics historical table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.staging_raw_table) return self.fetchone_query( sql_loader, 'create_youtube_statistics', params) def select_artists_links(self): """Selecting artists links from youtube_artist_to_track.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.artist_to_track_table) return self.fetchall_query(sql_loader, 'select_artists', params) def select_artists_stats(self): """Load temp staging raw table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.youtube_statistics_table) result = self.fetchall_query( sql_loader, 'select_artists_stats', params) return result def select_recent_stats(self): """Select artists with updated stats in last 3 days .""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.youtube_statistics_table) result = self.fetchall_query(sql_loader, 'select_recent_stats', params) return result def update_artist_stats(self, stats: dict): """Load main statistics table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.youtube_statistics_table, current_subscribers=stats.get('current_subscribers'), last_week_subscribers=stats.get('last_week_subscribers'), weekly_change_in_subscribers=stats.get( 'weekly_change_in_subscribers'), current_weekly_change_in_subscribers=stats.get( 'current_weekly_change_in_subscribers'), daily_change_in_subscribers=stats.get( 'daily_change_in_subscribers'), percentage_change_in_subscribers=stats.get( 'percentage_change_in_subscribers'), acceleration_subscribers=stats.get('acceleration_subscribers'), current_videos=stats.get('videos_count'), current_likes=stats.get('current_likes'), last_week_likes=stats.get('last_week_likes'), weekly_change_in_likes=stats.get('weekly_change_in_likes'), current_weekly_change_in_likes=stats.get( 'current_weekly_change_in_likes'), daily_change_in_likes=stats.get('daily_change_in_likes'), percentage_change_in_likes=stats.get('percentage_change_in_likes'), acceleration_likes=stats.get('acceleration_likes'), current_dislikes=stats.get('current_dislikes'), last_week_dislikes=stats.get('last_week_dislikes'), weekly_change_in_dislikes=stats.get('weekly_change_in_dislikes'), percentage_change_in_dislikes=stats.get( 'percentage_change_in_dislikes'), current_views=stats.get('current_views'), last_week_views=stats.get('last_week_views'), weekly_change_in_views=stats.get('weekly_change_in_views'), current_weekly_change_in_views=stats.get( 'current_weekly_change_in_views'), daily_change_in_views=stats.get('daily_change_in_views'), percentage_change_in_views=stats.get('percentage_change_in_views'), acceleration_views=stats.get('acceleration_views'), current_comments=stats.get('current_comments'), last_week_comments=stats.get('last_week_comments'), weekly_change_in_comments=stats.get('weekly_change_in_comments'), percentage_change_in_comments=stats.get( 'percentage_change_in_comments'), likes_engagement=stats.get('likes_engagement'), comments_engagement=stats.get('comments_engagement'), the_latest_release=stats.get('the_latest_release'), last_processing_date=stats.get( 'last_processing_date').strftime('%d-%m-%Y'), last_week_processing_date=stats.get( 'last_week_processing_date').strftime('%d-%m-%Y'), artist=stats.get('id') ) result = self.fetchall_query( sql_loader, 'update_artists_stats', params) return result def update_artist_stats_last_processing_date( self, artist: str, last_processing_date: datetime): """Update artists statistics table on failure.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.youtube_statistics_table, last_processing_date=last_processing_date, artist=artist ) result = self.fetchall_query( sql_loader, 'update_artists_stats_last_processing_date', params) return result def update_stating_raw_youtube_stats(self): """Load staging raw table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.youtube_statistics_table, staging_raw_table_name=self.staging_raw_table ) result = self.fetchall_query( sql_loader, 'update_stating_raw_youtube_stats', params) return result def delete_from_staging_raw(self): """Delete from staging raw today's rows to prevent duplicates.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], staging_raw_table=self.staging_raw_table, date=datetime.datetime.today().strftime('%Y-%m-%d')) result = self.fetchall_query( sql_loader, 'delete_from_staging_raw', params) return result def select_weekly_change(self, artist, first_day_of_period=0, last_day_of_period=-7): """Select artist weekly change in metrics from staging raw table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.staging_raw_table, artist=artist, first_day_of_period=first_day_of_period, last_day_of_period=last_day_of_period ) result = self.fetchall_query( sql_loader, 'select_weekly_change', params) try: weeklies = result[0] if len(result[0]) == 5 else [0] * 5 except IndexError: weeklies = [0] * 5 return dict(zip(['subscribers', 'likes', 'dislikes', 'views', 'comments'], list(weeklies))) def insert_artist_stats( self, artist, stats, table_name=config.snowflake_table_names['youtube_statistics']): """First YouTube statistics insertion.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=table_name, artist=artist, artist_id=stats.get('id'), current_subscribers=stats.get('current_subscribers'), last_week_subscribers=stats.get('last_week_subscribers'), weekly_change_in_subscribers=stats.get( 'weekly_change_in_subscribers'), current_weekly_change_in_subscribers=stats.get( 'current_weekly_change_in_subscribers'), daily_change_in_subscribers=stats.get( 'daily_change_in_subscribers'), percentage_change_in_subscribers=stats.get( 'percentage_change_in_subscribers'), acceleration_subscribers=stats.get('acceleration_subscribers'), current_videos=stats.get('videos_count'), current_likes=stats.get('current_likes'), last_week_likes=stats.get('last_week_likes'), weekly_change_in_likes=stats.get('weekly_change_in_likes'), current_weekly_change_in_likes=stats.get( 'current_weekly_change_in_likes'), daily_change_in_likes=stats.get('daily_change_in_likes'), percentage_change_in_likes=stats.get('percentage_change_in_likes'), acceleration_likes=stats.get('acceleration_likes'), current_dislikes=stats.get('current_dislikes'), last_week_dislikes=stats.get('last_week_dislikes'), weekly_change_in_dislikes=stats.get('weekly_change_in_dislikes'), percentage_change_in_dislikes=stats.get( 'percentage_change_in_dislikes'), current_views=stats.get('current_views'), last_week_views=stats.get('last_week_views'), weekly_change_in_views=stats.get('weekly_change_in_views'), current_weekly_change_in_views=stats.get( 'current_weekly_change_in_views'), daily_change_in_views=stats.get('daily_change_in_views'), percentage_change_in_views=stats.get('percentage_change_in_views'), acceleration_views=stats.get('acceleration_views'), current_comments=stats.get('current_comments'), last_week_comments=stats.get('last_week_comments'), weekly_change_in_comments=stats.get('weekly_change_in_comments'), percentage_change_in_comments=stats.get( 'percentage_change_in_comments'), likes_engagement=stats.get('likes_engagement'), comments_engagement=stats.get('comments_engagement'), the_latest_release=stats.get('the_latest_release'), last_processing_date=stats.get( 'last_processing_date').strftime('%d-%m-%Y'), last_week_processing_date=stats.get( 'last_week_processing_date').strftime('%d-%m-%Y') ) result = self.fetchall_query( sql_loader, 'insert_artists_stats', params) return result