<s> import sys <EOL> import time <EOL> from django . conf import settings <EOL> from django . utils . datastructures import DictWrapper <EOL> TEST_DATABASE_PREFIX = '<STR_LIT>' <EOL> class BaseDatabaseCreation ( object ) : <EOL> """<STR_LIT>""" <EOL> data_types = { } <EOL> def __init__ ( self , connection ) : <EOL> self . connection = connection <EOL> def _digest ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( abs ( hash ( args ) ) % <NUM_LIT> L ) <EOL> def db_type ( self , field ) : <EOL> return self . _db_type ( field , field . get_internal_type ( ) ) <EOL> def related_db_type ( self , field ) : <EOL> return self . _db_type ( field , field . get_related_internal_type ( ) ) <EOL> def _db_type ( self , field , internal_type ) : <EOL> data = DictWrapper ( field . __dict__ , self . connection . ops . quote_name , "<STR_LIT>" ) <EOL> try : <EOL> return self . connection . creation . data_types [ internal_type ] % data <EOL> except KeyError : <EOL> return None <EOL> def sql_create_model ( self , model , style , known_models = set ( ) ) : <EOL> """<STR_LIT>""" <EOL> opts = model . _meta <EOL> if not opts . managed or opts . proxy : <EOL> return [ ] , { } <EOL> final_output = [ ] <EOL> table_output = [ ] <EOL> pending_references = { } <EOL> qn = self . connection . ops . quote_name <EOL> for f in opts . local_fields : <EOL> col_type = f . db_type ( connection = self . connection ) <EOL> tablespace = f . db_tablespace or opts . db_tablespace <EOL> if col_type is None : <EOL> continue <EOL> field_output = [ style . SQL_FIELD ( qn ( f . column ) ) , <EOL> style . SQL_COLTYPE ( col_type ) ] <EOL> if not f . null : <EOL> field_output . append ( style . SQL_KEYWORD ( '<STR_LIT>' ) ) <EOL> if f . primary_key : <EOL> field_output . append ( style . SQL_KEYWORD ( '<STR_LIT>' ) ) <EOL> elif f . unique : <EOL> field_output . append ( style . SQL_KEYWORD ( '<STR_LIT>' ) ) <EOL> if tablespace and f . unique : <EOL> field_output . append ( self . connection . ops . tablespace_sql ( tablespace , inline = True ) ) <EOL> if f . rel : <EOL> ref_output , pending = self . sql_for_inline_foreign_key_references ( f , known_models , style ) <EOL> if pending : <EOL> pr = pending_references . setdefault ( f . rel . to , [ ] ) . append ( ( model , f ) ) <EOL> else : <EOL> field_output . extend ( ref_output ) <EOL> table_output . append ( '<STR_LIT:U+0020>' . join ( field_output ) ) <EOL> for field_constraints in opts . unique_together : <EOL> table_output . append ( style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT>' % "<STR_LIT:U+002CU+0020>" . join ( [ style . SQL_FIELD ( qn ( opts . get_field ( f ) . column ) ) for f in field_constraints ] ) ) <EOL> full_statement = [ style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT:U+0020>' + style . SQL_TABLE ( qn ( opts . db_table ) ) + '<STR_LIT>' ] <EOL> for i , line in enumerate ( table_output ) : <EOL> full_statement . append ( '<STR_LIT>' % ( line , i < len ( table_output ) - <NUM_LIT:1> and '<STR_LIT:U+002C>' or '<STR_LIT>' ) ) <EOL> full_statement . append ( '<STR_LIT:)>' ) <EOL> if opts . db_tablespace : <EOL> full_statement . append ( self . connection . ops . tablespace_sql ( opts . db_tablespace ) ) <EOL> full_statement . append ( '<STR_LIT:;>' ) <EOL> final_output . append ( '<STR_LIT:\n>' . join ( full_statement ) ) <EOL> if opts . has_auto_field : <EOL> auto_column = opts . auto_field . db_column or opts . auto_field . name <EOL> autoinc_sql = self . connection . ops . autoinc_sql ( opts . db_table , auto_column ) <EOL> if autoinc_sql : <EOL> for stmt in autoinc_sql : <EOL> final_output . append ( stmt ) <EOL> return final_output , pending_references <EOL> def sql_for_inline_foreign_key_references ( self , field , known_models , style ) : <EOL> "<STR_LIT>" <EOL> qn = self . connection . ops . quote_name <EOL> if field . rel . to in known_models : <EOL> output = [ style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT:U+0020>' + style . SQL_TABLE ( qn ( field . rel . to . _meta . db_table ) ) + '<STR_LIT>' + style . SQL_FIELD ( qn ( field . rel . to . _meta . get_field ( field . rel . field_name ) . column ) ) + '<STR_LIT:)>' + <EOL> self . connection . ops . deferrable_sql ( ) <EOL> ] <EOL> pending = False <EOL> else : <EOL> output = [ ] <EOL> pending = True <EOL> return output , pending <EOL> def sql_for_pending_references ( self , model , style , pending_references ) : <EOL> "<STR_LIT>" <EOL> from django . db . backends . util import truncate_name <EOL> if not model . _meta . managed or model . _meta . proxy : <EOL> return [ ] <EOL> qn = self . connection . ops . quote_name <EOL> final_output = [ ] <EOL> opts = model . _meta <EOL> if model in pending_references : <EOL> for rel_class , f in pending_references [ model ] : <EOL> rel_opts = rel_class . _meta <EOL> r_table = rel_opts . db_table <EOL> r_col = f . column <EOL> table = opts . db_table <EOL> col = opts . get_field ( f . rel . field_name ) . column <EOL> r_name = '<STR_LIT>' % ( r_col , col , self . _digest ( r_table , table ) ) <EOL> final_output . append ( style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT>' % ( qn ( r_table ) , qn ( truncate_name ( r_name , self . connection . ops . max_name_length ( ) ) ) , <EOL> qn ( r_col ) , qn ( table ) , qn ( col ) , <EOL> self . connection . ops . deferrable_sql ( ) ) ) <EOL> del pending_references [ model ] <EOL> return final_output <EOL> def sql_for_many_to_many ( self , model , style ) : <EOL> "<STR_LIT>" <EOL> import warnings <EOL> warnings . warn ( <EOL> '<STR_LIT>' , <EOL> DeprecationWarning <EOL> ) <EOL> output = [ ] <EOL> for f in model . _meta . local_many_to_many : <EOL> if model . _meta . managed or f . rel . to . _meta . managed : <EOL> output . extend ( self . sql_for_many_to_many_field ( model , f , style ) ) <EOL> return output <EOL> def sql_for_many_to_many_field ( self , model , f , style ) : <EOL> "<STR_LIT>" <EOL> import warnings <EOL> warnings . warn ( <EOL> '<STR_LIT>' , <EOL> DeprecationWarning <EOL> ) <EOL> from django . db import models <EOL> from django . db . backends . util import truncate_name <EOL> output = [ ] <EOL> if f . auto_created : <EOL> opts = model . _meta <EOL> qn = self . connection . ops . quote_name <EOL> tablespace = f . db_tablespace or opts . db_tablespace <EOL> if tablespace : <EOL> sql = self . connection . ops . tablespace_sql ( tablespace , inline = True ) <EOL> if sql : <EOL> tablespace_sql = '<STR_LIT:U+0020>' + sql <EOL> else : <EOL> tablespace_sql = '<STR_LIT>' <EOL> else : <EOL> tablespace_sql = '<STR_LIT>' <EOL> table_output = [ style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT:U+0020>' + style . SQL_TABLE ( qn ( f . m2m_db_table ( ) ) ) + '<STR_LIT>' ] <EOL> table_output . append ( '<STR_LIT>' % <EOL> ( style . SQL_FIELD ( qn ( '<STR_LIT:id>' ) ) , <EOL> style . SQL_COLTYPE ( models . AutoField ( primary_key = True ) . db_type ( connection = self . connection ) ) , <EOL> style . SQL_KEYWORD ( '<STR_LIT>' ) , <EOL> tablespace_sql ) ) <EOL> deferred = [ ] <EOL> inline_output , deferred = self . sql_for_inline_many_to_many_references ( model , f , style ) <EOL> table_output . extend ( inline_output ) <EOL> table_output . append ( '<STR_LIT>' % <EOL> ( style . SQL_KEYWORD ( '<STR_LIT>' ) , <EOL> style . SQL_FIELD ( qn ( f . m2m_column_name ( ) ) ) , <EOL> style . SQL_FIELD ( qn ( f . m2m_reverse_name ( ) ) ) , <EOL> tablespace_sql ) ) <EOL> table_output . append ( '<STR_LIT:)>' ) <EOL> if opts . db_tablespace : <EOL> table_output . append ( self . connection . ops . tablespace_sql ( opts . db_tablespace ) ) <EOL> table_output . append ( '<STR_LIT:;>' ) <EOL> output . append ( '<STR_LIT:\n>' . join ( table_output ) ) <EOL> for r_table , r_col , table , col in deferred : <EOL> r_name = '<STR_LIT>' % ( r_col , col , self . _digest ( r_table , table ) ) <EOL> output . append ( style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT>' % <EOL> ( qn ( r_table ) , <EOL> qn ( truncate_name ( r_name , self . connection . ops . max_name_length ( ) ) ) , <EOL> qn ( r_col ) , qn ( table ) , qn ( col ) , <EOL> self . connection . ops . deferrable_sql ( ) ) ) <EOL> autoinc_sql = self . connection . ops . autoinc_sql ( f . m2m_db_table ( ) , '<STR_LIT:id>' ) <EOL> if autoinc_sql : <EOL> for stmt in autoinc_sql : <EOL> output . append ( stmt ) <EOL> return output <EOL> def sql_for_inline_many_to_many_references ( self , model , field , style ) : <EOL> "<STR_LIT>" <EOL> import warnings <EOL> warnings . warn ( <EOL> '<STR_LIT>' , <EOL> DeprecationWarning <EOL> ) <EOL> from django . db import models <EOL> opts = model . _meta <EOL> qn = self . connection . ops . quote_name <EOL> table_output = [ <EOL> '<STR_LIT>' % <EOL> ( style . SQL_FIELD ( qn ( field . m2m_column_name ( ) ) ) , <EOL> style . SQL_COLTYPE ( models . ForeignKey ( model ) . db_type ( connection = self . connection ) ) , <EOL> style . SQL_KEYWORD ( '<STR_LIT>' ) , <EOL> style . SQL_TABLE ( qn ( opts . db_table ) ) , <EOL> style . SQL_FIELD ( qn ( opts . pk . column ) ) , <EOL> self . connection . ops . deferrable_sql ( ) ) , <EOL> '<STR_LIT>' % <EOL> ( style . SQL_FIELD ( qn ( field . m2m_reverse_name ( ) ) ) , <EOL> style . SQL_COLTYPE ( models . ForeignKey ( field . rel . to ) . db_type ( connection = self . connection ) ) , <EOL> style . SQL_KEYWORD ( '<STR_LIT>' ) , <EOL> style . SQL_TABLE ( qn ( field . rel . to . _meta . db_table ) ) , <EOL> style . SQL_FIELD ( qn ( field . rel . to . _meta . pk . column ) ) , <EOL> self . connection . ops . deferrable_sql ( ) ) <EOL> ] <EOL> deferred = [ ] <EOL> return table_output , deferred <EOL> def sql_indexes_for_model ( self , model , style ) : <EOL> "<STR_LIT>" <EOL> if not model . _meta . managed or model . _meta . proxy : <EOL> return [ ] <EOL> output = [ ] <EOL> for f in model . _meta . local_fields : <EOL> output . extend ( self . sql_indexes_for_field ( model , f , style ) ) <EOL> return output <EOL> def sql_indexes_for_field ( self , model , f , style ) : <EOL> "<STR_LIT>" <EOL> from django . db . backends . util import truncate_name <EOL> if f . db_index and not f . unique : <EOL> qn = self . connection . ops . quote_name <EOL> tablespace = f . db_tablespace or model . _meta . db_tablespace <EOL> if tablespace : <EOL> sql = self . connection . ops . tablespace_sql ( tablespace ) <EOL> if sql : <EOL> tablespace_sql = '<STR_LIT:U+0020>' + sql <EOL> else : <EOL> tablespace_sql = '<STR_LIT>' <EOL> else : <EOL> tablespace_sql = '<STR_LIT>' <EOL> i_name = '<STR_LIT>' % ( model . _meta . db_table , self . _digest ( f . column ) ) <EOL> output = [ style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT:U+0020>' + <EOL> style . SQL_TABLE ( qn ( truncate_name ( i_name , self . connection . ops . max_name_length ( ) ) ) ) + '<STR_LIT:U+0020>' + <EOL> style . SQL_KEYWORD ( '<STR_LIT>' ) + '<STR_LIT:U+0020>' + <EOL> style . SQL_TABLE ( qn ( model . _meta . db_table ) ) + '<STR_LIT:U+0020>' + <EOL> "<STR_LIT>" % style . SQL_FIELD ( qn ( f . column ) ) + <EOL> "<STR_LIT>" % tablespace_sql ] <EOL> else : <EOL> output = [ ] <EOL> return output <EOL> def sql_destroy_model ( self , model , references_to_delete , style ) : <EOL> "<STR_LIT>" <EOL> if not model . _meta . managed or model . _meta . proxy : <EOL> return [ ] <EOL> qn = self . connection . ops . quote_name <EOL> output = [ '<STR_LIT>' % ( style . SQL_KEYWORD ( '<STR_LIT>' ) , <EOL> style . SQL_TABLE ( qn ( model . _meta . db_table ) ) ) ] <EOL> if model in references_to_delete : <EOL> output . extend ( self . sql_remove_table_constraints ( model , references_to_delete , style ) ) <EOL> if model . _meta . has_auto_field : <EOL> ds = self . connection . ops . drop_sequence_sql ( model . _meta . db_table ) <EOL> if ds : <EOL> output . append ( ds ) <EOL> return output <EOL> def sql_remove_table_constraints ( self , model , references_to_delete , style ) : <EOL> from django . db . backends . util import truncate_name <EOL> if not model . _meta . managed or model . _meta . proxy : <EOL> return [ ] <EOL> output = [ ] <EOL> qn = self . connection . ops . quote_name <EOL> for rel_class , f in references_to_delete [ model ] : <EOL> table = rel_class . _meta . db_table <EOL> col = f . column <EOL> r_table = model . _meta . db_table <EOL> r_col = model . _meta . get_field ( f . rel . field_name ) . column <EOL> r_name = '<STR_LIT>' % ( col , r_col , self . _digest ( table , r_table ) ) <EOL> output . append ( '<STR_LIT>' % ( style . SQL_KEYWORD ( '<STR_LIT>' ) , <EOL> style . SQL_TABLE ( qn ( table ) ) , <EOL> style . SQL_KEYWORD ( self . connection . ops . drop_foreignkey_sql ( ) ) , <EOL> style . SQL_FIELD ( qn ( truncate_name ( r_name , self . connection . ops . max_name_length ( ) ) ) ) ) ) <EOL> del references_to_delete [ model ] <EOL> return output <EOL> def sql_destroy_many_to_many ( self , model , f , style ) : <EOL> "<STR_LIT>" <EOL> import warnings <EOL> warnings . warn ( <EOL> '<STR_LIT>' , <EOL> DeprecationWarning <EOL> ) <EOL> qn = self . connection . ops . quote_name <EOL> output = [ ] <EOL> if f . auto_created : <EOL> output . append ( "<STR_LIT>" % ( style . SQL_KEYWORD ( '<STR_LIT>' ) , <EOL> style . SQL_TABLE ( qn ( f . m2m_db_table ( ) ) ) ) ) <EOL> ds = self . connection . ops . drop_sequence_sql ( "<STR_LIT>" % ( model . _meta . db_table , f . column ) ) <EOL> if ds : <EOL> output . append ( ds ) <EOL> return output <EOL> def create_test_db ( self , verbosity = <NUM_LIT:1> , autoclobber = False ) : <EOL> """<STR_LIT>""" <EOL> from django . core . management import call_command <EOL> test_database_name = self . _get_test_db_name ( ) <EOL> if verbosity >= <NUM_LIT:1> : <EOL> test_db_repr = '<STR_LIT>' <EOL> if verbosity >= <NUM_LIT:2> : <EOL> test_db_repr = "<STR_LIT>" % test_database_name <EOL> print "<STR_LIT>" % ( self . connection . alias , test_db_repr ) <EOL> self . _create_test_db ( verbosity , autoclobber ) <EOL> self . connection . close ( ) <EOL> self . connection . settings_dict [ "<STR_LIT>" ] = test_database_name <EOL> self . connection . features . confirm ( ) <EOL> call_command ( '<STR_LIT>' , <EOL> verbosity = max ( verbosity - <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> interactive = False , <EOL> database = self . connection . alias , <EOL> load_initial_data = False ) <EOL> call_command ( '<STR_LIT>' , <EOL> verbosity = max ( verbosity - <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> interactive = False , <EOL> database = self . connection . alias ) <EOL> from django . core . cache import get_cache <EOL> from django . core . cache . backends . db import BaseDatabaseCache <EOL> for cache_alias in settings . CACHES : <EOL> cache = get_cache ( cache_alias ) <EOL> if isinstance ( cache , BaseDatabaseCache ) : <EOL> from django . db import router <EOL> if router . allow_syncdb ( self . connection . alias , cache . cache_model_class ) : <EOL> call_command ( '<STR_LIT>' , cache . _table , database = self . connection . alias ) <EOL> cursor = self . connection . cursor ( ) <EOL> return test_database_name <EOL> def _get_test_db_name ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . connection . settings_dict [ '<STR_LIT>' ] : <EOL> return self . connection . settings_dict [ '<STR_LIT>' ] <EOL> return TEST_DATABASE_PREFIX + self . connection . settings_dict [ '<STR_LIT>' ] <EOL> def _create_test_db ( self , verbosity , autoclobber ) : <EOL> "<STR_LIT>" <EOL> suffix = self . sql_table_creation_suffix ( ) <EOL> test_database_name = self . _get_test_db_name ( ) <EOL> qn = self . connection . ops . quote_name <EOL> cursor = self . connection . cursor ( ) <EOL> self . set_autocommit ( ) <EOL> try : <EOL> cursor . execute ( "<STR_LIT>" % ( qn ( test_database_name ) , suffix ) ) <EOL> except Exception , e : <EOL> sys . stderr . write ( "<STR_LIT>" % e ) <EOL> if not autoclobber : <EOL> confirm = raw_input ( "<STR_LIT>" % test_database_name ) <EOL> if autoclobber or confirm == '<STR_LIT:yes>' : <EOL> try : <EOL> if verbosity >= <NUM_LIT:1> : <EOL> print "<STR_LIT>" % self . connection . alias <EOL> cursor . execute ( "<STR_LIT>" % qn ( test_database_name ) ) <EOL> cursor . execute ( "<STR_LIT>" % ( qn ( test_database_name ) , suffix ) ) <EOL> except Exception , e : <EOL> sys . stderr . write ( "<STR_LIT>" % e ) <EOL> sys . exit ( <NUM_LIT:2> ) <EOL> else : <EOL> print "<STR_LIT>" <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> return test_database_name <EOL> def destroy_test_db ( self , old_database_name , verbosity = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> self . connection . close ( ) <EOL> test_database_name = self . connection . settings_dict [ '<STR_LIT>' ] <EOL> if verbosity >= <NUM_LIT:1> : <EOL> test_db_repr = '<STR_LIT>' <EOL> if verbosity >= <NUM_LIT:2> : <EOL> test_db_repr = "<STR_LIT>" % test_database_name <EOL> print "<STR_LIT>" % ( self . connection . alias , test_db_repr ) <EOL> self . connection . settings_dict [ '<STR_LIT>' ] = old_database_name <EOL> self . _destroy_test_db ( test_database_name , verbosity ) <EOL> def _destroy_test_db ( self , test_database_name , verbosity ) : <EOL> "<STR_LIT>" <EOL> cursor = self . connection . cursor ( ) <EOL> self . set_autocommit ( ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> cursor . execute ( "<STR_LIT>" % self . connection . ops . quote_name ( test_database_name ) ) <EOL> self . connection . close ( ) <EOL> def set_autocommit ( self ) : <EOL> "<STR_LIT>" <EOL> if hasattr ( self . connection . connection , "<STR_LIT>" ) : <EOL> if callable ( self . connection . connection . autocommit ) : <EOL> self . connection . connection . autocommit ( True ) <EOL> else : <EOL> self . connection . connection . autocommit = True <EOL> elif hasattr ( self . connection . connection , "<STR_LIT>" ) : <EOL> self . connection . connection . set_isolation_level ( <NUM_LIT:0> ) <EOL> def sql_table_creation_suffix ( self ) : <EOL> "<STR_LIT>" <EOL> return '<STR_LIT>' <EOL> def test_db_signature ( self ) : <EOL> """<STR_LIT>""" <EOL> settings_dict = self . connection . settings_dict <EOL> return ( <EOL> settings_dict [ '<STR_LIT>' ] , <EOL> settings_dict [ '<STR_LIT>' ] , <EOL> settings_dict [ '<STR_LIT>' ] , <EOL> settings_dict [ '<STR_LIT>' ] <EOL> ) </s>
