<s> """<STR_LIT>""" <EOL> from . . utils . helpers import stringify_credentials <EOL> from . column import Column <EOL> from . parsers import create_exists_clone <EOL> from . parsers import parse_create_table <EOL> from . relation import Relation <EOL> from . select_statement import SelectStatement <EOL> from . sql import SqlScript <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> def comma_seperated ( elements ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:U+002C>' . join ( elements ) <EOL> class Table ( Relation ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , sql ) : <EOL> """<STR_LIT>""" <EOL> super ( Table , self ) . __init__ ( ) <EOL> if isinstance ( sql , str ) : <EOL> sql = SqlScript ( sql ) <EOL> if isinstance ( sql , SqlScript ) : <EOL> sql = sql . statements [ <NUM_LIT:0> ] <EOL> parameters = parse_create_table ( sql . sql ( ) ) <EOL> self . sql_statement = sql <EOL> self . parameters = parameters <EOL> self . full_name = parameters . get ( '<STR_LIT>' ) <EOL> self . temporary = parameters . get ( '<STR_LIT>' ) <EOL> self . exists_check = parameters . get ( '<STR_LIT>' , False ) <EOL> self . sort_keys = parameters . get ( '<STR_LIT>' , list ( ) ) <EOL> self . dist_keys = parameters . get ( '<STR_LIT>' , list ( ) ) <EOL> self . diststyle = parameters . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _constraints = parameters . get ( '<STR_LIT>' , list ( ) ) <EOL> self . _columns = dict ( ) <EOL> for column_params in parameters . get ( '<STR_LIT>' , list ( ) ) : <EOL> column_name = column_params [ '<STR_LIT>' ] <EOL> self . _columns [ column_name ] = Column ( ** column_params ) <EOL> self . schema_name , self . table_name = self . initialize_name ( ) <EOL> self . update_attributes_from_columns ( ) <EOL> self . update_columns_with_constrains ( ) <EOL> def update_attributes_from_columns ( self ) : <EOL> """<STR_LIT>""" <EOL> distkeys = self . dist_keys <EOL> sortkeys = self . sort_keys <EOL> for column in self . _columns . values ( ) : <EOL> if column . is_distkey : <EOL> distkeys . append ( column . name ) <EOL> if column . is_sortkey : <EOL> sortkeys . append ( column . name ) <EOL> self . dist_keys = list ( set ( distkeys ) ) <EOL> self . sort_keys = list ( set ( sortkeys ) ) <EOL> def update_columns_with_constrains ( self ) : <EOL> """<STR_LIT>""" <EOL> for constraint in self . _constraints : <EOL> for col_name in constraint . get ( '<STR_LIT>' , list ( ) ) : <EOL> self . _columns [ col_name ] . primary = True <EOL> def columns ( self ) : <EOL> """<STR_LIT>""" <EOL> return sorted ( self . _columns . values ( ) , key = lambda x : x . position ) <EOL> def column ( self , column_name ) : <EOL> """<STR_LIT>""" <EOL> return self . _columns . get ( column_name , None ) <EOL> @ property <EOL> def primary_keys ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ c for c in self . columns ( ) if c . primary ] <EOL> @ property <EOL> def primary_key_names ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ c . name for c in self . columns ( ) if c . primary ] <EOL> def foreign_key_references ( self ) : <EOL> """<STR_LIT>""" <EOL> result = list ( ) <EOL> for column in self . columns ( ) : <EOL> if column . fk_table is not None : <EOL> result . append ( ( <EOL> [ column . name ] , column . fk_table , [ column . fk_reference ] ) ) <EOL> for constraint in self . _constraints : <EOL> if '<STR_LIT>' in constraint : <EOL> result . append ( ( constraint . get ( '<STR_LIT>' ) , <EOL> constraint . get ( '<STR_LIT>' ) , <EOL> constraint . get ( '<STR_LIT>' ) ) ) <EOL> return result <EOL> @ property <EOL> def dependencies ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ table_name for _ , table_name , _ <EOL> in self . foreign_key_references ( ) ] <EOL> def temporary_clone_script ( self ) : <EOL> """<STR_LIT>""" <EOL> table_name = self . table_name + '<STR_LIT>' <EOL> columns = comma_seperated ( <EOL> [ '<STR_LIT>' % ( c . column_name , c . column_type ) for c in self . columns ( ) ] ) <EOL> if self . primary_keys : <EOL> sql = """<STR_LIT>""" . format ( <EOL> table_name = table_name , columns = columns , <EOL> primary_keys = comma_seperated ( self . primary_key_names ) ) <EOL> else : <EOL> sql = """<STR_LIT>""" . format ( <EOL> table_name = table_name , columns = columns ) <EOL> return SqlScript ( sql ) <EOL> def exists_clone_script ( self ) : <EOL> """<STR_LIT>""" <EOL> return SqlScript ( create_exists_clone ( self . sql_statement . sql ( ) ) ) <EOL> def drop_script ( self ) : <EOL> """<STR_LIT>""" <EOL> return SqlScript ( '<STR_LIT>' % self . full_name ) <EOL> def analyze_script ( self ) : <EOL> """<STR_LIT>""" <EOL> return SqlScript ( '<STR_LIT>' % self . full_name ) <EOL> def rename_script ( self , new_name ) : <EOL> """<STR_LIT>""" <EOL> return SqlScript ( <EOL> '<STR_LIT>' % ( self . full_name , new_name ) ) <EOL> def delete_script ( self , where_condition = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return SqlScript ( <EOL> '<STR_LIT>' % ( self . full_name , where_condition ) ) <EOL> def foreign_key_reference_script ( self , source_columns , reference_name , <EOL> reference_columns ) : <EOL> """<STR_LIT>""" <EOL> sql = """<STR_LIT>""" . format ( source_name = self . full_name , <EOL> source_columns = comma_seperated ( source_columns ) , <EOL> reference_name = reference_name , <EOL> reference_columns = comma_seperated ( reference_columns ) ) <EOL> return SqlScript ( sql ) <EOL> def select_duplicates_script ( self ) : <EOL> """<STR_LIT>""" <EOL> pk_columns = comma_seperated ( self . primary_key_names ) <EOL> sql = """<STR_LIT>""" . format ( table_name = self . full_name , <EOL> pk_columns = pk_columns ) <EOL> return SqlScript ( sql ) <EOL> def _source_sql ( self , source_relation , filter_clause = None ) : <EOL> """<STR_LIT>""" <EOL> if not ( isinstance ( source_relation , Relation ) or <EOL> isinstance ( source_relation , SelectStatement ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if len ( self . columns ( ) ) < len ( source_relation . columns ( ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if isinstance ( source_relation , SelectStatement ) : <EOL> source_sql = '<STR_LIT:(>' + source_relation . sql ( ) + '<STR_LIT:)>' <EOL> else : <EOL> source_sql = source_relation . full_name <EOL> if filter_clause is not None : <EOL> source_sql = source_sql + '<STR_LIT:U+0020>' + filter_clause <EOL> return source_sql <EOL> def insert_script ( self , source_relation , filter_clause = None ) : <EOL> """<STR_LIT>""" <EOL> sql = '<STR_LIT>' % ( <EOL> self . full_name , self . _source_sql ( source_relation , filter_clause ) ) <EOL> return SqlScript ( sql ) <EOL> def delete_matching_rows_script ( self , source_relation ) : <EOL> """<STR_LIT>""" <EOL> if len ( self . primary_keys ) == <NUM_LIT:0> : <EOL> raise RuntimeError ( <EOL> '<STR_LIT>' ) <EOL> source_col_names , pk_names = [ ] , [ ] <EOL> for column in self . columns ( ) : <EOL> if column . primary : <EOL> pk_names . append ( column . name ) <EOL> source_col_names . append ( column . name ) <EOL> where_condition = '<STR_LIT>' % ( <EOL> comma_seperated ( pk_names ) , comma_seperated ( source_col_names ) , <EOL> self . _source_sql ( source_relation ) ) <EOL> return self . delete_script ( where_condition ) <EOL> def de_duplication_script ( self ) : <EOL> """<STR_LIT>""" <EOL> if len ( self . primary_keys ) == <NUM_LIT:0> : <EOL> logger . error ( <EOL> '<STR_LIT>' ) <EOL> return SqlScript ( ) <EOL> script = self . temporary_clone_script ( ) <EOL> column_names = [ c . name for c in self . columns ( ) ] <EOL> temp_table = self . __class__ ( script ) <EOL> script . append ( temp_table . insert_script ( self ) ) <EOL> script . append ( self . delete_script ( ) ) <EOL> sql = """<STR_LIT>""" . format ( table_name = self . full_name , <EOL> column_names = comma_seperated ( column_names ) , <EOL> pk_names = comma_seperated ( self . primary_key_names ) , <EOL> temp_table = temp_table . full_name ) <EOL> script . append ( SqlScript ( sql ) ) <EOL> return script <EOL> def upsert_script ( self , source_relation , enforce_primary_key = True , <EOL> delete_existing = False , filter_clause = None ) : <EOL> """<STR_LIT>""" <EOL> script = self . temporary_clone_script ( ) <EOL> temp_table = self . __class__ ( script ) <EOL> script . append ( temp_table . insert_script ( source_relation , filter_clause ) ) <EOL> if enforce_primary_key : <EOL> script . append ( temp_table . de_duplication_script ( ) ) <EOL> if delete_existing : <EOL> script . append ( self . delete_script ( ) ) <EOL> else : <EOL> script . append ( self . delete_matching_rows_script ( temp_table ) ) <EOL> script . append ( self . insert_script ( temp_table ) ) <EOL> script . append ( temp_table . drop_script ( ) ) <EOL> return script <EOL> def check_not_exists_script ( self ) : <EOL> """<STR_LIT>""" <EOL> return SqlScript ( """<STR_LIT>""" % ( self . schema_name , self . table_name ) ) <EOL> def unload_script ( self , s3_path , access_key , secret_key , token = None ) : <EOL> """<STR_LIT>""" <EOL> script = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) . format ( <EOL> select_script = self . select_script ( ) , <EOL> s3_path = s3_path , <EOL> creds = stringify_credentials ( access_key , secret_key , token ) <EOL> ) <EOL> return SqlScript ( script ) <EOL> def load_script ( self , s3_path , access_key , secret_key , token = None ) : <EOL> """<STR_LIT>""" <EOL> script = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) . format ( <EOL> name = self . full_name , <EOL> s3_path = s3_path , <EOL> creds = stringify_credentials ( access_key , secret_key , token ) <EOL> ) <EOL> return SqlScript ( script ) </s>
