<s> """<STR_LIT>""" <EOL> import re <EOL> from . base import ( BIT , MySQLDialect , MySQLExecutionContext ) <EOL> from ... import types as sqltypes , util <EOL> class _oursqlBIT ( BIT ) : <EOL> def result_processor ( self , dialect , coltype ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> class MySQLExecutionContext_oursql ( MySQLExecutionContext ) : <EOL> @ property <EOL> def plain_query ( self ) : <EOL> return self . execution_options . get ( '<STR_LIT>' , False ) <EOL> class MySQLDialect_oursql ( MySQLDialect ) : <EOL> driver = '<STR_LIT>' <EOL> if util . py2k : <EOL> supports_unicode_binds = True <EOL> supports_unicode_statements = True <EOL> supports_native_decimal = True <EOL> supports_sane_rowcount = True <EOL> supports_sane_multi_rowcount = True <EOL> execution_ctx_cls = MySQLExecutionContext_oursql <EOL> colspecs = util . update_copy ( <EOL> MySQLDialect . colspecs , <EOL> { <EOL> sqltypes . Time : sqltypes . Time , <EOL> BIT : _oursqlBIT , <EOL> } <EOL> ) <EOL> @ classmethod <EOL> def dbapi ( cls ) : <EOL> return __import__ ( '<STR_LIT>' ) <EOL> def do_execute ( self , cursor , statement , parameters , context = None ) : <EOL> """<STR_LIT>""" <EOL> if context and context . plain_query : <EOL> cursor . execute ( statement , plain_query = True ) <EOL> else : <EOL> cursor . execute ( statement , parameters ) <EOL> def do_begin ( self , connection ) : <EOL> connection . cursor ( ) . execute ( '<STR_LIT>' , plain_query = True ) <EOL> def _xa_query ( self , connection , query , xid ) : <EOL> if util . py2k : <EOL> arg = connection . connection . _escape_string ( xid ) <EOL> else : <EOL> charset = self . _connection_charset <EOL> arg = connection . connection . _escape_string ( <EOL> xid . encode ( charset ) ) . decode ( charset ) <EOL> arg = "<STR_LIT>" % arg <EOL> connection . execution_options ( <EOL> _oursql_plain_query = True ) . execute ( query % arg ) <EOL> def do_begin_twophase ( self , connection , xid ) : <EOL> self . _xa_query ( connection , '<STR_LIT>' , xid ) <EOL> def do_prepare_twophase ( self , connection , xid ) : <EOL> self . _xa_query ( connection , '<STR_LIT>' , xid ) <EOL> self . _xa_query ( connection , '<STR_LIT>' , xid ) <EOL> def do_rollback_twophase ( self , connection , xid , is_prepared = True , <EOL> recover = False ) : <EOL> if not is_prepared : <EOL> self . _xa_query ( connection , '<STR_LIT>' , xid ) <EOL> self . _xa_query ( connection , '<STR_LIT>' , 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> self . _xa_query ( connection , '<STR_LIT>' , xid ) <EOL> def has_table ( self , connection , table_name , schema = None ) : <EOL> return MySQLDialect . has_table ( <EOL> self , <EOL> connection . connect ( ) . execution_options ( _oursql_plain_query = True ) , <EOL> table_name , <EOL> schema <EOL> ) <EOL> def get_table_options ( self , connection , table_name , schema = None , ** kw ) : <EOL> return MySQLDialect . get_table_options ( <EOL> self , <EOL> connection . connect ( ) . execution_options ( _oursql_plain_query = True ) , <EOL> table_name , <EOL> schema = schema , <EOL> ** kw <EOL> ) <EOL> def get_columns ( self , connection , table_name , schema = None , ** kw ) : <EOL> return MySQLDialect . get_columns ( <EOL> self , <EOL> connection . connect ( ) . execution_options ( _oursql_plain_query = True ) , <EOL> table_name , <EOL> schema = schema , <EOL> ** kw <EOL> ) <EOL> def get_view_names ( self , connection , schema = None , ** kw ) : <EOL> return MySQLDialect . get_view_names ( <EOL> self , <EOL> connection . connect ( ) . execution_options ( _oursql_plain_query = True ) , <EOL> schema = schema , <EOL> ** kw <EOL> ) <EOL> def get_table_names ( self , connection , schema = None , ** kw ) : <EOL> return MySQLDialect . get_table_names ( <EOL> self , <EOL> connection . connect ( ) . execution_options ( _oursql_plain_query = True ) , <EOL> schema <EOL> ) <EOL> def get_schema_names ( self , connection , ** kw ) : <EOL> return MySQLDialect . get_schema_names ( <EOL> self , <EOL> connection . connect ( ) . execution_options ( _oursql_plain_query = True ) , <EOL> ** kw <EOL> ) <EOL> def initialize ( self , connection ) : <EOL> return MySQLDialect . initialize ( <EOL> self , <EOL> connection . execution_options ( _oursql_plain_query = True ) <EOL> ) <EOL> def _show_create_table ( self , connection , table , charset = None , <EOL> full_name = None ) : <EOL> return MySQLDialect . _show_create_table ( <EOL> self , <EOL> connection . contextual_connect ( close_with_result = True ) . <EOL> execution_options ( _oursql_plain_query = True ) , <EOL> table , charset , full_name <EOL> ) <EOL> def is_disconnect ( self , e , connection , cursor ) : <EOL> if isinstance ( e , self . dbapi . ProgrammingError ) : <EOL> return e . errno is None and '<STR_LIT>' not in e . args [ <NUM_LIT:1> ] and e . args [ <NUM_LIT:1> ] . endswith ( '<STR_LIT>' ) <EOL> else : <EOL> return e . errno in ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> def create_connect_args ( self , url ) : <EOL> opts = url . translate_connect_args ( database = '<STR_LIT>' , username = '<STR_LIT:user>' , <EOL> password = '<STR_LIT>' ) <EOL> opts . update ( url . query ) <EOL> util . coerce_kw_type ( opts , '<STR_LIT:port>' , int ) <EOL> util . coerce_kw_type ( opts , '<STR_LIT>' , bool ) <EOL> util . coerce_kw_type ( opts , '<STR_LIT>' , bool ) <EOL> util . coerce_kw_type ( opts , '<STR_LIT>' , bool ) <EOL> util . coerce_kw_type ( opts , '<STR_LIT>' , bool ) <EOL> if opts . pop ( '<STR_LIT>' , False ) : <EOL> opts [ '<STR_LIT>' ] = None <EOL> else : <EOL> util . coerce_kw_type ( opts , '<STR_LIT>' , str ) <EOL> opts [ '<STR_LIT>' ] = opts . get ( '<STR_LIT>' , True ) <EOL> util . coerce_kw_type ( opts , '<STR_LIT>' , bool ) <EOL> opts . setdefault ( '<STR_LIT>' , True ) <EOL> ssl = { } <EOL> for key in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] : <EOL> if key in opts : <EOL> ssl [ key [ <NUM_LIT:4> : ] ] = opts [ key ] <EOL> util . coerce_kw_type ( ssl , key [ <NUM_LIT:4> : ] , str ) <EOL> del opts [ key ] <EOL> if ssl : <EOL> opts [ '<STR_LIT>' ] = ssl <EOL> return [ [ ] , opts ] <EOL> def _get_server_version_info ( self , connection ) : <EOL> dbapi_con = connection . connection <EOL> version = [ ] <EOL> r = re . compile ( '<STR_LIT>' ) <EOL> for n in r . split ( dbapi_con . server_info ) : <EOL> try : <EOL> version . append ( int ( n ) ) <EOL> except ValueError : <EOL> version . append ( n ) <EOL> return tuple ( version ) <EOL> def _extract_error_code ( self , exception ) : <EOL> return exception . errno <EOL> def _detect_charset ( self , connection ) : <EOL> """<STR_LIT>""" <EOL> return connection . connection . charset <EOL> def _compat_fetchall ( self , rp , charset = None ) : <EOL> """<STR_LIT>""" <EOL> return rp . fetchall ( ) <EOL> def _compat_fetchone ( self , rp , charset = None ) : <EOL> """<STR_LIT>""" <EOL> return rp . fetchone ( ) <EOL> def _compat_first ( self , rp , charset = None ) : <EOL> return rp . first ( ) <EOL> dialect = MySQLDialect_oursql </s>
