<s> """<STR_LIT>""" <EOL> import datetime <EOL> import os <EOL> import re <EOL> import sys <EOL> import warnings <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> try : <EOL> import pyodbc as Database <EOL> except ImportError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % e ) <EOL> m = re . match ( r'<STR_LIT>' , Database . version ) <EOL> vlist = list ( m . groups ( ) ) <EOL> if vlist [ <NUM_LIT:3> ] is None : vlist [ <NUM_LIT:3> ] = '<STR_LIT>' <EOL> pyodbc_ver = tuple ( map ( int , vlist ) ) <EOL> if pyodbc_ver < ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> ) : <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % Database . version ) <EOL> from django . db import utils <EOL> try : <EOL> from django . db . backends . base . base import BaseDatabaseWrapper <EOL> from django . db . backends . base . features import BaseDatabaseFeatures <EOL> from django . db . backends . base . validation import BaseDatabaseValidation <EOL> except ImportError : <EOL> from django . db . backends import BaseDatabaseWrapper , BaseDatabaseFeatures , BaseDatabaseValidation <EOL> from django . db . backends . signals import connection_created <EOL> from django . conf import settings <EOL> from django import VERSION as DjangoVersion <EOL> if DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:9> ) : <EOL> _DJANGO_VERSION = <NUM_LIT> <EOL> elif DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:8> ) : <EOL> _DJANGO_VERSION = <NUM_LIT> <EOL> elif DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:7> ) : <EOL> _DJANGO_VERSION = <NUM_LIT> <EOL> elif DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:6> ) : <EOL> _DJANGO_VERSION = <NUM_LIT:16> <EOL> elif DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:5> ) : <EOL> _DJANGO_VERSION = <NUM_LIT:15> <EOL> elif DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:4> ) : <EOL> _DJANGO_VERSION = <NUM_LIT> <EOL> elif DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> _DJANGO_VERSION = <NUM_LIT> <EOL> elif DjangoVersion [ : <NUM_LIT:2> ] == ( <NUM_LIT:1> , <NUM_LIT:2> ) : <EOL> _DJANGO_VERSION = <NUM_LIT:12> <EOL> else : <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % DjangoVersion [ : <NUM_LIT:2> ] ) <EOL> from django_pyodbc . operations import DatabaseOperations <EOL> from django_pyodbc . client import DatabaseClient <EOL> from django_pyodbc . compat import binary_type , text_type , timezone <EOL> from django_pyodbc . creation import DatabaseCreation <EOL> from django_pyodbc . introspection import DatabaseIntrospection <EOL> DatabaseError = Database . Error <EOL> IntegrityError = Database . IntegrityError <EOL> class DatabaseFeatures ( BaseDatabaseFeatures ) : <EOL> can_use_chunked_reads = False <EOL> can_return_id_from_insert = True <EOL> supports_microsecond_precision = False <EOL> supports_regex_backreferencing = False <EOL> supports_subqueries_in_group_by = False <EOL> supports_transactions = True <EOL> allow_sliced_subqueries = False <EOL> supports_paramstyle_pyformat = False <EOL> supports_timezones = False <EOL> supports_sequence_reset = False <EOL> supports_tablespaces = True <EOL> ignores_nulls_in_unique_constraints = False <EOL> can_introspect_autofield = True <EOL> def _supports_transactions ( self ) : <EOL> return self . supports_transactions <EOL> class DatabaseWrapper ( BaseDatabaseWrapper ) : <EOL> _DJANGO_VERSION = _DJANGO_VERSION <EOL> drv_name = None <EOL> driver_supports_utf8 = None <EOL> MARS_Connection = False <EOL> unicode_results = False <EOL> datefirst = <NUM_LIT:7> <EOL> Database = Database <EOL> limit_table_list = False <EOL> vendor = '<STR_LIT>' <EOL> operators = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( DatabaseWrapper , self ) . __init__ ( * args , ** kwargs ) <EOL> options = self . settings_dict . get ( '<STR_LIT>' , None ) <EOL> if options : <EOL> self . MARS_Connection = options . get ( '<STR_LIT>' , False ) <EOL> self . datefirst = options . get ( '<STR_LIT>' , <NUM_LIT:7> ) <EOL> self . unicode_results = options . get ( '<STR_LIT>' , False ) <EOL> self . encoding = options . get ( '<STR_LIT>' , '<STR_LIT:utf-8>' ) <EOL> self . driver_supports_utf8 = options . get ( '<STR_LIT>' , None ) <EOL> self . driver_needs_utf8 = options . get ( '<STR_LIT>' , None ) <EOL> self . limit_table_list = options . get ( '<STR_LIT>' , False ) <EOL> self . collation = options . get ( '<STR_LIT>' , None ) <EOL> if self . collation : <EOL> self . operators = dict ( self . __class__ . operators ) <EOL> ops = { } <EOL> for op in self . operators : <EOL> sql = self . operators [ op ] <EOL> if sql . startswith ( '<STR_LIT>' ) : <EOL> ops [ op ] = '<STR_LIT>' % ( sql , self . collation ) <EOL> self . operators . update ( ops ) <EOL> self . test_create = self . settings_dict . get ( '<STR_LIT>' , True ) <EOL> if _DJANGO_VERSION >= <NUM_LIT> : <EOL> self . features = DatabaseFeatures ( self ) <EOL> else : <EOL> self . features = DatabaseFeatures ( ) <EOL> self . ops = DatabaseOperations ( self ) <EOL> self . client = DatabaseClient ( self ) <EOL> self . creation = DatabaseCreation ( self ) <EOL> self . introspection = DatabaseIntrospection ( self ) <EOL> self . validation = BaseDatabaseValidation ( self ) <EOL> self . connection = None <EOL> def get_connection_params ( self ) : <EOL> settings_dict = self . settings_dict <EOL> if not settings_dict [ '<STR_LIT>' ] : <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> raise ImproperlyConfigured ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> conn_params = { <EOL> '<STR_LIT>' : settings_dict [ '<STR_LIT>' ] , <EOL> } <EOL> conn_params . update ( settings_dict [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in conn_params : <EOL> del conn_params [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> conn_params [ '<STR_LIT:user>' ] = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> conn_params [ '<STR_LIT:password>' ] = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> conn_params [ '<STR_LIT:host>' ] = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> conn_params [ '<STR_LIT:port>' ] = settings_dict [ '<STR_LIT>' ] <EOL> return conn_params <EOL> def get_new_connection ( self , conn_params ) : <EOL> return Database . connect ( ** conn_params ) <EOL> def init_connection_state ( self ) : <EOL> pass <EOL> def _set_autocommit ( self , autocommit ) : <EOL> pass <EOL> def _get_connection_string ( self ) : <EOL> settings_dict = self . settings_dict <EOL> db_str , user_str , passwd_str , port_str = None , None , "<STR_LIT>" , None <EOL> options = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> db_str = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> host_str = settings_dict [ '<STR_LIT>' ] <EOL> else : <EOL> host_str = '<STR_LIT:localhost>' <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> user_str = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> passwd_str = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> port_str = settings_dict [ '<STR_LIT>' ] <EOL> if not db_str : <EOL> raise ImproperlyConfigured ( '<STR_LIT>' ) <EOL> cstr_parts = [ ] <EOL> if '<STR_LIT>' in options : <EOL> driver = options [ '<STR_LIT>' ] <EOL> else : <EOL> if os . name == '<STR_LIT>' : <EOL> driver = '<STR_LIT>' <EOL> else : <EOL> driver = '<STR_LIT>' <EOL> if driver == '<STR_LIT>' or driver . endswith ( '<STR_LIT>' ) : <EOL> driver_is_freetds = True <EOL> else : <EOL> driver_is_freetds = False <EOL> ms_drivers = re . compile ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in options : <EOL> cstr_parts . append ( '<STR_LIT>' % options [ '<STR_LIT>' ] ) <EOL> else : <EOL> if os . path . isabs ( driver ) : <EOL> cstr_parts . append ( '<STR_LIT>' % driver ) <EOL> else : <EOL> cstr_parts . append ( '<STR_LIT>' % driver ) <EOL> if ms_drivers . match ( driver ) or driver_is_freetds and options . get ( '<STR_LIT>' , False ) : <EOL> if port_str : <EOL> host_str += '<STR_LIT>' % port_str <EOL> cstr_parts . append ( '<STR_LIT>' % host_str ) <EOL> else : <EOL> cstr_parts . append ( '<STR_LIT>' % host_str ) <EOL> if user_str : <EOL> cstr_parts . append ( '<STR_LIT>' % ( user_str , passwd_str ) ) <EOL> else : <EOL> if ms_drivers . match ( driver ) : <EOL> cstr_parts . append ( '<STR_LIT>' ) <EOL> else : <EOL> cstr_parts . append ( '<STR_LIT>' ) <EOL> cstr_parts . append ( '<STR_LIT>' % db_str ) <EOL> if self . MARS_Connection : <EOL> cstr_parts . append ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in options : <EOL> cstr_parts . append ( options [ '<STR_LIT>' ] ) <EOL> connectionstring = '<STR_LIT:;>' . join ( cstr_parts ) <EOL> return connectionstring <EOL> def _cursor ( self ) : <EOL> new_conn = False <EOL> settings_dict = self . settings_dict <EOL> if self . connection is None : <EOL> new_conn = True <EOL> connstr = self . _get_connection_string ( ) <EOL> options = settings_dict [ '<STR_LIT>' ] <EOL> autocommit = options . get ( '<STR_LIT>' , False ) <EOL> if self . unicode_results : <EOL> self . connection = Database . connect ( connstr , autocommit = autocommit , unicode_results = '<STR_LIT:True>' ) <EOL> else : <EOL> self . connection = Database . connect ( connstr , autocommit = autocommit ) <EOL> connection_created . send ( sender = self . __class__ , connection = self ) <EOL> cursor = self . connection . cursor ( ) <EOL> if new_conn : <EOL> if not self . ops . is_db2 : <EOL> cursor . execute ( "<STR_LIT>" % self . datefirst ) <EOL> if self . ops . sql_server_ver < <NUM_LIT> : <EOL> self . creation . data_types [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . features . can_return_id_from_insert = False <EOL> ms_sqlncli = re . compile ( '<STR_LIT>' ) <EOL> self . drv_name = self . connection . getinfo ( Database . SQL_DRIVER_NAME ) . upper ( ) <EOL> if self . ops . sql_server_ver >= <NUM_LIT> and ms_sqlncli . match ( self . drv_name ) and self . MARS_Connection : <EOL> self . features . can_use_chunked_reads = True <EOL> if self . drv_name . startswith ( '<STR_LIT>' ) : <EOL> if not self . connection . autocommit : <EOL> self . connection . commit ( ) <EOL> freetds_version = self . connection . getinfo ( Database . SQL_DRIVER_VER ) <EOL> if self . driver_supports_utf8 is None : <EOL> try : <EOL> from distutils . version import LooseVersion <EOL> except ImportError : <EOL> warnings . warn ( Warning ( '<STR_LIT>' ) ) <EOL> self . driver_supports_utf8 = not freetds_version . startswith ( '<STR_LIT>' ) <EOL> else : <EOL> self . driver_supports_utf8 = LooseVersion ( freetds_version ) >= LooseVersion ( '<STR_LIT>' ) <EOL> elif self . driver_supports_utf8 is None : <EOL> self . driver_supports_utf8 = ( self . drv_name == '<STR_LIT>' <EOL> or ms_sqlncli . match ( self . drv_name ) ) <EOL> return CursorWrapper ( cursor , self . driver_supports_utf8 , self . encoding ) <EOL> def _execute_foreach ( self , sql , table_names = None ) : <EOL> cursor = self . cursor ( ) <EOL> if not table_names : <EOL> table_names = self . introspection . get_table_list ( cursor ) <EOL> for table_name in table_names : <EOL> cursor . execute ( sql % self . ops . quote_name ( table_name ) ) <EOL> def check_constraints ( self , table_names = None ) : <EOL> self . _execute_foreach ( '<STR_LIT>' , table_names ) <EOL> def disable_constraint_checking ( self ) : <EOL> self . _execute_foreach ( '<STR_LIT>' ) <EOL> return True <EOL> def enable_constraint_checking ( self ) : <EOL> self . check_constraints ( ) <EOL> class CursorWrapper ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cursor , driver_supports_utf8 , encoding = "<STR_LIT>" ) : <EOL> self . cursor = cursor <EOL> self . driver_supports_utf8 = driver_supports_utf8 <EOL> self . last_sql = '<STR_LIT>' <EOL> self . last_params = ( ) <EOL> self . encoding = encoding <EOL> def close ( self ) : <EOL> try : <EOL> self . cursor . close ( ) <EOL> except Database . ProgrammingError : <EOL> pass <EOL> def format_sql ( self , sql , n_params = None ) : <EOL> if not self . driver_supports_utf8 and isinstance ( sql , text_type ) : <EOL> sql = sql . encode ( '<STR_LIT:utf-8>' ) <EOL> if n_params is not None : <EOL> try : <EOL> sql = sql % tuple ( '<STR_LIT:?>' * n_params ) <EOL> except : <EOL> pass <EOL> else : <EOL> if '<STR_LIT:%s>' in sql : <EOL> sql = sql . replace ( '<STR_LIT:%s>' , '<STR_LIT:?>' ) <EOL> return sql <EOL> def format_params ( self , params ) : <EOL> fp = [ ] <EOL> for p in params : <EOL> if isinstance ( p , text_type ) : <EOL> if not self . driver_supports_utf8 : <EOL> fp . append ( p . encode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> fp . append ( p ) <EOL> elif isinstance ( p , binary_type ) : <EOL> if not self . driver_supports_utf8 : <EOL> fp . append ( p . decode ( self . encoding ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> fp . append ( p ) <EOL> elif isinstance ( p , type ( True ) ) : <EOL> if p : <EOL> fp . append ( <NUM_LIT:1> ) <EOL> else : <EOL> fp . append ( <NUM_LIT:0> ) <EOL> else : <EOL> fp . append ( p ) <EOL> return tuple ( fp ) <EOL> def execute ( self , sql , params = ( ) ) : <EOL> self . last_sql = sql <EOL> sql = self . format_sql ( sql , len ( params ) ) <EOL> params = self . format_params ( params ) <EOL> self . last_params = params <EOL> try : <EOL> return self . cursor . execute ( sql , params ) <EOL> except IntegrityError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> raise utils . IntegrityError ( * e . args ) <EOL> except DatabaseError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> raise utils . DatabaseError ( * e . args ) <EOL> def executemany ( self , sql , params_list ) : <EOL> sql = self . format_sql ( sql ) <EOL> if not params_list : <EOL> if '<STR_LIT:?>' in sql : <EOL> return <EOL> else : <EOL> raw_pll = params_list <EOL> params_list = [ self . format_params ( p ) for p in raw_pll ] <EOL> try : <EOL> return self . cursor . executemany ( sql , params_list ) <EOL> except IntegrityError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> raise utils . IntegrityError ( * e . args ) <EOL> except DatabaseError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> raise utils . DatabaseError ( * e . args ) <EOL> def format_results ( self , rows ) : <EOL> """<STR_LIT>""" <EOL> needs_utc = _DJANGO_VERSION >= <NUM_LIT> and settings . USE_TZ <EOL> if not ( needs_utc or not self . driver_supports_utf8 ) : <EOL> return tuple ( rows ) <EOL> fr = [ ] <EOL> for row in rows : <EOL> if not self . driver_supports_utf8 and isinstance ( row , binary_type ) : <EOL> row = row . decode ( self . encoding ) <EOL> elif needs_utc and isinstance ( row , datetime . datetime ) : <EOL> row = row . replace ( tzinfo = timezone . utc ) <EOL> fr . append ( row ) <EOL> return tuple ( fr ) <EOL> def fetchone ( self ) : <EOL> row = self . cursor . fetchone ( ) <EOL> if row is not None : <EOL> return self . format_results ( row ) <EOL> return [ ] <EOL> def fetchmany ( self , chunk ) : <EOL> return [ self . format_results ( row ) for row in self . cursor . fetchmany ( chunk ) ] <EOL> def fetchall ( self ) : <EOL> return [ self . format_results ( row ) for row in self . cursor . fetchall ( ) ] <EOL> def __getattr__ ( self , attr ) : <EOL> if attr in self . __dict__ : <EOL> return self . __dict__ [ attr ] <EOL> return getattr ( self . cursor , attr ) <EOL> def __iter__ ( self ) : <EOL> return iter ( self . cursor ) <EOL> def savepoint_commit ( self , sid ) : <EOL> if self . queries : <EOL> self . queries . append ( { <EOL> '<STR_LIT>' : '<STR_LIT>' % self . ops . quote_name ( sid ) , <EOL> '<STR_LIT:time>' : '<STR_LIT>' , <EOL> } ) </s>
