"""Dimension Entity. ==================== Module for creating dimension entity objects. Primary purpose will be creating s3 configs and hydrating sql for ETL execution. """ from dim_refresh_etl.conf import getconf from dim_refresh_etl.conf.config import SF_CONFIG STAGING_BUCKET_DEV = 'dev-etl-data' STAGING_BUCKET_SHADOW = 'shadow-etl-data' STAGING_BUCKET_PROD = 'prod-staging-dim-imports' class SimpleDimension(object): """Dimension Entity Object. Super dumb entity object that is created via a yaml config file. May be useful down the road if we want to make hydrating the query list different for different dimensions or add Dimension specific tasks Is probably over engineering for now... """ CONFIG_PACKAGE = 'dim_refresh_etl.conf.dim' DEFAULT_INSERT_COUNT_SQL = ( 'SELECT count(*) ' 'FROM {db}.{schema}.{dim_table} ' "WHERE date_created = '{current_timestamp}'") DEFAULT_UPDATE_COUNT_SQL = ( 'SELECT count(*) ' 'FROM {db}.{schema}.{dim_table} ' "WHERE last_updated = '{current_timestamp}' " 'AND (date_created <> last_updated OR date_created IS NULL)') REQUIRED_ATTR = ( 'source_db', 'dim_table', 'export_sql', 'update_sql', 'insert_sql', 'delete_sql', 'insert_count_sql', 'update_count_sql') def __init__(self, dimension_config, test_mode=False): """Create a Dimension entity. Args: dimension_config (str): name of the dimension config being init'd Should be a file in CONFIG_PATH//yml """ dim_conf = getconf( config_name=dimension_config, package=self.CONFIG_PACKAGE) for k, v in dim_conf.items(): setattr(self, k, v) if 'insert_count_sql' not in dim_conf: self.insert_count_sql = self.DEFAULT_INSERT_COUNT_SQL if 'update_sql' not in dim_conf: self.update_sql = None if 'update_new_rows_sql' not in dim_conf: self.update_new_rows_sql = None if 'delete_sql' not in dim_conf: self.delete_sql = None if 'related_sql' not in dim_conf: self.related_sql = None if 'update_count_sql' not in dim_conf: self.update_count_sql = self.DEFAULT_UPDATE_COUNT_SQL for attr in self.REQUIRED_ATTR: assert hasattr(self, attr), \ "Dimension config must specify '{}'".format(attr) self.staging_table = staging_table(self.dim_table) # allow easy e2e testing if test_mode: self.dim_table = '{}_test'.format(self.dim_table) def hydrate_query(self, sql, current_timestamp=None): """Hydrate a list of sql for a dimension entity. Simple hydration method to populate keywords in a list of sql for a dimension entity Args: current_timestamp (str): current timestamp in '%Y-%m-%d %H:%M:%S' query_list (list): list of queries to hydrate Returns: list: list of sql with keywords hydrated """ hydrated_sql = sql.format( dim_table=self.dim_table, staging_table=self.staging_table, db=SF_CONFIG['db'], schema=SF_CONFIG['schema'], current_timestamp=current_timestamp) return hydrated_sql def hydrate_query_list(self, current_timestamp, query_list=None): """Hydrate a list of sql for a dimension entity. Simple hydration method to populate keywords in a list of sql for a dimension entity Args: current_timestamp (str): current timestamp in '%Y-%m-%d %H:%M:%S' query_list (list): list of queries to hydrate Returns: list: list of sql with keywords hydrated """ hydrated_sql = [] # no queries in query_list (valid for related_sql) if not query_list: return hydrated_sql for sql in query_list: hydrated_sql.append( self.hydrate_query(sql, current_timestamp)) return hydrated_sql def create_entity(dimension_config, test_mode=False): """Help factory method to create Dimension entity. Helper method to create Dimension entity. Will be useful if there are different Dimension classes for different dimensions (not so useful now). Args: dimension_config (str): name of the dimension config being init'd Should be a file in CONFIG_PATH//yml Returns: SimpleDimension: The Dimension entity object corresponding to config """ return SimpleDimension(dimension_config, test_mode) def s3_staging_bucket(env): """Get s3 bucket for an environment. Args: env (str): env Returns: str: s3 bucket for entity staging files (wo 's3://' prefix) """ if env == 'prod': return STAGING_BUCKET_PROD elif env == 'shadow': return STAGING_BUCKET_SHADOW else: return STAGING_BUCKET_DEV def s3_staging_file(dim_table): """Get s3 staging file name. Args: dim_table (str): name of the dimension table being refreshed Returns: str: name of s3 staging file ex. s3://foo/dim_table.gz -> dim_table.gz """ return '{}.gz'.format(dim_table) def s3_staging_key(dim_table, dt): """Get s3 staging key. Args: dim_table (str): name of the dimension table being refreshed dt (str): 'Y-%m-%d' date s3 staging files is being generated Returns: str: s3 staging key for given dim_table / dt ex. s3://foo/bar/1984/dim_table.gz -> bar/1984/dim_table.gz """ return '{s3_key_path}/{file}'.format( s3_key_path=s3_key_path(dim_table, dt=dt), file=s3_staging_file(dim_table)) def s3_key_path(dim_table, dt): """Get s3 staging key path. Args: dim_table (str): name of the dimension table being refreshed dt (str): 'Y-%m-%d' date s3 staging files is being generated Returns: str: s3 staging path for given dim_table / dt ex. s3://foo/bar/dt=1984-01-01/dim_table.gz -> bar/dt=1984-01-01 """ return 'dim/staging-{dim_table}/dt={dt}'.format( dim_table=dim_table, dt=dt) def s3_staging_path(env, dim_table, dt): """Get s3 staging path. Args: env (str): environment swf is running in (prod, dev, test) dim_table (str): name of the dimension table being refreshed dt (str): 'Y-%m-%d' date s3 staging files is being generated Returns: str: s3 staging path for given dim_table / dt ex. s3://foo/bar/1984/dim_table.gz -> s3://foo/bar/1984/ """ return 's3://{s3_staging_bucket}/{s3_key_path}/'.format( s3_staging_bucket=s3_staging_bucket(env), s3_key_path=s3_key_path(dim_table, dt)) def staging_table(dim_table): """Get dimension staging table name. Args: dim_table (str): name of the dimension table being refreshed Returns: str: returns 'staging_{dim_table}' """ return 'staging_{}'.format(dim_table)