"""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 SoundCloudStatsSFExecutor(SnowflakeSQLExecutor): """Helper class to abstract Snowflake operations.""" @property def stats_name(self): """Name of the stats. Should match dir name of these stats, stats_name in config.py of a statistics. Returns: str: Stats name, e.g. 'soundcloud_statistics'. """ return config.FLOW_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['soundcloud_artist_to_track'] @property def soundcloud_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['soundcloud_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_all_links_table', params) def create_soundcloud_statistics(self): """Create soundcloud statistics historical table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.soundcloud_statistics_table) return self.fetchone_query( sql_loader, 'create_soundcloud_statistics', params) def create_staging_raw_soundcloud_statistics(self): """Create staging raw soundcloud 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_soundcloud_statistics', params) def select_artists_links(self): """Select artists links from soundcloud_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.soundcloud_statistics_table) result = self.fetchall_query( sql_loader, 'select_artists_stats', params) return result def update_artist_stats(self, stats: dict, artist): """Load main statistics table.""" stats['the_latest_release'] = datetime.datetime.strptime( '2000-01-01', '%Y-%m-%d') \ 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.soundcloud_statistics_table, artist=artist, soundcloud_link=stats.get('soundcloud_link'), current_followers_count=stats.get('current_followers_count'), last_week_followers_count=stats.get('last_week_followers_count'), weekly_change_in_followers_count=stats.get( 'weekly_change_in_followers_count'), current_weekly_change_in_followers_count=stats.get( 'current_weekly_change_in_followers_count'), daily_change_in_followers_count=stats.get( 'daily_change_in_followers_count'), percentage_change_in_followers_count=stats.get( 'percentage_change_in_followers_count'), acceleration_followers_count=stats.get( 'acceleration_followers_count'), current_followings_count=stats.get('current_followings_count'), last_week_followings_count=stats.get('last_week_followings_count'), weekly_change_in_followings_count=stats.get( 'weekly_change_in_followings_count'), percentage_change_in_followings_count=stats.get( 'percentage_change_in_followings_count'), current_tracks_count=stats.get('current_tracks_count'), last_week_tracks_count=stats.get('last_week_tracks_count'), weekly_change_in_tracks_count=stats.get( 'weekly_change_in_tracks_count'), percentage_change_in_tracks_count=stats.get( 'percentage_change_in_tracks_count'), current_plays_count=stats.get('current_plays_count'), last_week_plays_count=stats.get('last_week_plays_count'), weekly_change_in_plays_count=stats.get( 'weekly_change_in_plays_count'), current_weekly_change_in_plays_count=stats.get( 'current_weekly_change_in_plays_count'), daily_change_in_plays_count=stats.get( 'daily_change_in_plays_count'), percentage_change_in_plays_count=stats.get( 'percentage_change_in_plays_count'), acceleration_plays_count=stats.get('acceleration_plays_count'), current_likes_count=stats.get('current_likes_count'), last_week_likes_count=stats.get('last_week_likes_count'), weekly_change_in_likes_count=stats.get( 'weekly_change_in_likes_count'), current_weekly_change_in_likes_count=stats.get( 'current_weekly_change_in_likes_count'), daily_change_in_likes_count=stats.get( 'daily_change_in_likes_count'), percentage_change_in_likes_count=stats.get( 'percentage_change_in_likes_count'), acceleration_likes_count=stats.get('acceleration_likes_count'), current_reposts_count=stats.get('current_reposts_count'), last_week_reposts_count=stats.get('last_week_reposts_count'), weekly_change_in_reposts_count=stats.get( 'weekly_change_in_reposts_count'), percentage_change_in_reposts_count=stats.get( 'percentage_change_in_reposts_count'), plays_engagement=stats.get('plays_engagement'), the_latest_release=stats.get( 'the_latest_release').strftime('%d-%m-%Y'), 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, '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.soundcloud_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 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]) == 6 else [0] * 6 except IndexError: weeklies = [0] * 6 return dict(zip([ 'followers_count', 'followings_count', 'tracks_count', 'plays_count', 'likes_count', 'reposts_count'], list(weeklies))) def update_stating_raw_soundcloud_stats(self): """Load staging raw table.""" params = dict( db=self.sf_config['db'], schema=self.sf_config['schema'], table_name=self.soundcloud_statistics_table, staging_raw_table_name=self.staging_raw_table ) result = self.fetchall_query( sql_loader, 'update_stating_raw_soundcloud_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 insert_artist_stats( self, artist, stats, table_name=config.snowflake_table_names['soundcloud_statistics']): """First soundcloud statistics insertion.""" stats['the_latest_release'] = datetime.datetime.strptime( '2000-01-01', '%Y-%m-%d') \ 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, soundcloud_link=stats.get('soundcloud_link'), current_followers_count=stats.get('current_followers_count'), last_week_followers_count=stats.get('last_week_followers_count'), weekly_change_in_followers_count=stats.get( 'weekly_change_in_followers_count'), current_weekly_change_in_followers_count=stats.get( 'current_weekly_change_in_followers_count'), daily_change_in_followers_count=stats.get( 'daily_change_in_followers_count'), percentage_change_in_followers_count=stats.get( 'percentage_change_in_followers_count'), acceleration_followers_count=stats.get( 'acceleration_followers_count'), current_followings_count=stats.get('current_followings_count'), last_week_followings_count=stats.get('last_week_followings_count'), weekly_change_in_followings_count=stats.get( 'weekly_change_in_followings_count'), percentage_change_in_followings_count=stats.get( 'percentage_change_in_followings_count'), current_tracks_count=stats.get('current_tracks_count'), last_week_tracks_count=stats.get('last_week_tracks_count'), weekly_change_in_tracks_count=stats.get( 'weekly_change_in_tracks_count'), percentage_change_in_tracks_count=stats.get( 'percentage_change_in_tracks_count'), current_plays_count=stats.get('current_plays_count'), last_week_plays_count=stats.get('last_week_plays_count'), weekly_change_in_plays_count=stats.get( 'weekly_change_in_plays_count'), current_weekly_change_in_plays_count=stats.get( 'current_weekly_change_in_plays_count'), daily_change_in_plays_count=stats.get( 'daily_change_in_plays_count'), percentage_change_in_plays_count=stats.get( 'percentage_change_in_plays_count'), acceleration_plays_count=stats.get('acceleration_plays_count'), current_likes_count=stats.get('current_likes_count'), last_week_likes_count=stats.get('last_week_likes_count'), weekly_change_in_likes_count=stats.get( 'weekly_change_in_likes_count'), current_weekly_change_in_likes_count=stats.get( 'current_weekly_change_in_likes_count'), daily_change_in_likes_count=stats.get( 'daily_change_in_likes_count'), percentage_change_in_likes_count=stats.get( 'percentage_change_in_likes_count'), acceleration_likes_count=stats.get('acceleration_likes_count'), current_reposts_count=stats.get('current_reposts_count'), last_week_reposts_count=stats.get('last_week_reposts_count'), weekly_change_in_reposts_count=stats.get( 'weekly_change_in_reposts_count'), percentage_change_in_reposts_count=stats.get( 'percentage_change_in_reposts_count'), plays_engagement=stats.get('plays_engagement'), the_latest_release=stats.get( 'the_latest_release').strftime('%d-%m-%Y'), 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