"""Snowflake connector class for MRC 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 SpotifyStatsSFExecutor(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. 'spotify_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['spotify_artist_to_track'] @property def spotify_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['spotify_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_artist_link_table(self): """Create artists links table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.artist_to_track_table) return self.fetchone_query( sql_loader, 'create_spotify_artist_to_track', params) def create_spotify_statistics(self): """Create spotify statistics historical table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.spotify_statistics_table) return self.fetchone_query( sql_loader, 'create_spotify_statistics', params) def create_staging_raw_spotify_statistics(self): """Create staging raw spotify 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_spotify_statistics', params) def load_artist_link_table( self, aws=None, key_dir=None): """Load artists links table. Args: aws (dict): AWS credentials to fill a template of COPY SQL statement. key_dir (str): A S3 path to load files from. from_config (bool): load links from config. """ params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.artist_to_track_table, s3_path=key_dir, aws_key_id=aws['access_key'], aws_secret_key=aws['access_secret']) self.execute_query(sql_loader, 'load_spotify_artists_to_track', params) def select_artists_links(self): """Select artists links from spotify_artist_to_track.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.artist_to_track_table) result = self.fetchall_query(sql_loader, 'select_artists', params) return result 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.spotify_statistics_table) result = self.fetchall_query( sql_loader, 'select_artists_stats', params) return result def delete_from_staging_raw(self): """Select artists links from instagram_artist_to_track.""" 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 update_artist_stats(self, stats: dict): """Load main statistics table.""" stats['the_latest_release'] = '2000-01-01' if stats.get( 'the_latest_release') is None else stats.get( 'the_latest_release') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.spotify_statistics_table, monthly_listeners=stats.get('current_monthly_listeners'), last_week_monthly_listeners=stats.get( 'last_week_monthly_listeners'), weekly_change_in_monthly_listeners=stats.get( 'weekly_change_in_monthly_listeners'), current_weekly_change_in_monthly_listeners=stats.get( 'current_weekly_change_in_monthly_listeners'), daily_change_in_monthly_listeners=stats.get( 'daily_change_in_monthly_listeners'), percentage_change_in_monthly_listeners=stats.get( 'percentage_change_in_monthly_listeners'), acceleration_monthly_listeners=stats.get( 'acceleration_monthly_listeners'), current_followers=stats.get('current_followers'), last_week_followers=stats.get('last_week_followers'), current_weekly_change_in_followers=stats.get( 'current_weekly_change_in_followers'), weekly_change_in_followers=stats.get('weekly_change_in_followers'), daily_change_in_followers=stats.get('daily_change_in_followers'), percentage_change_in_followers=stats.get( 'percentage_change_in_followers'), acceleration_followers=stats.get('acceleration_followers'), spotify_popularity=stats.get('current_popularity'), last_week_spotify_popularity=stats.get('last_week_popularity'), editorial_playlists_adds=stats.get( 'current_editorial_playlists_adds'), editorial_playlists_adds_count=stats.get( 'current_editorial_playlists_adds_count'), last_week_editorial_playlists_adds=stats.get( 'last_week_editorial_playlists_adds'), last_week_editorial_playlists_adds_count=stats.get( 'last_week_editorial_playlists_adds_count'), notable_playlists_adds=stats.get('current_notable_playlists_adds'), notable_playlists_adds_count=stats.get( 'current_notable_playlists_adds_count'), last_week_notable_playlists_adds=stats.get( 'last_week_notable_playlists_adds'), last_week_notable_playlists_adds_count=stats.get( 'last_week_notable_playlists_adds_count'), 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.spotify_statistics_table, last_processing_date=last_processing_date.strftime('%d-%m-%Y'), artist=artist ) result = self.fetchall_query( sql_loader, 'update_artists_stats_last_processing_date', 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]) == 2 else [0] * 2 except IndexError: weeklies = [0, 0] return dict(zip(['monthly_listeners', 'followers'], list(weeklies))) def update_stating_raw_spotify_stats(self): """Load staging raw table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.spotify_statistics_table, staging_raw_table_name=self.staging_raw_table ) result = self.fetchall_query( sql_loader, 'update_stating_raw_spotify_stats', params) return result def insert_artist_stats( self, artist, stats, table_name=config.snowflake_table_names['spotify_statistics']): """First Spotify statistics insertion.""" stats['the_latest_release'] = \ '2000-01-01' if stats.get( 'the_latest_release') is None else stats.get( 'the_latest_release') params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=table_name, artist=artist, artist_id=stats.get('id'), monthly_listeners=stats.get('current_monthly_listeners'), last_week_monthly_listeners=stats.get( 'last_week_monthly_listeners'), weekly_change_in_monthly_listeners=stats.get( 'weekly_change_in_monthly_listeners'), current_weekly_change_in_monthly_listeners=stats.get( 'current_weekly_change_in_monthly_listeners'), daily_change_in_monthly_listeners=stats.get( 'daily_change_in_monthly_listeners'), percentage_change_in_monthly_listeners=stats.get( 'percentage_change_in_monthly_listeners'), acceleration_monthly_listeners=stats.get( 'acceleration_monthly_listeners'), current_followers=stats.get('current_followers'), last_week_followers=stats.get('last_week_followers'), weekly_change_in_followers=stats.get( 'weekly_change_in_followers'), current_weekly_change_in_followers=stats.get( 'current_weekly_change_in_followers'), daily_change_in_followers=stats.get( 'daily_change_in_followers'), percentage_change_in_followers=stats.get( 'percentage_change_in_followers'), acceleration_followers=stats.get('acceleration_followers'), spotify_popularity=stats.get('current_popularity'), last_week_spotify_popularity=stats.get('last_week_popularity'), editorial_playlists_adds=stats.get( 'current_editorial_playlists_adds'), editorial_playlists_adds_count=stats.get( 'current_editorial_playlists_adds_count'), last_week_editorial_playlists_adds=stats.get( 'last_week_editorial_playlists_adds'), last_week_editorial_playlists_adds_count=stats.get( 'last_week_editorial_playlists_adds_count'), notable_playlists_adds=stats.get('current_notable_playlists_adds'), notable_playlists_adds_count=stats.get( 'current_notable_playlists_adds_count'), last_week_notable_playlists_adds=stats.get( 'last_week_notable_playlists_adds'), last_week_notable_playlists_adds_count=stats.get( 'last_week_notable_playlists_adds_count'), 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