"""Data types classes designed to generate Snowflake SQL. Note: Sometimes in this module we're using {{}} braces in strings to be formatted. The reason is following: we must format them twice (first in sql() method, and later we format the result of sql() in sql_snowflake() method). """ from snowflake_etl.conf import getconf TYPE_MAP = getconf('schema_map') class SQLType: """Represents a simple SQL data type. Instances of this class and it's subclasses know how to produce SQL strings or different flavors, namely Redshift, MySQL and Snowflake. And can be used for conversion from Redshift and MySQL to Snowflake. """ SQL_TEMPLATE = '{type}' def __init__(self, type): """ Initialise SQLType object with value. Args: type (str): Basic data type (e.g. REAL, NUMBER, etc.). """ self.type = type.upper() self.TYPE_MAP = getconf('schema_map') def sql(self): """ Generate source SQL definition for the data type. Returns: str: Source style SQL for the data type. """ return self.SQL_TEMPLATE.format_map(self.__dict__) def sql_snowflake(self, db_type): """ Generate Snowflake SQL definition for the data type. Args: db_type (str): Source database type (e.g., 'redshift'). Returns: str: Snowflake style SQL for the data type. """ if db_type == 'mysql': type_map = self.TYPE_MAP['mysql_to_snowflake'] elif db_type == 'redshift': type_map = self.TYPE_MAP['redshift_to_snowflake'] sf_template = type_map[self.sql()] return sf_template.format_map(self.__dict__) class SQLNumType(SQLType): """Represents a numeric SQL data type.""" SQL_TEMPLATE = '{type}({{precision}},{{scale}})' def __init__(self, type, precision, scale): """ Initialise SQLNumType object with values. Args: type (str): Basic data type (e.g. REAL, NUMBER, etc.). precision (int): Numeric precision. scale (int): Numeric scale. """ super(SQLNumType, self).__init__(type) self.precision = precision self.scale = scale class SQLCharType(SQLType): """Represents a character SQL data type.""" SQL_TEMPLATE = '{type}({{length}})' def __init__(self, type, length): """ Initialise SQLCharType object with values. Args: type (str): Basic data type (e.g. REAL, NUMBER, etc.). length (int): Max number of characters. """ super(SQLCharType, self).__init__(type) self.length = length class SQLColumn: """This encapsulates all state relevant to standard source columns. Note: Contains methods for formatting both source and Snowflake SQL. """ @classmethod def from_json(cls, column_name, data_type, character_maximum_length, numeric_precision, numeric_scale, is_nullable, **kwargs): """Construct source column from source-style schema dict. Notes: this contructor is meant to be used with kwargs unpacked from each column element in the schema dictionary. Args: column_name (str): Column name. data_type (str): Column data type. character_maximum_length (int): Max length of char types. numeric_precision (int): Precision of numeric types. numeric_scale (int): Scale of numeric types. is_nullable (str): "YES" / "NO" of nullability. Return: SQLColumn: SQL column. """ # null checks help discern between and 0 and None # numeric types if numeric_precision is not None and numeric_scale is not None: data_type = SQLNumType( data_type, numeric_precision, numeric_scale) # character types elif character_maximum_length is not None: data_type = SQLCharType(data_type, character_maximum_length) # other types else: data_type = SQLType(data_type) # nullable nullable = is_nullable == 'YES' return cls(column_name, data_type, nullable) def __init__(self, name, type, nullable=True): """ Initialise SQLColumn object with values. Args: name (str): Column name. type (str): Data type. nullable (bool): If value is nullable. """ self.name = name.upper() self.type = type self.nullable = nullable def sql(self): """Source-style SQL for a single column. Return: str: Source SQL. """ return '{name} {type}'.format(name=self.name, type=self.type.sql()) def sql_snowflake(self, db_type): """Snowflake-style SQL for a single column. Args: db_type (str): Source database type (e.g., 'redshift'). Return: str: Snowflake SQL. """ return '{name} {type} {nullable}'.format( name=self.name, type=self.type.sql_snowflake(db_type), nullable='' if self.nullable else 'NOT NULL').strip() class SQLTable: """Represents source table definition. Instances of this class can be useful in conversion between source table schema and other SQL flavors, namely Snowflake. """ @classmethod def from_json(cls, schema, table, columns, **kwargs): """Construct source table from source-style schema dict. Notes: this constructor is meant to be used with kwargs unpacked from the schema dictionary. Args: schema (str): Schema name. table (str): Table name. columns (list): List of columns. Return: SQLTable: Source table instance. """ columns = tuple(SQLColumn.from_json(**col) for col in columns) return cls(schema, table, columns) def __init__(self, schema, name, columns): """ Initialise SQLTable object with values. Args: schema (str): Schema name. name (str): Table name. columns (tuple): Collection of SQLColumn instances. """ self.schema = schema self.name = name self.columns = columns def sql(self): """ Generate source SQL definition for each column. Returns: tuple: Source SQL definitions for each table column. """ return tuple(col.sql() for col in self.columns) def sql_snowflake(self, db_type): """ Generate Snowflake SQL definition for each column. Returns: tuple: Snowflake SQL definitions for each table column. """ return tuple(col.sql_snowflake(db_type) for col in self.columns)