<s> """<STR_LIT>""" <EOL> import re <EOL> import sys <EOL> import json <EOL> from ... import schema as sa_schema <EOL> from ... import exc , log , sql , util <EOL> from ... sql import compiler , elements <EOL> from array import array as _array <EOL> from ... engine import reflection <EOL> from ... engine import default <EOL> from ... import types as sqltypes <EOL> from ... util import topological <EOL> from ... types import DATE , BOOLEAN , BLOB , BINARY , VARBINARY <EOL> from . import reflection as _reflection <EOL> from . types import BIGINT , BIT , CHAR , DECIMAL , DATETIME , DOUBLE , FLOAT , INTEGER , LONGBLOB , LONGTEXT , MEDIUMBLOB , MEDIUMINT , MEDIUMTEXT , NCHAR , NUMERIC , NVARCHAR , REAL , SMALLINT , TEXT , TIME , TIMESTAMP , TINYBLOB , TINYINT , TINYTEXT , VARCHAR , YEAR <EOL> from . types import _StringType , _IntegerType , _NumericType , _FloatType , _MatchType <EOL> from . enumerated import ENUM , SET <EOL> from . json import JSON , JSONIndexType , JSONPathType <EOL> RESERVED_WORDS = set ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:all>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT:default>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:false>' , '<STR_LIT>' , '<STR_LIT:float>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:ignore>' , '<STR_LIT>' , '<STR_LIT:index>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:int>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:key>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:left>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:null>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:replace>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT:right>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:to>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:true>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> ] ) <EOL> AUTOCOMMIT_RE = re . compile ( <EOL> r'<STR_LIT>' , <EOL> re . I | re . UNICODE ) <EOL> SET_RE = re . compile ( <EOL> r'<STR_LIT>' , <EOL> re . I | re . UNICODE ) <EOL> MSTime = TIME <EOL> MSSet = SET <EOL> MSEnum = ENUM <EOL> MSLongBlob = LONGBLOB <EOL> MSMediumBlob = MEDIUMBLOB <EOL> MSTinyBlob = TINYBLOB <EOL> MSBlob = BLOB <EOL> MSBinary = BINARY <EOL> MSVarBinary = VARBINARY <EOL> MSNChar = NCHAR <EOL> MSNVarChar = NVARCHAR <EOL> MSChar = CHAR <EOL> MSString = VARCHAR <EOL> MSLongText = LONGTEXT <EOL> MSMediumText = MEDIUMTEXT <EOL> MSTinyText = TINYTEXT <EOL> MSText = TEXT <EOL> MSYear = YEAR <EOL> MSTimeStamp = TIMESTAMP <EOL> MSBit = BIT <EOL> MSSmallInteger = SMALLINT <EOL> MSTinyInteger = TINYINT <EOL> MSMediumInteger = MEDIUMINT <EOL> MSBigInteger = BIGINT <EOL> MSNumeric = NUMERIC <EOL> MSDecimal = DECIMAL <EOL> MSDouble = DOUBLE <EOL> MSReal = REAL <EOL> MSFloat = FLOAT <EOL> MSInteger = INTEGER <EOL> colspecs = { <EOL> _IntegerType : _IntegerType , <EOL> _NumericType : _NumericType , <EOL> _FloatType : _FloatType , <EOL> sqltypes . Numeric : NUMERIC , <EOL> sqltypes . Float : FLOAT , <EOL> sqltypes . Time : TIME , <EOL> sqltypes . Enum : ENUM , <EOL> sqltypes . MatchType : _MatchType , <EOL> sqltypes . JSON : JSON , <EOL> sqltypes . JSON . JSONIndexType : JSONIndexType , <EOL> sqltypes . JSON . JSONPathType : JSONPathType <EOL> } <EOL> ischema_names = { <EOL> '<STR_LIT>' : BIGINT , <EOL> '<STR_LIT>' : BINARY , <EOL> '<STR_LIT>' : BIT , <EOL> '<STR_LIT>' : BLOB , <EOL> '<STR_LIT>' : BOOLEAN , <EOL> '<STR_LIT>' : CHAR , <EOL> '<STR_LIT:date>' : DATE , <EOL> '<STR_LIT>' : DATETIME , <EOL> '<STR_LIT>' : DECIMAL , <EOL> '<STR_LIT>' : DOUBLE , <EOL> '<STR_LIT>' : ENUM , <EOL> '<STR_LIT>' : DECIMAL , <EOL> '<STR_LIT:float>' : FLOAT , <EOL> '<STR_LIT:int>' : INTEGER , <EOL> '<STR_LIT>' : INTEGER , <EOL> '<STR_LIT>' : JSON , <EOL> '<STR_LIT>' : LONGBLOB , <EOL> '<STR_LIT>' : LONGTEXT , <EOL> '<STR_LIT>' : MEDIUMBLOB , <EOL> '<STR_LIT>' : MEDIUMINT , <EOL> '<STR_LIT>' : MEDIUMTEXT , <EOL> '<STR_LIT>' : NCHAR , <EOL> '<STR_LIT>' : NVARCHAR , <EOL> '<STR_LIT>' : NUMERIC , <EOL> '<STR_LIT>' : SET , <EOL> '<STR_LIT>' : SMALLINT , <EOL> '<STR_LIT:text>' : TEXT , <EOL> '<STR_LIT:time>' : TIME , <EOL> '<STR_LIT>' : TIMESTAMP , <EOL> '<STR_LIT>' : TINYBLOB , <EOL> '<STR_LIT>' : TINYINT , <EOL> '<STR_LIT>' : TINYTEXT , <EOL> '<STR_LIT>' : VARBINARY , <EOL> '<STR_LIT>' : VARCHAR , <EOL> '<STR_LIT>' : YEAR , <EOL> } <EOL> class MySQLExecutionContext ( default . DefaultExecutionContext ) : <EOL> def should_autocommit_text ( self , statement ) : <EOL> return AUTOCOMMIT_RE . match ( statement ) <EOL> class MySQLCompiler ( compiler . SQLCompiler ) : <EOL> render_table_with_column_in_update_from = True <EOL> """<STR_LIT>""" <EOL> extract_map = compiler . SQLCompiler . extract_map . copy ( ) <EOL> extract_map . update ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def visit_random_func ( self , fn , ** kw ) : <EOL> return "<STR_LIT>" % self . function_argspec ( fn ) <EOL> def visit_utc_timestamp_func ( self , fn , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_sysdate_func ( self , fn , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_json_getitem_op_binary ( self , binary , operator , ** kw ) : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( binary . left ) , <EOL> self . process ( binary . right ) ) <EOL> def visit_json_path_getitem_op_binary ( self , binary , operator , ** kw ) : <EOL> return "<STR_LIT>" % ( <EOL> self . process ( binary . left ) , <EOL> self . process ( binary . right ) ) <EOL> def visit_concat_op_binary ( self , binary , operator , ** kw ) : <EOL> return "<STR_LIT>" % ( self . process ( binary . left ) , <EOL> self . process ( binary . right ) ) <EOL> def visit_match_op_binary ( self , binary , operator , ** kw ) : <EOL> return "<STR_LIT>" % ( self . process ( binary . left ) , self . process ( binary . right ) ) <EOL> def get_from_hint_text ( self , table , text ) : <EOL> return text <EOL> def visit_typeclause ( self , typeclause , type_ = None ) : <EOL> if type_ is None : <EOL> type_ = typeclause . type . dialect_impl ( self . dialect ) <EOL> if isinstance ( type_ , sqltypes . TypeDecorator ) : <EOL> return self . visit_typeclause ( typeclause , type_ . impl ) <EOL> elif isinstance ( type_ , sqltypes . Integer ) : <EOL> if getattr ( type_ , '<STR_LIT>' , False ) : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return '<STR_LIT>' <EOL> elif isinstance ( type_ , sqltypes . TIMESTAMP ) : <EOL> return '<STR_LIT>' <EOL> elif isinstance ( type_ , ( sqltypes . DECIMAL , sqltypes . DateTime , <EOL> sqltypes . Date , sqltypes . Time ) ) : <EOL> return self . dialect . type_compiler . process ( type_ ) <EOL> elif isinstance ( type_ , sqltypes . String ) and not isinstance ( type_ , ( ENUM , SET ) ) : <EOL> adapted = CHAR . _adapt_string_for_cast ( type_ ) <EOL> return self . dialect . type_compiler . process ( adapted ) <EOL> elif isinstance ( type_ , sqltypes . _Binary ) : <EOL> return '<STR_LIT>' <EOL> elif isinstance ( type_ , sqltypes . JSON ) : <EOL> return "<STR_LIT>" <EOL> elif isinstance ( type_ , sqltypes . NUMERIC ) : <EOL> return self . dialect . type_compiler . process ( <EOL> type_ ) . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> else : <EOL> return None <EOL> def visit_cast ( self , cast , ** kwargs ) : <EOL> if not self . dialect . _supports_cast : <EOL> util . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return self . process ( cast . clause . self_group ( ) ) <EOL> type_ = self . process ( cast . typeclause ) <EOL> if type_ is None : <EOL> util . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> self . dialect . type_compiler . process ( cast . typeclause . type ) ) <EOL> return self . process ( cast . clause . self_group ( ) ) <EOL> return '<STR_LIT>' % ( self . process ( cast . clause ) , type_ ) <EOL> def render_literal_value ( self , value , type_ ) : <EOL> value = super ( MySQLCompiler , self ) . render_literal_value ( value , type_ ) <EOL> if self . dialect . _backslash_escapes : <EOL> value = value . replace ( '<STR_LIT:\\>' , '<STR_LIT>' ) <EOL> return value <EOL> def visit_true ( self , element , ** kw ) : <EOL> return "<STR_LIT:true>" <EOL> def visit_false ( self , element , ** kw ) : <EOL> return "<STR_LIT:false>" <EOL> def get_select_precolumns ( self , select , ** kw ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( select . _distinct , util . string_types ) : <EOL> return select . _distinct . upper ( ) + "<STR_LIT:U+0020>" <EOL> elif select . _distinct : <EOL> return "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" <EOL> def visit_join ( self , join , asfrom = False , ** kwargs ) : <EOL> if join . full : <EOL> join_type = "<STR_LIT>" <EOL> elif join . isouter : <EOL> join_type = "<STR_LIT>" <EOL> else : <EOL> join_type = "<STR_LIT>" <EOL> return '<STR_LIT>' . join ( <EOL> ( self . process ( join . left , asfrom = True , ** kwargs ) , <EOL> join_type , <EOL> self . process ( join . right , asfrom = True , ** kwargs ) , <EOL> "<STR_LIT>" , <EOL> self . process ( join . onclause , ** kwargs ) ) ) <EOL> def for_update_clause ( self , select , ** kw ) : <EOL> if select . _for_update_arg . read : <EOL> return "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" <EOL> def limit_clause ( self , select , ** kw ) : <EOL> limit_clause , offset_clause = select . _limit_clause , select . _offset_clause <EOL> if limit_clause is None and offset_clause is None : <EOL> return '<STR_LIT>' <EOL> elif offset_clause is not None : <EOL> if limit_clause is None : <EOL> return '<STR_LIT>' % ( <EOL> self . process ( offset_clause , ** kw ) , <EOL> "<STR_LIT>" ) <EOL> else : <EOL> return '<STR_LIT>' % ( <EOL> self . process ( offset_clause , ** kw ) , <EOL> self . process ( limit_clause , ** kw ) ) <EOL> else : <EOL> return '<STR_LIT>' % ( self . process ( limit_clause , ** kw ) , ) <EOL> def update_limit_clause ( self , update_stmt ) : <EOL> limit = update_stmt . kwargs . get ( '<STR_LIT>' % self . dialect . name , None ) <EOL> if limit : <EOL> return "<STR_LIT>" % limit <EOL> else : <EOL> return None <EOL> def update_tables_clause ( self , update_stmt , from_table , <EOL> extra_froms , ** kw ) : <EOL> return '<STR_LIT:U+002CU+0020>' . join ( t . _compiler_dispatch ( self , asfrom = True , ** kw ) <EOL> for t in [ from_table ] + list ( extra_froms ) ) <EOL> def update_from_clause ( self , update_stmt , from_table , <EOL> extra_froms , from_hints , ** kw ) : <EOL> return None <EOL> class MySQLDDLCompiler ( compiler . DDLCompiler ) : <EOL> def get_column_specification ( self , column , ** kw ) : <EOL> """<STR_LIT>""" <EOL> colspec = [ <EOL> self . preparer . format_column ( column ) , <EOL> self . dialect . type_compiler . process ( <EOL> column . type , type_expression = column ) <EOL> ] <EOL> is_timestamp = isinstance ( column . type , sqltypes . TIMESTAMP ) <EOL> if not column . nullable : <EOL> colspec . append ( '<STR_LIT>' ) <EOL> elif column . nullable and is_timestamp : <EOL> colspec . append ( '<STR_LIT>' ) <EOL> default = self . get_column_default_string ( column ) <EOL> if default is not None : <EOL> colspec . append ( '<STR_LIT>' + default ) <EOL> if column . table is not None and column is column . table . _autoincrement_column and column . server_default is None : <EOL> colspec . append ( '<STR_LIT>' ) <EOL> return '<STR_LIT:U+0020>' . join ( colspec ) <EOL> def post_create_table ( self , table ) : <EOL> """<STR_LIT>""" <EOL> table_opts = [ ] <EOL> opts = dict ( <EOL> ( <EOL> k [ len ( self . dialect . name ) + <NUM_LIT:1> : ] . upper ( ) , <EOL> v <EOL> ) <EOL> for k , v in table . kwargs . items ( ) <EOL> if k . startswith ( '<STR_LIT>' % self . dialect . name ) <EOL> ) <EOL> for opt in topological . sort ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] , opts ) : <EOL> arg = opts [ opt ] <EOL> if opt in _reflection . _options_of_type_string : <EOL> arg = "<STR_LIT>" % arg . replace ( "<STR_LIT:\\>" , "<STR_LIT>" ) . replace ( "<STR_LIT:'>" , "<STR_LIT>" ) <EOL> if opt in ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) : <EOL> opt = opt . replace ( '<STR_LIT:_>' , '<STR_LIT:U+0020>' ) <EOL> joiner = '<STR_LIT:=>' <EOL> if opt in ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) : <EOL> joiner = '<STR_LIT:U+0020>' <EOL> table_opts . append ( joiner . join ( ( opt , arg ) ) ) <EOL> return '<STR_LIT:U+0020>' . join ( table_opts ) <EOL> def visit_create_index ( self , create ) : <EOL> index = create . element <EOL> self . _verify_index_table ( index ) <EOL> preparer = self . preparer <EOL> table = preparer . format_table ( index . table ) <EOL> columns = [ self . sql_compiler . process ( expr , include_table = False , <EOL> literal_binds = True ) <EOL> for expr in index . expressions ] <EOL> name = self . _prepared_index_name ( index ) <EOL> text = "<STR_LIT>" <EOL> if index . unique : <EOL> text += "<STR_LIT>" <EOL> text += "<STR_LIT>" % ( name , table ) <EOL> length = index . dialect_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if length is not None : <EOL> if isinstance ( length , dict ) : <EOL> columns = '<STR_LIT:U+002CU+0020>' . join ( <EOL> '<STR_LIT>' % ( expr , length [ col . name ] ) if col . name in length <EOL> else <EOL> ( <EOL> '<STR_LIT>' % ( expr , length [ expr ] ) if expr in length <EOL> else '<STR_LIT:%s>' % expr <EOL> ) <EOL> for col , expr in zip ( index . expressions , columns ) <EOL> ) <EOL> else : <EOL> columns = '<STR_LIT:U+002CU+0020>' . join ( <EOL> '<STR_LIT>' % ( col , length ) <EOL> for col in columns <EOL> ) <EOL> else : <EOL> columns = '<STR_LIT:U+002CU+0020>' . join ( columns ) <EOL> text += '<STR_LIT>' % columns <EOL> using = index . dialect_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if using is not None : <EOL> text += "<STR_LIT>" % ( preparer . quote ( using ) ) <EOL> return text <EOL> def visit_primary_key_constraint ( self , constraint ) : <EOL> text = super ( MySQLDDLCompiler , self ) . visit_primary_key_constraint ( constraint ) <EOL> using = constraint . dialect_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if using : <EOL> text += "<STR_LIT>" % ( self . preparer . quote ( using ) ) <EOL> return text <EOL> def visit_drop_index ( self , drop ) : <EOL> index = drop . element <EOL> return "<STR_LIT>" % ( <EOL> self . _prepared_index_name ( index , <EOL> include_schema = False ) , <EOL> self . preparer . format_table ( index . table ) ) <EOL> def visit_drop_constraint ( self , drop ) : <EOL> constraint = drop . element <EOL> if isinstance ( constraint , sa_schema . ForeignKeyConstraint ) : <EOL> qual = "<STR_LIT>" <EOL> const = self . preparer . format_constraint ( constraint ) <EOL> elif isinstance ( constraint , sa_schema . PrimaryKeyConstraint ) : <EOL> qual = "<STR_LIT>" <EOL> const = "<STR_LIT>" <EOL> elif isinstance ( constraint , sa_schema . UniqueConstraint ) : <EOL> qual = "<STR_LIT>" <EOL> const = self . preparer . format_constraint ( constraint ) <EOL> else : <EOL> qual = "<STR_LIT>" <EOL> const = self . preparer . format_constraint ( constraint ) <EOL> return "<STR_LIT>" % ( self . preparer . format_table ( constraint . table ) , <EOL> qual , const ) <EOL> def define_constraint_match ( self , constraint ) : <EOL> if constraint . match is not None : <EOL> raise exc . CompileError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return "<STR_LIT>" <EOL> class MySQLTypeCompiler ( compiler . GenericTypeCompiler ) : <EOL> def _extend_numeric ( self , type_ , spec ) : <EOL> "<STR_LIT>" <EOL> if not self . _mysql_type ( type_ ) : <EOL> return spec <EOL> if type_ . unsigned : <EOL> spec += '<STR_LIT>' <EOL> if type_ . zerofill : <EOL> spec += '<STR_LIT>' <EOL> return spec <EOL> def _extend_string ( self , type_ , defaults , spec ) : <EOL> """<STR_LIT>""" <EOL> def attr ( name ) : <EOL> return getattr ( type_ , name , defaults . get ( name ) ) <EOL> if attr ( '<STR_LIT>' ) : <EOL> charset = '<STR_LIT>' % attr ( '<STR_LIT>' ) <EOL> elif attr ( '<STR_LIT:ascii>' ) : <EOL> charset = '<STR_LIT>' <EOL> elif attr ( '<STR_LIT>' ) : <EOL> charset = '<STR_LIT>' <EOL> else : <EOL> charset = None <EOL> if attr ( '<STR_LIT>' ) : <EOL> collation = '<STR_LIT>' % type_ . collation <EOL> elif attr ( '<STR_LIT>' ) : <EOL> collation = '<STR_LIT>' <EOL> else : <EOL> collation = None <EOL> if attr ( '<STR_LIT>' ) : <EOL> return '<STR_LIT:U+0020>' . join ( [ c for c in ( '<STR_LIT>' , spec , collation ) <EOL> if c is not None ] ) <EOL> return '<STR_LIT:U+0020>' . join ( [ c for c in ( spec , charset , collation ) <EOL> if c is not None ] ) <EOL> def _mysql_type ( self , type_ ) : <EOL> return isinstance ( type_ , ( _StringType , _NumericType ) ) <EOL> def visit_NUMERIC ( self , type_ , ** kw ) : <EOL> if type_ . precision is None : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> elif type_ . scale is None : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . precision } ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . precision , <EOL> '<STR_LIT>' : type_ . scale } ) <EOL> def visit_DECIMAL ( self , type_ , ** kw ) : <EOL> if type_ . precision is None : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> elif type_ . scale is None : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . precision } ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . precision , <EOL> '<STR_LIT>' : type_ . scale } ) <EOL> def visit_DOUBLE ( self , type_ , ** kw ) : <EOL> if type_ . precision is not None and type_ . scale is not None : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . precision , <EOL> '<STR_LIT>' : type_ . scale } ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , '<STR_LIT>' ) <EOL> def visit_REAL ( self , type_ , ** kw ) : <EOL> if type_ . precision is not None and type_ . scale is not None : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . precision , <EOL> '<STR_LIT>' : type_ . scale } ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , '<STR_LIT>' ) <EOL> def visit_FLOAT ( self , type_ , ** kw ) : <EOL> if self . _mysql_type ( type_ ) and type_ . scale is not None and type_ . precision is not None : <EOL> return self . _extend_numeric ( <EOL> type_ , "<STR_LIT>" % ( type_ . precision , type_ . scale ) ) <EOL> elif type_ . precision is not None : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % ( type_ . precision , ) ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> def visit_INTEGER ( self , type_ , ** kw ) : <EOL> if self . _mysql_type ( type_ ) and type_ . display_width is not None : <EOL> return self . _extend_numeric ( <EOL> type_ , "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . display_width } ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> def visit_BIGINT ( self , type_ , ** kw ) : <EOL> if self . _mysql_type ( type_ ) and type_ . display_width is not None : <EOL> return self . _extend_numeric ( <EOL> type_ , "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . display_width } ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> def visit_MEDIUMINT ( self , type_ , ** kw ) : <EOL> if self . _mysql_type ( type_ ) and type_ . display_width is not None : <EOL> return self . _extend_numeric ( <EOL> type_ , "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . display_width } ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> def visit_TINYINT ( self , type_ , ** kw ) : <EOL> if self . _mysql_type ( type_ ) and type_ . display_width is not None : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % type_ . display_width ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> def visit_SMALLINT ( self , type_ , ** kw ) : <EOL> if self . _mysql_type ( type_ ) and type_ . display_width is not None : <EOL> return self . _extend_numeric ( type_ , <EOL> "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . display_width } <EOL> ) <EOL> else : <EOL> return self . _extend_numeric ( type_ , "<STR_LIT>" ) <EOL> def visit_BIT ( self , type_ , ** kw ) : <EOL> if type_ . length is not None : <EOL> return "<STR_LIT>" % type_ . length <EOL> else : <EOL> return "<STR_LIT>" <EOL> def visit_DATETIME ( self , type_ , ** kw ) : <EOL> if getattr ( type_ , '<STR_LIT>' , None ) : <EOL> return "<STR_LIT>" % type_ . fsp <EOL> else : <EOL> return "<STR_LIT>" <EOL> def visit_DATE ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_TIME ( self , type_ , ** kw ) : <EOL> if getattr ( type_ , '<STR_LIT>' , None ) : <EOL> return "<STR_LIT>" % type_ . fsp <EOL> else : <EOL> return "<STR_LIT>" <EOL> def visit_TIMESTAMP ( self , type_ , ** kw ) : <EOL> if getattr ( type_ , '<STR_LIT>' , None ) : <EOL> return "<STR_LIT>" % type_ . fsp <EOL> else : <EOL> return "<STR_LIT>" <EOL> def visit_YEAR ( self , type_ , ** kw ) : <EOL> if type_ . display_width is None : <EOL> return "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" % type_ . display_width <EOL> def visit_TEXT ( self , type_ , ** kw ) : <EOL> if type_ . length : <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" % type_ . length ) <EOL> else : <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" ) <EOL> def visit_TINYTEXT ( self , type_ , ** kw ) : <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" ) <EOL> def visit_MEDIUMTEXT ( self , type_ , ** kw ) : <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" ) <EOL> def visit_LONGTEXT ( self , type_ , ** kw ) : <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" ) <EOL> def visit_VARCHAR ( self , type_ , ** kw ) : <EOL> if type_ . length : <EOL> return self . _extend_string ( <EOL> type_ , { } , "<STR_LIT>" % type_ . length ) <EOL> else : <EOL> raise exc . CompileError ( <EOL> "<STR_LIT>" % <EOL> self . dialect . name ) <EOL> def visit_CHAR ( self , type_ , ** kw ) : <EOL> if type_ . length : <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" % <EOL> { '<STR_LIT>' : type_ . length } ) <EOL> else : <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" ) <EOL> def visit_NVARCHAR ( self , type_ , ** kw ) : <EOL> if type_ . length : <EOL> return self . _extend_string ( <EOL> type_ , { '<STR_LIT>' : True } , <EOL> "<STR_LIT>" % { '<STR_LIT>' : type_ . length } ) <EOL> else : <EOL> raise exc . CompileError ( <EOL> "<STR_LIT>" % <EOL> self . dialect . name ) <EOL> def visit_NCHAR ( self , type_ , ** kw ) : <EOL> if type_ . length : <EOL> return self . _extend_string ( <EOL> type_ , { '<STR_LIT>' : True } , <EOL> "<STR_LIT>" % { '<STR_LIT>' : type_ . length } ) <EOL> else : <EOL> return self . _extend_string ( type_ , { '<STR_LIT>' : True } , "<STR_LIT>" ) <EOL> def visit_VARBINARY ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" % type_ . length <EOL> def visit_JSON ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_large_binary ( self , type_ , ** kw ) : <EOL> return self . visit_BLOB ( type_ ) <EOL> def visit_enum ( self , type_ , ** kw ) : <EOL> if not type_ . native_enum : <EOL> return super ( MySQLTypeCompiler , self ) . visit_enum ( type_ ) <EOL> else : <EOL> return self . _visit_enumerated_values ( "<STR_LIT>" , type_ , type_ . enums ) <EOL> def visit_BLOB ( self , type_ , ** kw ) : <EOL> if type_ . length : <EOL> return "<STR_LIT>" % type_ . length <EOL> else : <EOL> return "<STR_LIT>" <EOL> def visit_TINYBLOB ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_MEDIUMBLOB ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def visit_LONGBLOB ( self , type_ , ** kw ) : <EOL> return "<STR_LIT>" <EOL> def _visit_enumerated_values ( self , name , type_ , enumerated_values ) : <EOL> quoted_enums = [ ] <EOL> for e in enumerated_values : <EOL> quoted_enums . append ( "<STR_LIT>" % e . replace ( "<STR_LIT:'>" , "<STR_LIT>" ) ) <EOL> return self . _extend_string ( type_ , { } , "<STR_LIT>" % ( <EOL> name , "<STR_LIT:U+002C>" . join ( quoted_enums ) ) <EOL> ) <EOL> def visit_ENUM ( self , type_ , ** kw ) : <EOL> return self . _visit_enumerated_values ( "<STR_LIT>" , type_ , <EOL> type_ . _enumerated_values ) <EOL> def visit_SET ( self , type_ , ** kw ) : <EOL> return self . _visit_enumerated_values ( "<STR_LIT>" , type_ , <EOL> type_ . _enumerated_values ) <EOL> def visit_BOOLEAN ( self , type , ** kw ) : <EOL> return "<STR_LIT>" <EOL> class MySQLIdentifierPreparer ( compiler . IdentifierPreparer ) : <EOL> reserved_words = RESERVED_WORDS <EOL> def __init__ ( self , dialect , server_ansiquotes = False , ** kw ) : <EOL> if not server_ansiquotes : <EOL> quote = "<STR_LIT>" <EOL> else : <EOL> quote = '<STR_LIT:">' <EOL> super ( MySQLIdentifierPreparer , self ) . __init__ ( <EOL> dialect , <EOL> initial_quote = quote , <EOL> escape_quote = quote ) <EOL> def _quote_free_identifiers ( self , * ids ) : <EOL> """<STR_LIT>""" <EOL> return tuple ( [ self . quote_identifier ( i ) for i in ids if i is not None ] ) <EOL> @ log . class_logger <EOL> class MySQLDialect ( default . DefaultDialect ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> supports_alter = True <EOL> supports_native_boolean = False <EOL> max_identifier_length = <NUM_LIT:255> <EOL> max_index_name_length = <NUM_LIT:64> <EOL> supports_native_enum = True <EOL> supports_sane_rowcount = True <EOL> supports_sane_multi_rowcount = False <EOL> supports_multivalues_insert = True <EOL> default_paramstyle = '<STR_LIT>' <EOL> colspecs = colspecs <EOL> statement_compiler = MySQLCompiler <EOL> ddl_compiler = MySQLDDLCompiler <EOL> type_compiler = MySQLTypeCompiler <EOL> ischema_names = ischema_names <EOL> preparer = MySQLIdentifierPreparer <EOL> _backslash_escapes = True <EOL> _server_ansiquotes = False <EOL> construct_arguments = [ <EOL> ( sa_schema . Table , { <EOL> "<STR_LIT:*>" : None <EOL> } ) , <EOL> ( sql . Update , { <EOL> "<STR_LIT>" : None <EOL> } ) , <EOL> ( sa_schema . PrimaryKeyConstraint , { <EOL> "<STR_LIT>" : None <EOL> } ) , <EOL> ( sa_schema . Index , { <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> } ) <EOL> ] <EOL> def __init__ ( self , isolation_level = None , json_serializer = None , <EOL> json_deserializer = None , ** kwargs ) : <EOL> kwargs . pop ( '<STR_LIT>' , None ) <EOL> default . DefaultDialect . __init__ ( self , ** kwargs ) <EOL> self . isolation_level = isolation_level <EOL> self . _json_serializer = json_serializer <EOL> self . _json_deserializer = json_deserializer <EOL> def on_connect ( self ) : <EOL> if self . isolation_level is not None : <EOL> def connect ( conn ) : <EOL> self . set_isolation_level ( conn , self . isolation_level ) <EOL> return connect <EOL> else : <EOL> return None <EOL> _isolation_lookup = set ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def set_isolation_level ( self , connection , level ) : <EOL> level = level . replace ( '<STR_LIT:_>' , '<STR_LIT:U+0020>' ) <EOL> if hasattr ( connection , '<STR_LIT>' ) : <EOL> connection = connection . connection <EOL> self . _set_isolation_level ( connection , level ) <EOL> def _set_isolation_level ( self , connection , level ) : <EOL> if level not in self . _isolation_lookup : <EOL> raise exc . ArgumentError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( level , self . name , "<STR_LIT:U+002CU+0020>" . join ( self . _isolation_lookup ) ) <EOL> ) <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( "<STR_LIT>" % level ) <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> cursor . close ( ) <EOL> def get_isolation_level ( self , connection ) : <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> val = cursor . fetchone ( ) [ <NUM_LIT:0> ] <EOL> cursor . close ( ) <EOL> if util . py3k and isinstance ( val , bytes ) : <EOL> val = val . decode ( ) <EOL> return val . upper ( ) . replace ( "<STR_LIT:->" , "<STR_LIT:U+0020>" ) <EOL> def do_commit ( self , dbapi_connection ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> dbapi_connection . commit ( ) <EOL> except Exception : <EOL> if self . server_version_info < ( <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:15> ) : <EOL> args = sys . exc_info ( ) [ <NUM_LIT:1> ] . args <EOL> if args and args [ <NUM_LIT:0> ] == <NUM_LIT> : <EOL> return <EOL> raise <EOL> def do_rollback ( self , dbapi_connection ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> dbapi_connection . rollback ( ) <EOL> except Exception : <EOL> if self . server_version_info < ( <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:15> ) : <EOL> args = sys . exc_info ( ) [ <NUM_LIT:1> ] . args <EOL> if args and args [ <NUM_LIT:0> ] == <NUM_LIT> : <EOL> return <EOL> raise <EOL> def do_begin_twophase ( self , connection , xid ) : <EOL> connection . execute ( sql . text ( "<STR_LIT>" ) , xid = xid ) <EOL> def do_prepare_twophase ( self , connection , xid ) : <EOL> connection . execute ( sql . text ( "<STR_LIT>" ) , xid = xid ) <EOL> connection . execute ( sql . text ( "<STR_LIT>" ) , xid = xid ) <EOL> def do_rollback_twophase ( self , connection , xid , is_prepared = True , <EOL> recover = False ) : <EOL> if not is_prepared : <EOL> connection . execute ( sql . text ( "<STR_LIT>" ) , xid = xid ) <EOL> connection . execute ( sql . text ( "<STR_LIT>" ) , xid = xid ) <EOL> def do_commit_twophase ( self , connection , xid , is_prepared = True , <EOL> recover = False ) : <EOL> if not is_prepared : <EOL> self . do_prepare_twophase ( connection , xid ) <EOL> connection . execute ( sql . text ( "<STR_LIT>" ) , xid = xid ) <EOL> def do_recover_twophase ( self , connection ) : <EOL> resultset = connection . execute ( "<STR_LIT>" ) <EOL> return [ row [ '<STR_LIT:data>' ] [ <NUM_LIT:0> : row [ '<STR_LIT>' ] ] for row in resultset ] <EOL> def is_disconnect ( self , e , connection , cursor ) : <EOL> if isinstance ( e , ( self . dbapi . OperationalError , <EOL> self . dbapi . ProgrammingError ) ) : <EOL> return self . _extract_error_code ( e ) in ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> elif isinstance ( e , self . dbapi . InterfaceError ) : <EOL> return "<STR_LIT>" in str ( e ) <EOL> else : <EOL> return False <EOL> def _compat_fetchall ( self , rp , charset = None ) : <EOL> """<STR_LIT>""" <EOL> return [ _DecodingRowProxy ( row , charset ) for row in rp . fetchall ( ) ] <EOL> def _compat_fetchone ( self , rp , charset = None ) : <EOL> """<STR_LIT>""" <EOL> return _DecodingRowProxy ( rp . fetchone ( ) , charset ) <EOL> def _compat_first ( self , rp , charset = None ) : <EOL> """<STR_LIT>""" <EOL> return _DecodingRowProxy ( rp . first ( ) , charset ) <EOL> def _extract_error_code ( self , exception ) : <EOL> raise NotImplementedError ( ) <EOL> def _get_default_schema_name ( self , connection ) : <EOL> return connection . execute ( '<STR_LIT>' ) . scalar ( ) <EOL> def has_table ( self , connection , table_name , schema = None ) : <EOL> full_name = '<STR_LIT:.>' . join ( self . identifier_preparer . _quote_free_identifiers ( <EOL> schema , table_name ) ) <EOL> st = "<STR_LIT>" % full_name <EOL> rs = None <EOL> try : <EOL> try : <EOL> rs = connection . execution_options ( <EOL> skip_user_error_events = True ) . execute ( st ) <EOL> have = rs . fetchone ( ) is not None <EOL> rs . close ( ) <EOL> return have <EOL> except exc . DBAPIError as e : <EOL> if self . _extract_error_code ( e . orig ) == <NUM_LIT> : <EOL> return False <EOL> raise <EOL> finally : <EOL> if rs : <EOL> rs . close ( ) <EOL> def initialize ( self , connection ) : <EOL> self . _connection_charset = self . _detect_charset ( connection ) <EOL> self . _detect_ansiquotes ( connection ) <EOL> if self . _server_ansiquotes : <EOL> self . identifier_preparer = self . preparer ( <EOL> self , server_ansiquotes = self . _server_ansiquotes ) <EOL> default . DefaultDialect . initialize ( self , connection ) <EOL> @ property <EOL> def _is_mariadb ( self ) : <EOL> return '<STR_LIT>' in self . server_version_info <EOL> @ property <EOL> def _supports_cast ( self ) : <EOL> return self . server_version_info is None or self . server_version_info >= ( <NUM_LIT:4> , <NUM_LIT:0> , <NUM_LIT:2> ) <EOL> @ reflection . cache <EOL> def get_schema_names ( self , connection , ** kw ) : <EOL> rp = connection . execute ( "<STR_LIT>" ) <EOL> return [ r [ <NUM_LIT:0> ] for r in rp ] <EOL> @ reflection . cache <EOL> def get_table_names ( self , connection , schema = None , ** kw ) : <EOL> """<STR_LIT>""" <EOL> if schema is not None : <EOL> current_schema = schema <EOL> else : <EOL> current_schema = self . default_schema_name <EOL> charset = self . _connection_charset <EOL> if self . server_version_info < ( <NUM_LIT:5> , <NUM_LIT:0> , <NUM_LIT:2> ) : <EOL> rp = connection . execute ( <EOL> "<STR_LIT>" % <EOL> self . identifier_preparer . quote_identifier ( current_schema ) ) <EOL> return [ row [ <NUM_LIT:0> ] for <EOL> row in self . _compat_fetchall ( rp , charset = charset ) ] <EOL> else : <EOL> rp = connection . execute ( <EOL> "<STR_LIT>" % <EOL> self . identifier_preparer . quote_identifier ( current_schema ) ) <EOL> return [ row [ <NUM_LIT:0> ] <EOL> for row in self . _compat_fetchall ( rp , charset = charset ) <EOL> if row [ <NUM_LIT:1> ] == '<STR_LIT>' ] <EOL> @ reflection . cache <EOL> def get_view_names ( self , connection , schema = None , ** kw ) : <EOL> if self . server_version_info < ( <NUM_LIT:5> , <NUM_LIT:0> , <NUM_LIT:2> ) : <EOL> raise NotImplementedError <EOL> if schema is None : <EOL> schema = self . default_schema_name <EOL> if self . server_version_info < ( <NUM_LIT:5> , <NUM_LIT:0> , <NUM_LIT:2> ) : <EOL> return self . get_table_names ( connection , schema ) <EOL> charset = self . _connection_charset <EOL> rp = connection . execute ( <EOL> "<STR_LIT>" % <EOL> self . identifier_preparer . quote_identifier ( schema ) ) <EOL> return [ row [ <NUM_LIT:0> ] <EOL> for row in self . _compat_fetchall ( rp , charset = charset ) <EOL> if row [ <NUM_LIT:1> ] in ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> @ reflection . cache <EOL> def get_table_options ( self , connection , table_name , schema = None , ** kw ) : <EOL> parsed_state = self . _parsed_state_or_create ( <EOL> connection , table_name , schema , ** kw ) <EOL> return parsed_state . table_options <EOL> @ reflection . cache <EOL> def get_columns ( self , connection , table_name , schema = None , ** kw ) : <EOL> parsed_state = self . _parsed_state_or_create ( <EOL> connection , table_name , schema , ** kw ) <EOL> return parsed_state . columns <EOL> @ reflection . cache <EOL> def get_pk_constraint ( self , connection , table_name , schema = None , ** kw ) : <EOL> parsed_state = self . _parsed_state_or_create ( <EOL> connection , table_name , schema , ** kw ) <EOL> for key in parsed_state . keys : <EOL> if key [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> cols = [ s [ <NUM_LIT:0> ] for s in key [ '<STR_LIT>' ] ] <EOL> return { '<STR_LIT>' : cols , '<STR_LIT:name>' : None } <EOL> return { '<STR_LIT>' : [ ] , '<STR_LIT:name>' : None } <EOL> @ reflection . cache <EOL> def get_foreign_keys ( self , connection , table_name , schema = None , ** kw ) : <EOL> parsed_state = self . _parsed_state_or_create ( <EOL> connection , table_name , schema , ** kw ) <EOL> default_schema = None <EOL> fkeys = [ ] <EOL> for spec in parsed_state . constraints : <EOL> ref_name = spec [ '<STR_LIT>' ] [ - <NUM_LIT:1> ] <EOL> ref_schema = len ( spec [ '<STR_LIT>' ] ) > <NUM_LIT:1> and spec [ '<STR_LIT>' ] [ - <NUM_LIT:2> ] or schema <EOL> if not ref_schema : <EOL> if default_schema is None : <EOL> default_schema = connection . dialect . default_schema_name <EOL> if schema == default_schema : <EOL> ref_schema = schema <EOL> loc_names = spec [ '<STR_LIT>' ] <EOL> ref_names = spec [ '<STR_LIT>' ] <EOL> con_kw = { } <EOL> for opt in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if spec . get ( opt , False ) : <EOL> con_kw [ opt ] = spec [ opt ] <EOL> fkey_d = { <EOL> '<STR_LIT:name>' : spec [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT>' : loc_names , <EOL> '<STR_LIT>' : ref_schema , <EOL> '<STR_LIT>' : ref_name , <EOL> '<STR_LIT>' : ref_names , <EOL> '<STR_LIT>' : con_kw <EOL> } <EOL> fkeys . append ( fkey_d ) <EOL> return fkeys <EOL> @ reflection . cache <EOL> def get_indexes ( self , connection , table_name , schema = None , ** kw ) : <EOL> parsed_state = self . _parsed_state_or_create ( <EOL> connection , table_name , schema , ** kw ) <EOL> indexes = [ ] <EOL> for spec in parsed_state . keys : <EOL> unique = False <EOL> flavor = spec [ '<STR_LIT:type>' ] <EOL> if flavor == '<STR_LIT>' : <EOL> continue <EOL> if flavor == '<STR_LIT>' : <EOL> unique = True <EOL> elif flavor in ( None , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> pass <EOL> else : <EOL> self . logger . info ( <EOL> "<STR_LIT>" , flavor ) <EOL> pass <EOL> index_d = { } <EOL> index_d [ '<STR_LIT:name>' ] = spec [ '<STR_LIT:name>' ] <EOL> index_d [ '<STR_LIT>' ] = [ s [ <NUM_LIT:0> ] for s in spec [ '<STR_LIT>' ] ] <EOL> index_d [ '<STR_LIT>' ] = unique <EOL> if flavor : <EOL> index_d [ '<STR_LIT:type>' ] = flavor <EOL> indexes . append ( index_d ) <EOL> return indexes <EOL> @ reflection . cache <EOL> def get_unique_constraints ( self , connection , table_name , <EOL> schema = None , ** kw ) : <EOL> parsed_state = self . _parsed_state_or_create ( <EOL> connection , table_name , schema , ** kw ) <EOL> return [ <EOL> { <EOL> '<STR_LIT:name>' : key [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT>' : [ col [ <NUM_LIT:0> ] for col in key [ '<STR_LIT>' ] ] , <EOL> '<STR_LIT>' : key [ '<STR_LIT:name>' ] , <EOL> } <EOL> for key in parsed_state . keys <EOL> if key [ '<STR_LIT:type>' ] == '<STR_LIT>' <EOL> ] <EOL> @ reflection . cache <EOL> def get_view_definition ( self , connection , view_name , schema = None , ** kw ) : <EOL> charset = self . _connection_charset <EOL> full_name = '<STR_LIT:.>' . join ( self . identifier_preparer . _quote_free_identifiers ( <EOL> schema , view_name ) ) <EOL> sql = self . _show_create_table ( connection , None , charset , <EOL> full_name = full_name ) <EOL> return sql <EOL> def _parsed_state_or_create ( self , connection , table_name , <EOL> schema = None , ** kw ) : <EOL> return self . _setup_parser ( <EOL> connection , <EOL> table_name , <EOL> schema , <EOL> info_cache = kw . get ( '<STR_LIT>' , None ) <EOL> ) <EOL> @ util . memoized_property <EOL> def _tabledef_parser ( self ) : <EOL> """<STR_LIT>""" <EOL> if ( self . server_version_info < ( <NUM_LIT:4> , <NUM_LIT:1> ) and self . _server_ansiquotes ) : <EOL> preparer = self . preparer ( self , server_ansiquotes = False ) <EOL> else : <EOL> preparer = self . identifier_preparer <EOL> return _reflection . MySQLTableDefinitionParser ( self , preparer ) <EOL> @ reflection . cache <EOL> def _setup_parser ( self , connection , table_name , schema = None , ** kw ) : <EOL> charset = self . _connection_charset <EOL> parser = self . _tabledef_parser <EOL> full_name = '<STR_LIT:.>' . join ( self . identifier_preparer . _quote_free_identifiers ( <EOL> schema , table_name ) ) <EOL> sql = self . _show_create_table ( connection , None , charset , <EOL> full_name = full_name ) <EOL> if re . match ( r'<STR_LIT>' , sql ) : <EOL> columns = self . _describe_table ( connection , None , charset , <EOL> full_name = full_name ) <EOL> sql = parser . _describe_to_create ( table_name , columns ) <EOL> return parser . parse ( sql , charset ) <EOL> def _detect_charset ( self , connection ) : <EOL> raise NotImplementedError ( ) <EOL> def _detect_casing ( self , connection ) : <EOL> """<STR_LIT>""" <EOL> charset = self . _connection_charset <EOL> row = self . _compat_first ( connection . execute ( <EOL> "<STR_LIT>" ) , <EOL> charset = charset ) <EOL> if not row : <EOL> cs = <NUM_LIT:0> <EOL> else : <EOL> if row [ <NUM_LIT:1> ] == '<STR_LIT>' : <EOL> cs = <NUM_LIT:0> <EOL> elif row [ <NUM_LIT:1> ] == '<STR_LIT>' : <EOL> cs = <NUM_LIT:1> <EOL> else : <EOL> cs = int ( row [ <NUM_LIT:1> ] ) <EOL> return cs <EOL> def _detect_collations ( self , connection ) : <EOL> """<STR_LIT>""" <EOL> collations = { } <EOL> if self . server_version_info < ( <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT:0> ) : <EOL> pass <EOL> else : <EOL> charset = self . _connection_charset <EOL> rs = connection . execute ( '<STR_LIT>' ) <EOL> for row in self . _compat_fetchall ( rs , charset ) : <EOL> collations [ row [ <NUM_LIT:0> ] ] = row [ <NUM_LIT:1> ] <EOL> return collations <EOL> def _detect_ansiquotes ( self , connection ) : <EOL> """<STR_LIT>""" <EOL> row = self . _compat_first ( <EOL> connection . execute ( "<STR_LIT>" ) , <EOL> charset = self . _connection_charset ) <EOL> if not row : <EOL> mode = '<STR_LIT>' <EOL> else : <EOL> mode = row [ <NUM_LIT:1> ] or '<STR_LIT>' <EOL> if mode . isdigit ( ) : <EOL> mode_no = int ( mode ) <EOL> mode = ( mode_no | <NUM_LIT:4> == mode_no ) and '<STR_LIT>' or '<STR_LIT>' <EOL> self . _server_ansiquotes = '<STR_LIT>' in mode <EOL> self . _backslash_escapes = '<STR_LIT>' not in mode <EOL> def _show_create_table ( self , connection , table , charset = None , <EOL> full_name = None ) : <EOL> """<STR_LIT>""" <EOL> if full_name is None : <EOL> full_name = self . identifier_preparer . format_table ( table ) <EOL> st = "<STR_LIT>" % full_name <EOL> rp = None <EOL> try : <EOL> rp = connection . execution_options ( <EOL> skip_user_error_events = True ) . execute ( st ) <EOL> except exc . DBAPIError as e : <EOL> if self . _extract_error_code ( e . orig ) == <NUM_LIT> : <EOL> raise exc . NoSuchTableError ( full_name ) <EOL> else : <EOL> raise <EOL> row = self . _compat_first ( rp , charset = charset ) <EOL> if not row : <EOL> raise exc . NoSuchTableError ( full_name ) <EOL> return row [ <NUM_LIT:1> ] . strip ( ) <EOL> return sql <EOL> def _describe_table ( self , connection , table , charset = None , <EOL> full_name = None ) : <EOL> """<STR_LIT>""" <EOL> if full_name is None : <EOL> full_name = self . identifier_preparer . format_table ( table ) <EOL> st = "<STR_LIT>" % full_name <EOL> rp , rows = None , None <EOL> try : <EOL> try : <EOL> rp = connection . execution_options ( <EOL> skip_user_error_events = True ) . execute ( st ) <EOL> except exc . DBAPIError as e : <EOL> if self . _extract_error_code ( e . orig ) == <NUM_LIT> : <EOL> raise exc . NoSuchTableError ( full_name ) <EOL> else : <EOL> raise <EOL> rows = self . _compat_fetchall ( rp , charset = charset ) <EOL> finally : <EOL> if rp : <EOL> rp . close ( ) <EOL> return rows <EOL> class _DecodingRowProxy ( object ) : <EOL> """<STR_LIT>""" <EOL> _encoding_compat = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def __init__ ( self , rowproxy , charset ) : <EOL> self . rowproxy = rowproxy <EOL> self . charset = self . _encoding_compat . get ( charset , charset ) <EOL> def __getitem__ ( self , index ) : <EOL> item = self . rowproxy [ index ] <EOL> if isinstance ( item , _array ) : <EOL> item = item . tostring ( ) <EOL> if self . charset and isinstance ( item , util . binary_type ) : <EOL> return item . decode ( self . charset ) <EOL> else : <EOL> return item <EOL> def __getattr__ ( self , attr ) : <EOL> item = getattr ( self . rowproxy , attr ) <EOL> if isinstance ( item , _array ) : <EOL> item = item . tostring ( ) <EOL> if self . charset and isinstance ( item , util . binary_type ) : <EOL> return item . decode ( self . charset ) <EOL> else : <EOL> return item </s>
