<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> import datetime <EOL> import re <EOL> import sys <EOL> import warnings <EOL> from django . conf import settings <EOL> from django . db import utils <EOL> from django . db . backends import utils as backend_utils <EOL> from django . db . backends . base . base import BaseDatabaseWrapper <EOL> from django . utils import six , timezone <EOL> from django . utils . deprecation import RemovedInDjango20Warning <EOL> from django . utils . encoding import force_str <EOL> from django . utils . functional import cached_property <EOL> from django . utils . safestring import SafeBytes , SafeText <EOL> try : <EOL> import MySQLdb as Database <EOL> except ImportError as e : <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % e ) <EOL> from MySQLdb . constants import CLIENT , FIELD_TYPE <EOL> from MySQLdb . converters import Thing2Literal , conversions <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 . validation import DatabaseValidation <EOL> version = Database . version_info <EOL> if ( version < ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:1> ) or ( <EOL> version [ : <NUM_LIT:3> ] == ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:1> ) and ( len ( version ) < <NUM_LIT:5> or version [ <NUM_LIT:3> ] != '<STR_LIT>' or version [ <NUM_LIT:4> ] < <NUM_LIT:2> ) ) ) : <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % Database . __version__ ) <EOL> DatabaseError = Database . DatabaseError <EOL> IntegrityError = Database . IntegrityError <EOL> def adapt_datetime_warn_on_aware_datetime ( value , conv ) : <EOL> if settings . USE_TZ and timezone . is_aware ( value ) : <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , RemovedInDjango20Warning ) <EOL> value = value . astimezone ( timezone . utc ) . replace ( tzinfo = None ) <EOL> return Thing2Literal ( value . strftime ( "<STR_LIT>" ) , conv ) <EOL> django_conversions = conversions . copy ( ) <EOL> django_conversions . update ( { <EOL> FIELD_TYPE . TIME : backend_utils . typecast_time , <EOL> FIELD_TYPE . DECIMAL : backend_utils . typecast_decimal , <EOL> FIELD_TYPE . NEWDECIMAL : backend_utils . typecast_decimal , <EOL> datetime . datetime : adapt_datetime_warn_on_aware_datetime , <EOL> } ) <EOL> server_version_re = re . compile ( r'<STR_LIT>' ) <EOL> class CursorWrapper ( object ) : <EOL> """<STR_LIT>""" <EOL> codes_for_integrityerror = ( <NUM_LIT> , ) <EOL> def __init__ ( self , cursor ) : <EOL> self . cursor = cursor <EOL> def execute ( self , query , args = None ) : <EOL> try : <EOL> return self . cursor . execute ( query , args ) <EOL> except Database . OperationalError as e : <EOL> if e . args [ <NUM_LIT:0> ] in self . codes_for_integrityerror : <EOL> six . reraise ( utils . IntegrityError , utils . IntegrityError ( * tuple ( e . args ) ) , sys . exc_info ( ) [ <NUM_LIT:2> ] ) <EOL> raise <EOL> def executemany ( self , query , args ) : <EOL> try : <EOL> return self . cursor . executemany ( query , args ) <EOL> except Database . OperationalError as e : <EOL> if e . args [ <NUM_LIT:0> ] in self . codes_for_integrityerror : <EOL> six . reraise ( utils . IntegrityError , utils . IntegrityError ( * tuple ( e . args ) ) , sys . exc_info ( ) [ <NUM_LIT:2> ] ) <EOL> raise <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 iter ( self . cursor ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , type , value , traceback ) : <EOL> self . close ( ) <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:bool>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:date>' , <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:bool>' , <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:time>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> @ cached_property <EOL> def data_types ( self ) : <EOL> if self . features . supports_microsecond_precision : <EOL> return dict ( self . _data_types , DateTimeField = '<STR_LIT>' , TimeField = '<STR_LIT>' ) <EOL> else : <EOL> return self . _data_types <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> 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> Database = Database <EOL> SchemaEditorClass = DatabaseSchemaEditor <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( DatabaseWrapper , self ) . __init__ ( * args , ** kwargs ) <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 = DatabaseValidation ( self ) <EOL> def get_connection_params ( self ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : django_conversions , <EOL> '<STR_LIT>' : '<STR_LIT:utf8>' , <EOL> } <EOL> if six . PY2 : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> settings_dict = self . settings_dict <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> kwargs [ '<STR_LIT:user>' ] = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> kwargs [ '<STR_LIT>' ] = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> kwargs [ '<STR_LIT>' ] = force_str ( settings_dict [ '<STR_LIT>' ] ) <EOL> if settings_dict [ '<STR_LIT>' ] . startswith ( '<STR_LIT:/>' ) : <EOL> kwargs [ '<STR_LIT>' ] = settings_dict [ '<STR_LIT>' ] <EOL> elif settings_dict [ '<STR_LIT>' ] : <EOL> kwargs [ '<STR_LIT:host>' ] = settings_dict [ '<STR_LIT>' ] <EOL> if settings_dict [ '<STR_LIT>' ] : <EOL> kwargs [ '<STR_LIT:port>' ] = int ( settings_dict [ '<STR_LIT>' ] ) <EOL> kwargs [ '<STR_LIT>' ] = CLIENT . FOUND_ROWS <EOL> kwargs . update ( settings_dict [ '<STR_LIT>' ] ) <EOL> return kwargs <EOL> def get_new_connection ( self , conn_params ) : <EOL> conn = Database . connect ( ** conn_params ) <EOL> conn . encoders [ SafeText ] = conn . encoders [ six . text_type ] <EOL> conn . encoders [ SafeBytes ] = conn . encoders [ bytes ] <EOL> return conn <EOL> def init_connection_state ( self ) : <EOL> if self . features . is_sql_auto_is_null_enabled : <EOL> with self . cursor ( ) as cursor : <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> def create_cursor ( self ) : <EOL> cursor = self . connection . cursor ( ) <EOL> return CursorWrapper ( cursor ) <EOL> def _rollback ( self ) : <EOL> try : <EOL> BaseDatabaseWrapper . _rollback ( self ) <EOL> except Database . NotSupportedError : <EOL> pass <EOL> def _set_autocommit ( self , autocommit ) : <EOL> with self . wrap_database_errors : <EOL> self . connection . autocommit ( autocommit ) <EOL> def disable_constraint_checking ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cursor ( ) . execute ( '<STR_LIT>' ) <EOL> return True <EOL> def enable_constraint_checking ( self ) : <EOL> """<STR_LIT>""" <EOL> self . needs_rollback , needs_rollback = False , self . needs_rollback <EOL> try : <EOL> self . cursor ( ) . execute ( '<STR_LIT>' ) <EOL> finally : <EOL> self . needs_rollback = needs_rollback <EOL> def check_constraints ( self , table_names = None ) : <EOL> """<STR_LIT>""" <EOL> cursor = self . cursor ( ) <EOL> if table_names is None : <EOL> table_names = self . introspection . table_names ( cursor ) <EOL> for table_name in table_names : <EOL> primary_key_column_name = self . introspection . get_primary_key_column ( cursor , table_name ) <EOL> if not primary_key_column_name : <EOL> continue <EOL> key_columns = self . introspection . get_key_columns ( cursor , table_name ) <EOL> for column_name , referenced_table_name , referenced_column_name in key_columns : <EOL> cursor . execute ( <EOL> """<STR_LIT>""" % ( <EOL> primary_key_column_name , column_name , table_name , <EOL> referenced_table_name , column_name , referenced_column_name , <EOL> column_name , referenced_column_name , <EOL> ) <EOL> ) <EOL> for bad_row in cursor . fetchall ( ) : <EOL> raise utils . IntegrityError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( <EOL> table_name , bad_row [ <NUM_LIT:0> ] , table_name , column_name , <EOL> bad_row [ <NUM_LIT:1> ] , referenced_table_name , referenced_column_name , <EOL> ) <EOL> ) <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 mysql_version ( self ) : <EOL> with self . temporary_connection ( ) : <EOL> server_info = self . connection . get_server_info ( ) <EOL> match = server_version_re . match ( server_info ) <EOL> if not match : <EOL> raise Exception ( '<STR_LIT>' % server_info ) <EOL> return tuple ( int ( x ) for x in match . groups ( ) ) </s>
