<s> from collections import OrderedDict <EOL> from . . index import Index <EOL> from . . table import Table <EOL> from . . identifier import Identifier <EOL> from . . exceptions import DBALException <EOL> from ... utils import basestring <EOL> class Platform ( object ) : <EOL> _keywords = None <EOL> CREATE_INDEXES = <NUM_LIT:1> <EOL> CREATE_FOREIGNKEYS = <NUM_LIT:2> <EOL> INTERNAL_TYPE_MAPPING = { } <EOL> def __init__ ( self , version = None ) : <EOL> self . _version = None <EOL> def get_default_value_declaration_sql ( self , field ) : <EOL> default = '<STR_LIT>' <EOL> if not field . get ( '<STR_LIT>' ) : <EOL> default = '<STR_LIT>' <EOL> if '<STR_LIT:default>' in field and field [ '<STR_LIT:default>' ] is not None : <EOL> default = '<STR_LIT>' % field [ '<STR_LIT:default>' ] <EOL> if '<STR_LIT:type>' in field : <EOL> type = field [ '<STR_LIT:type>' ] <EOL> if type in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> default = '<STR_LIT>' % field [ '<STR_LIT:default>' ] <EOL> elif type in [ '<STR_LIT>' , '<STR_LIT>' ] and field [ '<STR_LIT:default>' ] in [ self . get_current_timestamp_sql ( ) , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> default = '<STR_LIT>' % self . get_current_timestamp_sql ( ) <EOL> elif type in [ '<STR_LIT:time>' ] and field [ '<STR_LIT:default>' ] in [ self . get_current_time_sql ( ) , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> default = '<STR_LIT>' % self . get_current_time_sql ( ) <EOL> elif type in [ '<STR_LIT:date>' ] and field [ '<STR_LIT:default>' ] in [ self . get_current_date_sql ( ) , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> default = '<STR_LIT>' % self . get_current_date_sql ( ) <EOL> elif type in [ '<STR_LIT>' ] : <EOL> default = '<STR_LIT>' % self . convert_booleans ( field [ '<STR_LIT:default>' ] ) <EOL> return default <EOL> def convert_booleans ( self , item ) : <EOL> if isinstance ( item , list ) : <EOL> for i , value in enumerate ( item ) : <EOL> if isinstance ( value , bool ) : <EOL> item [ i ] = int ( value ) <EOL> elif isinstance ( item , bool ) : <EOL> item = int ( item ) <EOL> return item <EOL> def get_check_declaration_sql ( self , definition ) : <EOL> """<STR_LIT>""" <EOL> constraints = [ ] <EOL> for field , def_ in definition . items ( ) : <EOL> if isinstance ( def_ , basestring ) : <EOL> constraints . append ( '<STR_LIT>' % def_ ) <EOL> else : <EOL> if '<STR_LIT>' in def_ : <EOL> constraints . append ( '<STR_LIT>' % ( field , def_ [ '<STR_LIT>' ] ) ) <EOL> if '<STR_LIT>' in def_ : <EOL> constraints . append ( '<STR_LIT>' % ( field , def_ [ '<STR_LIT>' ] ) ) <EOL> return '<STR_LIT:U+002CU+0020>' . join ( constraints ) <EOL> def get_unique_constraint_declaration_sql ( self , name , index ) : <EOL> """<STR_LIT>""" <EOL> columns = index . get_quoted_columns ( self ) <EOL> name = Identifier ( name ) <EOL> if not columns : <EOL> raise DBALException ( '<STR_LIT>' ) <EOL> return '<STR_LIT>' % ( name . get_quoted_name ( self ) , <EOL> self . get_index_field_declaration_list_sql ( columns ) , <EOL> self . get_partial_index_sql ( index ) ) <EOL> def get_index_declaration_sql ( self , name , index ) : <EOL> """<STR_LIT>""" <EOL> columns = index . get_quoted_columns ( self ) <EOL> name = Identifier ( name ) <EOL> if not columns : <EOL> raise DBALException ( '<STR_LIT>' ) <EOL> return '<STR_LIT>' % ( self . get_create_index_sql_flags ( index ) , <EOL> name . get_quoted_name ( self ) , <EOL> self . get_index_field_declaration_list_sql ( columns ) , <EOL> self . get_partial_index_sql ( index ) ) <EOL> def get_foreign_key_declaration_sql ( self , foreign_key ) : <EOL> """<STR_LIT>""" <EOL> sql = self . get_foreign_key_base_declaration_sql ( foreign_key ) <EOL> sql += self . get_advanced_foreign_key_options_sql ( foreign_key ) <EOL> return sql <EOL> def get_advanced_foreign_key_options_sql ( self , foreign_key ) : <EOL> """<STR_LIT>""" <EOL> query = '<STR_LIT>' <EOL> if self . supports_foreign_key_on_update ( ) and foreign_key . has_option ( '<STR_LIT>' ) : <EOL> query += '<STR_LIT>' % self . get_foreign_key_referential_action_sql ( foreign_key . get_option ( '<STR_LIT>' ) ) <EOL> if foreign_key . has_option ( '<STR_LIT>' ) : <EOL> query += '<STR_LIT>' % self . get_foreign_key_referential_action_sql ( foreign_key . get_option ( '<STR_LIT>' ) ) <EOL> return query <EOL> def get_foreign_key_referential_action_sql ( self , action ) : <EOL> """<STR_LIT>""" <EOL> action = action . upper ( ) <EOL> if action not in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> raise DBALException ( '<STR_LIT>' % action ) <EOL> return action <EOL> def get_foreign_key_base_declaration_sql ( self , foreign_key ) : <EOL> """<STR_LIT>""" <EOL> sql = '<STR_LIT>' <EOL> if foreign_key . get_name ( ) : <EOL> sql += '<STR_LIT>' % foreign_key . get_quoted_name ( self ) <EOL> sql += '<STR_LIT>' <EOL> if not foreign_key . get_local_columns ( ) : <EOL> raise DBALException ( '<STR_LIT>' ) <EOL> if not foreign_key . get_foreign_columns ( ) : <EOL> raise DBALException ( '<STR_LIT>' ) <EOL> if not foreign_key . get_foreign_table_name ( ) : <EOL> raise DBALException ( '<STR_LIT>' ) <EOL> sql += '<STR_LIT>' % ( '<STR_LIT:U+002CU+0020>' . join ( foreign_key . get_quoted_local_columns ( self ) ) , <EOL> foreign_key . get_quoted_foreign_table_name ( self ) , <EOL> '<STR_LIT:U+002CU+0020>' . join ( foreign_key . get_quoted_foreign_columns ( self ) ) ) <EOL> return sql <EOL> def get_current_date_sql ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_current_time_sql ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_current_timestamp_sql ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_sql_type_declaration ( self , column ) : <EOL> internal_type = column [ '<STR_LIT:type>' ] <EOL> return getattr ( self , '<STR_LIT>' % internal_type ) ( column ) <EOL> def get_column_declaration_list_sql ( self , fields ) : <EOL> """<STR_LIT>""" <EOL> query_fields = [ ] <EOL> for name , field in fields . items ( ) : <EOL> query_fields . append ( self . get_column_declaration_sql ( name , field ) ) <EOL> return '<STR_LIT:U+002CU+0020>' . join ( query_fields ) <EOL> def get_column_declaration_sql ( self , name , field ) : <EOL> if '<STR_LIT>' in field : <EOL> column_def = self . get_custom_type_declaration_sql ( field ) <EOL> else : <EOL> default = self . get_default_value_declaration_sql ( field ) <EOL> charset = field . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if charset : <EOL> charset = '<STR_LIT:U+0020>' + self . get_column_charset_declaration_sql ( charset ) <EOL> collation = field . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if charset : <EOL> charset = '<STR_LIT:U+0020>' + self . get_column_collation_declaration_sql ( charset ) <EOL> notnull = field . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if notnull : <EOL> notnull = '<STR_LIT>' <EOL> else : <EOL> notnull = '<STR_LIT>' <EOL> unique = field . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if unique : <EOL> unique = '<STR_LIT:U+0020>' + self . get_unique_field_declaration_sql ( ) <EOL> else : <EOL> unique = '<STR_LIT>' <EOL> check = field . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> type_decl = self . get_sql_type_declaration ( field ) <EOL> column_def = type_decl + charset + default + notnull + unique + check + collation <EOL> return name + '<STR_LIT:U+0020>' + column_def <EOL> def get_custom_type_declaration_sql ( self , column_def ) : <EOL> return column_def [ '<STR_LIT>' ] <EOL> def get_column_charset_declaration_sql ( self , charset ) : <EOL> return '<STR_LIT>' <EOL> def get_column_collation_declaration_sql ( self , collation ) : <EOL> if self . supports_column_collation ( ) : <EOL> return '<STR_LIT>' % collation <EOL> return '<STR_LIT>' <EOL> def supports_column_collation ( self ) : <EOL> return False <EOL> def get_unique_field_declaration_sql ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_string_type_declaration_sql ( self , column ) : <EOL> if '<STR_LIT>' not in column : <EOL> column [ '<STR_LIT>' ] = self . get_varchar_default_length ( ) <EOL> fixed = column . get ( '<STR_LIT>' , False ) <EOL> if column [ '<STR_LIT>' ] > self . get_varchar_max_length ( ) : <EOL> return self . get_clob_type_declaration_sql ( column ) <EOL> return self . get_varchar_type_declaration_sql_snippet ( column [ '<STR_LIT>' ] , fixed ) <EOL> def get_binary_type_declaration_sql ( self , column ) : <EOL> if '<STR_LIT>' not in column : <EOL> column [ '<STR_LIT>' ] = self . get_binary_default_length ( ) <EOL> fixed = column . get ( '<STR_LIT>' , False ) <EOL> if column [ '<STR_LIT>' ] > self . get_binary_max_length ( ) : <EOL> return self . get_blob_type_declaration_sql ( column ) <EOL> return self . get_binary_type_declaration_sql_snippet ( column [ '<STR_LIT>' ] , fixed ) <EOL> def get_varchar_type_declaration_sql_snippet ( self , length , fixed ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def get_binary_type_declaration_sql_snippet ( self , length , fixed ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def get_decimal_type_declaration_sql ( self , column ) : <EOL> if '<STR_LIT>' not in column or not column [ '<STR_LIT>' ] : <EOL> column [ '<STR_LIT>' ] = <NUM_LIT:10> <EOL> if '<STR_LIT>' not in column or not column [ '<STR_LIT>' ] : <EOL> column [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> return '<STR_LIT>' % ( column [ '<STR_LIT>' ] , column [ '<STR_LIT>' ] ) <EOL> def get_json_type_declaration_sql ( self , column ) : <EOL> return self . get_clob_type_declaration_sql ( column ) <EOL> def get_clob_type_declaration_sql ( self , column ) : <EOL> raise NotImplementedError ( ) <EOL> def get_text_type_declaration_sql ( self , column ) : <EOL> return self . get_clob_type_declaration_sql ( column ) <EOL> def get_blob_type_declaration_sql ( self , column ) : <EOL> raise NotImplementedError ( ) <EOL> def get_varchar_default_length ( self ) : <EOL> return <NUM_LIT:255> <EOL> def get_varchar_max_length ( self ) : <EOL> return <NUM_LIT> <EOL> def get_binary_default_length ( self ) : <EOL> return <NUM_LIT:255> <EOL> def get_binary_max_length ( self ) : <EOL> return <NUM_LIT> <EOL> def get_column_options ( self ) : <EOL> return [ ] <EOL> def get_type_mapping ( self , db_type ) : <EOL> return self . INTERNAL_TYPE_MAPPING [ db_type ] <EOL> def get_reserved_keywords_list ( self ) : <EOL> if self . _keywords : <EOL> return self . _keywords <EOL> klass = self . _get_reserved_keywords_class ( ) <EOL> keywords = klass ( ) <EOL> self . _keywords = keywords <EOL> return keywords <EOL> def _get_reserved_keywords_class ( self ) : <EOL> raise NotImplementedError <EOL> def get_index_field_declaration_list_sql ( self , fields ) : <EOL> """<STR_LIT>""" <EOL> ret = [ ] <EOL> for field in fields : <EOL> ret . append ( field ) <EOL> return '<STR_LIT:U+002CU+0020>' . join ( ret ) <EOL> def get_create_index_sql ( self , index , table ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( table , Table ) : <EOL> table = table . get_quoted_name ( self ) <EOL> name = index . get_quoted_name ( self ) <EOL> columns = index . get_quoted_columns ( self ) <EOL> if not columns : <EOL> raise DBALException ( '<STR_LIT>' ) <EOL> if index . is_primary ( ) : <EOL> return self . get_create_primary_key_sql ( index , table ) <EOL> query = '<STR_LIT>' % ( self . get_create_index_sql_flags ( index ) , name , table ) <EOL> query += '<STR_LIT>' % ( self . get_index_field_declaration_list_sql ( columns ) , <EOL> self . get_partial_index_sql ( index ) ) <EOL> return query <EOL> def get_partial_index_sql ( self , index ) : <EOL> """<STR_LIT>""" <EOL> if self . supports_partial_indexes ( ) and index . has_option ( '<STR_LIT>' ) : <EOL> return '<STR_LIT>' % index . get_option ( '<STR_LIT>' ) <EOL> return '<STR_LIT>' <EOL> def get_create_index_sql_flags ( self , index ) : <EOL> """<STR_LIT>""" <EOL> if index . is_unique ( ) : <EOL> return '<STR_LIT>' <EOL> return '<STR_LIT>' <EOL> def get_create_primary_key_sql ( self , index , table ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( table , <EOL> self . get_index_field_declaration_list_sql ( index . get_quoted_columns ( self ) ) ) <EOL> def get_create_foreign_key_sql ( self , foreign_key , table ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( table , Table ) : <EOL> table = table . get_quoted_name ( self ) <EOL> query = '<STR_LIT>' % ( table , self . get_foreign_key_declaration_sql ( foreign_key ) ) <EOL> return query <EOL> def get_drop_table_sql ( self , table ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( table , Table ) : <EOL> table = table . get_quoted_name ( self ) <EOL> return '<STR_LIT>' % table <EOL> def get_drop_index_sql ( self , index , table = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( index , Index ) : <EOL> index = index . get_quoted_name ( self ) <EOL> return '<STR_LIT>' % index <EOL> def get_create_table_sql ( self , table , create_flags = CREATE_INDEXES ) : <EOL> """<STR_LIT>""" <EOL> table_name = table . get_quoted_name ( self ) <EOL> options = dict ( ( k , v ) for k , v in table . get_options ( ) . items ( ) ) <EOL> options [ '<STR_LIT>' ] = OrderedDict ( ) <EOL> options [ '<STR_LIT>' ] = OrderedDict ( ) <EOL> options [ '<STR_LIT>' ] = [ ] <EOL> if create_flags & self . CREATE_INDEXES > <NUM_LIT:0> : <EOL> for index in table . get_indexes ( ) . values ( ) : <EOL> if index . is_primary ( ) : <EOL> options [ '<STR_LIT>' ] = index . get_quoted_columns ( self ) <EOL> options [ '<STR_LIT>' ] = index <EOL> else : <EOL> options [ '<STR_LIT>' ] [ index . get_quoted_name ( self ) ] = index <EOL> columns = OrderedDict ( ) <EOL> for column in table . get_columns ( ) . values ( ) : <EOL> column_data = column . to_dict ( ) <EOL> column_data [ '<STR_LIT:name>' ] = column . get_quoted_name ( self ) <EOL> if column . has_platform_option ( '<STR_LIT:version>' ) : <EOL> column_data [ '<STR_LIT:version>' ] = column . get_platform_option ( '<STR_LIT:version>' ) <EOL> else : <EOL> column_data [ '<STR_LIT:version>' ] = False <EOL> if column_data [ '<STR_LIT:type>' ] == '<STR_LIT:string>' and column_data [ '<STR_LIT>' ] is None : <EOL> column_data [ '<STR_LIT>' ] = <NUM_LIT:255> <EOL> if column . get_name ( ) in options [ '<STR_LIT>' ] : <EOL> column_data [ '<STR_LIT>' ] = True <EOL> columns [ column_data [ '<STR_LIT:name>' ] ] = column_data <EOL> if create_flags & self . CREATE_FOREIGNKEYS > <NUM_LIT:0> : <EOL> options [ '<STR_LIT>' ] = [ ] <EOL> for fk in table . get_foreign_keys ( ) . values ( ) : <EOL> options [ '<STR_LIT>' ] . append ( fk ) <EOL> sql = self . _get_create_table_sql ( table_name , columns , options ) <EOL> return sql <EOL> def _get_create_table_sql ( self , table_name , columns , options = None ) : <EOL> """<STR_LIT>""" <EOL> options = options or { } <EOL> column_list_sql = self . get_column_declaration_list_sql ( columns ) <EOL> if options . get ( '<STR_LIT>' ) : <EOL> for name , definition in options [ '<STR_LIT>' ] . items ( ) : <EOL> column_list_sql += '<STR_LIT>' % self . get_unique_constraint_declaration_sql ( name , definition ) <EOL> if options . get ( '<STR_LIT>' ) : <EOL> column_list_sql += '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( options [ '<STR_LIT>' ] ) <EOL> if options . get ( '<STR_LIT>' ) : <EOL> for index , definition in options [ '<STR_LIT>' ] : <EOL> column_list_sql += '<STR_LIT>' % self . get_index_declaration_sql ( index , definition ) <EOL> query = '<STR_LIT>' % ( table_name , column_list_sql ) <EOL> check = self . get_check_declaration_sql ( columns ) <EOL> if check : <EOL> query += '<STR_LIT>' % check <EOL> query += '<STR_LIT:)>' <EOL> sql = [ query ] <EOL> if options . get ( '<STR_LIT>' ) : <EOL> for definition in options [ '<STR_LIT>' ] : <EOL> sql . append ( self . get_create_foreign_key_sql ( definition , table_name ) ) <EOL> return sql <EOL> def quote_identifier ( self , string ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:.>' in string : <EOL> parts = list ( map ( self . quote_single_identifier , string . split ( '<STR_LIT:.>' ) ) ) <EOL> return '<STR_LIT:.>' . join ( parts ) <EOL> return self . quote_single_identifier ( string ) <EOL> def quote_single_identifier ( self , string ) : <EOL> """<STR_LIT>""" <EOL> c = self . get_identifier_quote_character ( ) <EOL> return '<STR_LIT>' % ( c , string . replace ( c , c + c ) , c ) <EOL> def get_identifier_quote_character ( self ) : <EOL> return '<STR_LIT:">' <EOL> def supports_indexes ( self ) : <EOL> return True <EOL> def supports_partial_indexes ( self ) : <EOL> return False <EOL> def supports_alter_table ( self ) : <EOL> return True <EOL> def supports_transactions ( self ) : <EOL> return True <EOL> def supports_primary_constraints ( self ) : <EOL> return True <EOL> def supports_foreign_key_constraints ( self ) : <EOL> return True <EOL> def supports_foreign_key_on_update ( self ) : <EOL> return self . supports_foreign_key_constraints ( ) </s>
