<s> """<STR_LIT>""" <EOL> import sys <EOL> from django . db . backends import * <EOL> from django . db . backends . postgresql_psycopg2 . operations import DatabaseOperations <EOL> from django . db . backends . postgresql_psycopg2 . client import DatabaseClient <EOL> from django . db . backends . postgresql_psycopg2 . creation import DatabaseCreation <EOL> from django . db . backends . postgresql_psycopg2 . version import get_version <EOL> from django . db . backends . postgresql_psycopg2 . introspection import DatabaseIntrospection <EOL> from django . utils . encoding import force_str <EOL> from django . utils . functional import cached_property <EOL> from django . utils . safestring import SafeText , SafeBytes <EOL> from django . utils . timezone import utc <EOL> try : <EOL> import psycopg2 as Database <EOL> import psycopg2 . extensions <EOL> except ImportError as e : <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % e ) <EOL> DatabaseError = Database . DatabaseError <EOL> IntegrityError = Database . IntegrityError <EOL> psycopg2 . extensions . register_type ( psycopg2 . extensions . UNICODE ) <EOL> psycopg2 . extensions . register_adapter ( SafeBytes , psycopg2 . extensions . QuotedString ) <EOL> psycopg2 . extensions . register_adapter ( SafeText , psycopg2 . extensions . QuotedString ) <EOL> def utc_tzinfo_factory ( offset ) : <EOL> if offset != <NUM_LIT:0> : <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> return utc <EOL> class DatabaseFeatures ( BaseDatabaseFeatures ) : <EOL> needs_datetime_string_cast = False <EOL> can_return_id_from_insert = True <EOL> requires_rollback_on_dirty_transaction = True <EOL> has_real_datatype = True <EOL> can_defer_constraint_checks = True <EOL> has_select_for_update = True <EOL> has_select_for_update_nowait = True <EOL> has_bulk_insert = True <EOL> uses_savepoints = True <EOL> supports_tablespaces = True <EOL> supports_transactions = True <EOL> can_distinct_on_fields = True <EOL> can_rollback_ddl = True <EOL> class DatabaseWrapper ( BaseDatabaseWrapper ) : <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> Database = Database <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( DatabaseWrapper , self ) . __init__ ( * args , ** kwargs ) <EOL> opts = self . settings_dict [ "<STR_LIT>" ] <EOL> RC = psycopg2 . extensions . ISOLATION_LEVEL_READ_COMMITTED <EOL> self . isolation_level = opts . get ( '<STR_LIT>' , RC ) <EOL> self . features = DatabaseFeatures ( self ) <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 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 '<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>' ] = force_str ( 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> settings_dict = self . settings_dict <EOL> self . connection . set_client_encoding ( '<STR_LIT>' ) <EOL> tz = '<STR_LIT>' if settings . USE_TZ else settings_dict . get ( '<STR_LIT>' ) <EOL> if tz : <EOL> try : <EOL> get_parameter_status = self . connection . get_parameter_status <EOL> except AttributeError : <EOL> conn_tz = None <EOL> else : <EOL> conn_tz = get_parameter_status ( '<STR_LIT>' ) <EOL> if conn_tz != tz : <EOL> self . connection . cursor ( ) . execute ( <EOL> self . ops . set_time_zone_sql ( ) , [ tz ] ) <EOL> self . connection . commit ( ) <EOL> self . connection . set_isolation_level ( self . isolation_level ) <EOL> def create_cursor ( self ) : <EOL> cursor = self . connection . cursor ( ) <EOL> cursor . tzinfo_factory = utc_tzinfo_factory if settings . USE_TZ else None <EOL> return cursor <EOL> def _set_isolation_level ( self , isolation_level ) : <EOL> assert isolation_level in range ( <NUM_LIT:1> , <NUM_LIT:5> ) <EOL> if self . psycopg2_version >= ( <NUM_LIT:2> , <NUM_LIT:4> , <NUM_LIT:2> ) : <EOL> self . connection . set_session ( isolation_level = isolation_level ) <EOL> else : <EOL> self . connection . set_isolation_level ( isolation_level ) <EOL> def _set_autocommit ( self , autocommit ) : <EOL> with self . wrap_database_errors : <EOL> if self . psycopg2_version >= ( <NUM_LIT:2> , <NUM_LIT:4> , <NUM_LIT:2> ) : <EOL> self . connection . autocommit = autocommit <EOL> else : <EOL> if autocommit : <EOL> level = psycopg2 . extensions . ISOLATION_LEVEL_AUTOCOMMIT <EOL> else : <EOL> level = self . isolation_level <EOL> self . connection . set_isolation_level ( level ) <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 . cursor ( ) . execute ( "<STR_LIT>" ) <EOL> except Database . Error : <EOL> return False <EOL> else : <EOL> return True <EOL> @ cached_property <EOL> def psycopg2_version ( self ) : <EOL> version = psycopg2 . __version__ . split ( '<STR_LIT:U+0020>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> return tuple ( int ( v ) for v in version . split ( '<STR_LIT:.>' ) ) <EOL> @ cached_property <EOL> def pg_version ( self ) : <EOL> with self . temporary_connection ( ) : <EOL> return get_version ( self . connection ) </s>
