<s> """<STR_LIT>""" <EOL> import sys <EOL> import re <EOL> import warnings <EOL> from . import Connector <EOL> class MxODBCConnector ( Connector ) : <EOL> driver = '<STR_LIT>' <EOL> supports_sane_multi_rowcount = False <EOL> supports_unicode_statements = True <EOL> supports_unicode_binds = True <EOL> supports_native_decimal = True <EOL> @ classmethod <EOL> def dbapi ( cls ) : <EOL> cls . _load_mx_exceptions ( ) <EOL> platform = sys . platform <EOL> if platform == '<STR_LIT:win32>' : <EOL> from mx . ODBC import Windows as module <EOL> elif '<STR_LIT>' in platform : <EOL> from mx . ODBC import unixODBC as module <EOL> elif platform == '<STR_LIT>' : <EOL> from mx . ODBC import iODBC as module <EOL> else : <EOL> raise ImportError ( "<STR_LIT>" ) <EOL> return module <EOL> @ classmethod <EOL> def _load_mx_exceptions ( cls ) : <EOL> """<STR_LIT>""" <EOL> global InterfaceError , ProgrammingError <EOL> from mx . ODBC import InterfaceError <EOL> from mx . ODBC import ProgrammingError <EOL> def on_connect ( self ) : <EOL> def connect ( conn ) : <EOL> conn . stringformat = self . dbapi . MIXED_STRINGFORMAT <EOL> conn . datetimeformat = self . dbapi . PYDATETIME_DATETIMEFORMAT <EOL> conn . decimalformat = self . dbapi . DECIMAL_DECIMALFORMAT <EOL> conn . errorhandler = self . _error_handler ( ) <EOL> return connect <EOL> def _error_handler ( self ) : <EOL> """<STR_LIT>""" <EOL> from mx . ODBC . Error import Warning as MxOdbcWarning <EOL> def error_handler ( connection , cursor , errorclass , errorvalue ) : <EOL> if issubclass ( errorclass , MxOdbcWarning ) : <EOL> errorclass . __bases__ = ( Warning , ) <EOL> warnings . warn ( message = str ( errorvalue ) , <EOL> category = errorclass , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> else : <EOL> raise errorclass ( errorvalue ) <EOL> return error_handler <EOL> def create_connect_args ( self , url ) : <EOL> """<STR_LIT>""" <EOL> opts = url . translate_connect_args ( username = '<STR_LIT:user>' ) <EOL> opts . update ( url . query ) <EOL> args = opts . pop ( '<STR_LIT:host>' ) <EOL> opts . pop ( '<STR_LIT:port>' , None ) <EOL> opts . pop ( '<STR_LIT>' , None ) <EOL> return ( args , ) , opts <EOL> def is_disconnect ( self , e , connection , cursor ) : <EOL> if isinstance ( e , self . dbapi . ProgrammingError ) : <EOL> return "<STR_LIT>" in str ( e ) <EOL> elif isinstance ( e , self . dbapi . Error ) : <EOL> return '<STR_LIT>' in str ( e ) <EOL> else : <EOL> return False <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 . getinfo ( <NUM_LIT> ) [ <NUM_LIT:1> ] ) : <EOL> try : <EOL> version . append ( int ( n ) ) <EOL> except ValueError : <EOL> version . append ( n ) <EOL> return tuple ( version ) <EOL> def _get_direct ( self , context ) : <EOL> if context : <EOL> native_odbc_execute = context . execution_options . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> return False if native_odbc_execute is True else True <EOL> else : <EOL> return True <EOL> def do_executemany ( self , cursor , statement , parameters , context = None ) : <EOL> cursor . executemany ( <EOL> statement , parameters , direct = self . _get_direct ( context ) ) <EOL> def do_execute ( self , cursor , statement , parameters , context = None ) : <EOL> cursor . execute ( statement , parameters , direct = self . _get_direct ( context ) ) </s>
