<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> import datetime <EOL> import decimal <EOL> import os <EOL> import platform <EOL> import sys <EOL> import warnings <EOL> from django . conf import settings <EOL> from django . db import utils <EOL> from django . db . backends . base . base import BaseDatabaseWrapper <EOL> from django . db . backends . base . validation import BaseDatabaseValidation <EOL> from django . utils import six , timezone <EOL> from django . utils . deprecation import RemovedInDjango20Warning <EOL> from django . utils . duration import duration_string <EOL> from django . utils . encoding import force_bytes , force_text <EOL> from django . utils . functional import cached_property <EOL> def _setup_environment ( environ ) : <EOL> if platform . system ( ) . upper ( ) . startswith ( '<STR_LIT>' ) : <EOL> try : <EOL> import ctypes <EOL> except ImportError as e : <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> raise ImproperlyConfigured ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % e ) <EOL> kernel32 = ctypes . CDLL ( '<STR_LIT>' ) <EOL> for name , value in environ : <EOL> kernel32 . SetEnvironmentVariableA ( name , value ) <EOL> else : <EOL> os . environ . update ( environ ) <EOL> _setup_environment ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] ) <EOL> try : <EOL> import cx_Oracle as Database <EOL> except ImportError as e : <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % e ) <EOL> from . client import DatabaseClient <EOL> from . creation import DatabaseCreation <EOL> from . features import DatabaseFeatures <EOL> from . introspection import DatabaseIntrospection <EOL> from . operations import DatabaseOperations <EOL> from . schema import DatabaseSchemaEditor <EOL> from . utils import Oracle_datetime , convert_unicode <EOL> DatabaseError = Database . DatabaseError <EOL> IntegrityError = Database . IntegrityError <EOL> class _UninitializedOperatorsDescriptor ( object ) : <EOL> def __get__ ( self , instance , cls = None ) : <EOL> if instance is None : <EOL> raise AttributeError ( "<STR_LIT>" ) <EOL> instance . cursor ( ) . close ( ) <EOL> return instance . __dict__ [ '<STR_LIT>' ] <EOL> class DatabaseWrapper ( BaseDatabaseWrapper ) : <EOL> vendor = '<STR_LIT>' <EOL> data_types = { <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> '<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> data_type_check_constraints = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> operators = _UninitializedOperatorsDescriptor ( ) <EOL> _standard_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> } <EOL> _likec_operators = _standard_operators . copy ( ) <EOL> _likec_operators . update ( { <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> pattern_esc = r"<STR_LIT>" <EOL> _pattern_ops = { <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> _standard_pattern_ops = { k : "<STR_LIT>" + v + "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> for k , v in _pattern_ops . items ( ) } <EOL> _likec_pattern_ops = { k : "<STR_LIT>" + v + "<STR_LIT>" <EOL> for k , v in _pattern_ops . items ( ) } <EOL> Database = Database <EOL> SchemaEditorClass = DatabaseSchemaEditor <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( DatabaseWrapper , self ) . __init__ ( * args , ** kwargs ) <EOL> self . features = DatabaseFeatures ( self ) <EOL> use_returning_into = self . settings_dict [ "<STR_LIT>" ] . get ( '<STR_LIT>' , True ) <EOL> self . features . can_return_id_from_insert = use_returning_into <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> def _connect_string ( self ) : <EOL> settings_dict = self . settings_dict <EOL> if not settings_dict [ '<STR_LIT>' ] . strip ( ) : <EOL> settings_dict [ '<STR_LIT>' ] = '<STR_LIT:localhost>' <EOL> if settings_dict [ '<STR_LIT>' ] . strip ( ) : <EOL> dsn = Database . makedsn ( settings_dict [ '<STR_LIT>' ] , <EOL> int ( settings_dict [ '<STR_LIT>' ] ) , <EOL> settings_dict [ '<STR_LIT>' ] ) <EOL> else : <EOL> dsn = settings_dict [ '<STR_LIT>' ] <EOL> return "<STR_LIT>" % ( settings_dict [ '<STR_LIT>' ] , <EOL> settings_dict [ '<STR_LIT>' ] , dsn ) <EOL> def get_connection_params ( self ) : <EOL> conn_params = self . settings_dict [ '<STR_LIT>' ] . copy ( ) <EOL> if '<STR_LIT>' in conn_params : <EOL> del conn_params [ '<STR_LIT>' ] <EOL> return conn_params <EOL> def get_new_connection ( self , conn_params ) : <EOL> conn_string = convert_unicode ( self . _connect_string ( ) ) <EOL> return Database . connect ( conn_string , ** conn_params ) <EOL> def init_connection_state ( self ) : <EOL> cursor = self . create_cursor ( ) <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> cursor . execute ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" + <EOL> ( "<STR_LIT>" if settings . USE_TZ else '<STR_LIT>' ) <EOL> ) <EOL> cursor . close ( ) <EOL> if '<STR_LIT>' not in self . __dict__ : <EOL> cursor = self . create_cursor ( ) <EOL> try : <EOL> cursor . execute ( "<STR_LIT>" <EOL> % self . _standard_operators [ '<STR_LIT>' ] , <EOL> [ '<STR_LIT:X>' ] ) <EOL> except DatabaseError : <EOL> self . operators = self . _likec_operators <EOL> self . pattern_ops = self . _likec_pattern_ops <EOL> else : <EOL> self . operators = self . _standard_operators <EOL> self . pattern_ops = self . _standard_pattern_ops <EOL> cursor . close ( ) <EOL> try : <EOL> self . connection . stmtcachesize = <NUM_LIT:20> <EOL> except AttributeError : <EOL> pass <EOL> if not self . get_autocommit ( ) : <EOL> self . commit ( ) <EOL> def create_cursor ( self ) : <EOL> return FormatStylePlaceholderCursor ( self . connection ) <EOL> def _commit ( self ) : <EOL> if self . connection is not None : <EOL> try : <EOL> return self . connection . commit ( ) <EOL> except Database . DatabaseError as e : <EOL> x = e . args [ <NUM_LIT:0> ] <EOL> if hasattr ( x , '<STR_LIT:code>' ) and hasattr ( x , '<STR_LIT:message>' ) and x . code == <NUM_LIT> and '<STR_LIT>' in x . message : <EOL> six . reraise ( utils . IntegrityError , utils . IntegrityError ( * tuple ( e . args ) ) , sys . exc_info ( ) [ <NUM_LIT:2> ] ) <EOL> raise <EOL> def _savepoint_commit ( self , sid ) : <EOL> if self . queries_logged : <EOL> self . queries_log . append ( { <EOL> '<STR_LIT>' : '<STR_LIT>' % self . ops . quote_name ( sid ) , <EOL> '<STR_LIT:time>' : '<STR_LIT>' , <EOL> } ) <EOL> def _set_autocommit ( self , autocommit ) : <EOL> with self . wrap_database_errors : <EOL> self . connection . autocommit = autocommit <EOL> def check_constraints ( self , table_names = None ) : <EOL> """<STR_LIT>""" <EOL> self . cursor ( ) . execute ( '<STR_LIT>' ) <EOL> self . cursor ( ) . execute ( '<STR_LIT>' ) <EOL> def is_usable ( self ) : <EOL> try : <EOL> self . connection . ping ( ) <EOL> except Database . Error : <EOL> return False <EOL> else : <EOL> return True <EOL> @ cached_property <EOL> def oracle_full_version ( self ) : <EOL> with self . temporary_connection ( ) : <EOL> return self . connection . version <EOL> @ cached_property <EOL> def oracle_version ( self ) : <EOL> try : <EOL> return int ( self . oracle_full_version . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] ) <EOL> except ValueError : <EOL> return None <EOL> class OracleParam ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , param , cursor , strings_only = False ) : <EOL> if settings . USE_TZ and ( isinstance ( param , datetime . datetime ) and <EOL> not isinstance ( param , Oracle_datetime ) ) : <EOL> if timezone . is_aware ( param ) : <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , RemovedInDjango20Warning ) <EOL> param = param . astimezone ( timezone . utc ) . replace ( tzinfo = None ) <EOL> param = Oracle_datetime . from_datetime ( param ) <EOL> if isinstance ( param , datetime . timedelta ) : <EOL> param = duration_string ( param ) <EOL> if '<STR_LIT:U+0020>' not in param : <EOL> param = '<STR_LIT>' + param <EOL> string_size = <NUM_LIT:0> <EOL> if param is True : <EOL> param = <NUM_LIT:1> <EOL> elif param is False : <EOL> param = <NUM_LIT:0> <EOL> if hasattr ( param , '<STR_LIT>' ) : <EOL> self . force_bytes = param . bind_parameter ( cursor ) <EOL> elif isinstance ( param , Database . Binary ) : <EOL> self . force_bytes = param <EOL> else : <EOL> self . force_bytes = convert_unicode ( param , cursor . charset , strings_only ) <EOL> if isinstance ( self . force_bytes , six . string_types ) : <EOL> string_size = len ( force_bytes ( param , cursor . charset , strings_only ) ) <EOL> if hasattr ( param , '<STR_LIT>' ) : <EOL> self . input_size = param . input_size <EOL> elif string_size > <NUM_LIT> : <EOL> self . input_size = Database . CLOB <EOL> else : <EOL> self . input_size = None <EOL> class VariableWrapper ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , var ) : <EOL> self . var = var <EOL> def bind_parameter ( self , cursor ) : <EOL> return self . var <EOL> def __getattr__ ( self , key ) : <EOL> return getattr ( self . var , key ) <EOL> def __setattr__ ( self , key , value ) : <EOL> if key == '<STR_LIT>' : <EOL> self . __dict__ [ key ] = value <EOL> else : <EOL> setattr ( self . var , key , value ) <EOL> class FormatStylePlaceholderCursor ( object ) : <EOL> """<STR_LIT>""" <EOL> charset = '<STR_LIT:utf-8>' <EOL> def __init__ ( self , connection ) : <EOL> self . cursor = connection . cursor ( ) <EOL> self . cursor . numbersAsStrings = True <EOL> self . cursor . arraysize = <NUM_LIT:100> <EOL> def _format_params ( self , params ) : <EOL> try : <EOL> return { k : OracleParam ( v , self , True ) for k , v in params . items ( ) } <EOL> except AttributeError : <EOL> return tuple ( OracleParam ( p , self , True ) for p in params ) <EOL> def _guess_input_sizes ( self , params_list ) : <EOL> if hasattr ( params_list [ <NUM_LIT:0> ] , '<STR_LIT>' ) : <EOL> sizes = { } <EOL> for params in params_list : <EOL> for k , value in params . items ( ) : <EOL> if value . input_size : <EOL> sizes [ k ] = value . input_size <EOL> self . setinputsizes ( ** sizes ) <EOL> else : <EOL> sizes = [ None ] * len ( params_list [ <NUM_LIT:0> ] ) <EOL> for params in params_list : <EOL> for i , value in enumerate ( params ) : <EOL> if value . input_size : <EOL> sizes [ i ] = value . input_size <EOL> self . setinputsizes ( * sizes ) <EOL> def _param_generator ( self , params ) : <EOL> if hasattr ( params , '<STR_LIT>' ) : <EOL> return { k : v . force_bytes for k , v in params . items ( ) } <EOL> else : <EOL> return [ p . force_bytes for p in params ] <EOL> def _fix_for_params ( self , query , params ) : <EOL> if query . endswith ( '<STR_LIT:;>' ) or query . endswith ( '<STR_LIT:/>' ) : <EOL> query = query [ : - <NUM_LIT:1> ] <EOL> if params is None : <EOL> params = [ ] <EOL> query = convert_unicode ( query , self . charset ) <EOL> elif hasattr ( params , '<STR_LIT>' ) : <EOL> args = { k : "<STR_LIT>" % k for k in params . keys ( ) } <EOL> query = convert_unicode ( query % args , self . charset ) <EOL> else : <EOL> args = [ ( '<STR_LIT>' % i ) for i in range ( len ( params ) ) ] <EOL> query = convert_unicode ( query % tuple ( args ) , self . charset ) <EOL> return query , self . _format_params ( params ) <EOL> def execute ( self , query , params = None ) : <EOL> query , params = self . _fix_for_params ( query , params ) <EOL> self . _guess_input_sizes ( [ params ] ) <EOL> try : <EOL> return self . cursor . execute ( query , self . _param_generator ( params ) ) <EOL> except Database . DatabaseError as e : <EOL> if hasattr ( e . args [ <NUM_LIT:0> ] , '<STR_LIT:code>' ) and e . args [ <NUM_LIT:0> ] . code == <NUM_LIT> and not isinstance ( e , IntegrityError ) : <EOL> six . reraise ( utils . IntegrityError , utils . IntegrityError ( * tuple ( e . args ) ) , sys . exc_info ( ) [ <NUM_LIT:2> ] ) <EOL> raise <EOL> def executemany ( self , query , params = None ) : <EOL> if not params : <EOL> return None <EOL> params_iter = iter ( params ) <EOL> query , firstparams = self . _fix_for_params ( query , next ( params_iter ) ) <EOL> formatted = [ firstparams ] + [ self . _format_params ( p ) for p in params_iter ] <EOL> self . _guess_input_sizes ( formatted ) <EOL> try : <EOL> return self . cursor . executemany ( query , [ self . _param_generator ( p ) for p in formatted ] ) <EOL> except Database . DatabaseError as e : <EOL> if hasattr ( e . args [ <NUM_LIT:0> ] , '<STR_LIT:code>' ) and e . args [ <NUM_LIT:0> ] . code == <NUM_LIT> and not isinstance ( e , IntegrityError ) : <EOL> six . reraise ( utils . IntegrityError , utils . IntegrityError ( * tuple ( e . args ) ) , sys . exc_info ( ) [ <NUM_LIT:2> ] ) <EOL> raise <EOL> def fetchone ( self ) : <EOL> row = self . cursor . fetchone ( ) <EOL> if row is None : <EOL> return row <EOL> return _rowfactory ( row , self . cursor ) <EOL> def fetchmany ( self , size = None ) : <EOL> if size is None : <EOL> size = self . arraysize <EOL> return tuple ( _rowfactory ( r , self . cursor ) for r in self . cursor . fetchmany ( size ) ) <EOL> def fetchall ( self ) : <EOL> return tuple ( _rowfactory ( r , self . cursor ) for r in self . cursor . fetchall ( ) ) <EOL> def close ( self ) : <EOL> try : <EOL> self . cursor . close ( ) <EOL> except Database . InterfaceError : <EOL> pass <EOL> def var ( self , * args ) : <EOL> return VariableWrapper ( self . cursor . var ( * args ) ) <EOL> def arrayvar ( self , * args ) : <EOL> return VariableWrapper ( self . cursor . arrayvar ( * args ) ) <EOL> def __getattr__ ( self , attr ) : <EOL> if attr in self . __dict__ : <EOL> return self . __dict__ [ attr ] <EOL> else : <EOL> return getattr ( self . cursor , attr ) <EOL> def __iter__ ( self ) : <EOL> return CursorIterator ( self . cursor ) <EOL> class CursorIterator ( six . Iterator ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cursor ) : <EOL> self . cursor = cursor <EOL> self . iter = iter ( cursor ) <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def __next__ ( self ) : <EOL> return _rowfactory ( next ( self . iter ) , self . cursor ) <EOL> def _rowfactory ( row , cursor ) : <EOL> casted = [ ] <EOL> for value , desc in zip ( row , cursor . description ) : <EOL> if value is not None and desc [ <NUM_LIT:1> ] is Database . NUMBER : <EOL> precision , scale = desc [ <NUM_LIT:4> : <NUM_LIT:6> ] <EOL> if scale == - <NUM_LIT> : <EOL> if precision == <NUM_LIT:0> : <EOL> if '<STR_LIT:.>' in value : <EOL> value = decimal . Decimal ( value ) <EOL> else : <EOL> value = int ( value ) <EOL> else : <EOL> value = float ( value ) <EOL> elif precision > <NUM_LIT:0> : <EOL> if scale == <NUM_LIT:0> : <EOL> value = int ( value ) <EOL> else : <EOL> value = decimal . Decimal ( value ) <EOL> elif '<STR_LIT:.>' in value : <EOL> value = decimal . Decimal ( value ) <EOL> else : <EOL> value = int ( value ) <EOL> elif desc [ <NUM_LIT:1> ] in ( Database . STRING , Database . FIXED_CHAR , <EOL> Database . LONG_STRING ) : <EOL> value = to_unicode ( value ) <EOL> casted . append ( value ) <EOL> return tuple ( casted ) <EOL> def to_unicode ( s ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( s , six . string_types ) : <EOL> return force_text ( s ) <EOL> return s </s>
