<s> from contextlib import contextmanager <EOL> from sqlalchemy . types import NULLTYPE , Integer <EOL> from sqlalchemy import schema as sa_schema <EOL> from . import util <EOL> from . compat import string_types <EOL> from . ddl import impl <EOL> __all__ = ( '<STR_LIT>' , ) <EOL> try : <EOL> from sqlalchemy . sql . naming import conv <EOL> except : <EOL> conv = None <EOL> class Operations ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , migration_context ) : <EOL> """<STR_LIT>""" <EOL> self . migration_context = migration_context <EOL> self . impl = migration_context . impl <EOL> @ classmethod <EOL> @ contextmanager <EOL> def context ( cls , migration_context ) : <EOL> from . op import _install_proxy , _remove_proxy <EOL> op = Operations ( migration_context ) <EOL> _install_proxy ( op ) <EOL> yield op <EOL> _remove_proxy ( ) <EOL> def _primary_key_constraint ( self , name , table_name , cols , schema = None ) : <EOL> m = self . _metadata ( ) <EOL> columns = [ sa_schema . Column ( n , NULLTYPE ) for n in cols ] <EOL> t1 = sa_schema . Table ( table_name , m , <EOL> * columns , <EOL> schema = schema ) <EOL> p = sa_schema . PrimaryKeyConstraint ( * columns , name = name ) <EOL> t1 . append_constraint ( p ) <EOL> return p <EOL> def _foreign_key_constraint ( self , name , source , referent , <EOL> local_cols , remote_cols , <EOL> onupdate = None , ondelete = None , <EOL> deferrable = None , source_schema = None , <EOL> referent_schema = None , initially = None , <EOL> match = None , ** dialect_kw ) : <EOL> m = self . _metadata ( ) <EOL> if source == referent : <EOL> t1_cols = local_cols + remote_cols <EOL> else : <EOL> t1_cols = local_cols <EOL> sa_schema . Table ( referent , m , <EOL> * [ sa_schema . Column ( n , NULLTYPE ) for n in remote_cols ] , <EOL> schema = referent_schema ) <EOL> t1 = sa_schema . Table ( source , m , <EOL> * [ sa_schema . Column ( n , NULLTYPE ) for n in t1_cols ] , <EOL> schema = source_schema ) <EOL> tname = "<STR_LIT>" % ( referent_schema , referent ) if referent_schema else referent <EOL> f = sa_schema . ForeignKeyConstraint ( local_cols , <EOL> [ "<STR_LIT>" % ( tname , n ) <EOL> for n in remote_cols ] , <EOL> name = name , <EOL> onupdate = onupdate , <EOL> ondelete = ondelete , <EOL> deferrable = deferrable , <EOL> initially = initially , <EOL> match = match , <EOL> ** dialect_kw <EOL> ) <EOL> t1 . append_constraint ( f ) <EOL> return f <EOL> def _unique_constraint ( self , name , source , local_cols , schema = None , ** kw ) : <EOL> t = sa_schema . Table ( source , self . _metadata ( ) , <EOL> * [ sa_schema . Column ( n , NULLTYPE ) for n in local_cols ] , <EOL> schema = schema ) <EOL> kw [ '<STR_LIT:name>' ] = name <EOL> uq = sa_schema . UniqueConstraint ( * [ t . c [ n ] for n in local_cols ] , ** kw ) <EOL> t . append_constraint ( uq ) <EOL> return uq <EOL> def _check_constraint ( self , name , source , condition , schema = None , ** kw ) : <EOL> t = sa_schema . Table ( source , self . _metadata ( ) , <EOL> sa_schema . Column ( '<STR_LIT:x>' , Integer ) , schema = schema ) <EOL> ck = sa_schema . CheckConstraint ( condition , name = name , ** kw ) <EOL> t . append_constraint ( ck ) <EOL> return ck <EOL> def _metadata ( self ) : <EOL> kw = { } <EOL> if '<STR_LIT>' in self . migration_context . opts : <EOL> mt = self . migration_context . opts [ '<STR_LIT>' ] <EOL> if hasattr ( mt , '<STR_LIT>' ) : <EOL> kw [ '<STR_LIT>' ] = mt . naming_convention <EOL> return sa_schema . MetaData ( ** kw ) <EOL> def _table ( self , name , * columns , ** kw ) : <EOL> m = self . _metadata ( ) <EOL> t = sa_schema . Table ( name , m , * columns , ** kw ) <EOL> for f in t . foreign_keys : <EOL> self . _ensure_table_for_fk ( m , f ) <EOL> return t <EOL> def _column ( self , name , type_ , ** kw ) : <EOL> return sa_schema . Column ( name , type_ , ** kw ) <EOL> def _index ( self , name , tablename , columns , schema = None , ** kw ) : <EOL> t = sa_schema . Table ( tablename or '<STR_LIT>' , self . _metadata ( ) , <EOL> * [ sa_schema . Column ( n , NULLTYPE ) for n in columns ] , <EOL> schema = schema <EOL> ) <EOL> return sa_schema . Index ( name , * [ t . c [ n ] for n in columns ] , ** kw ) <EOL> def _parse_table_key ( self , table_key ) : <EOL> if '<STR_LIT:.>' in table_key : <EOL> tokens = table_key . split ( '<STR_LIT:.>' ) <EOL> sname = "<STR_LIT:.>" . join ( tokens [ <NUM_LIT:0> : - <NUM_LIT:1> ] ) <EOL> tname = tokens [ - <NUM_LIT:1> ] <EOL> else : <EOL> tname = table_key <EOL> sname = None <EOL> return ( sname , tname ) <EOL> def _ensure_table_for_fk ( self , metadata , fk ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( fk . _colspec , string_types ) : <EOL> table_key , cname = fk . _colspec . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> sname , tname = self . _parse_table_key ( table_key ) <EOL> if table_key not in metadata . tables : <EOL> rel_t = sa_schema . Table ( tname , metadata , schema = sname ) <EOL> else : <EOL> rel_t = metadata . tables [ table_key ] <EOL> if cname not in rel_t . c : <EOL> rel_t . append_column ( sa_schema . Column ( cname , NULLTYPE ) ) <EOL> def get_context ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . migration_context <EOL> def rename_table ( self , old_table_name , new_table_name , schema = None ) : <EOL> """<STR_LIT>""" <EOL> self . impl . rename_table ( <EOL> old_table_name , <EOL> new_table_name , <EOL> schema = schema <EOL> ) <EOL> @ util . _with_legacy_names ( [ ( '<STR_LIT:name>' , '<STR_LIT>' ) ] ) <EOL> def alter_column ( self , table_name , column_name , <EOL> nullable = None , <EOL> server_default = False , <EOL> new_column_name = None , <EOL> type_ = None , <EOL> autoincrement = None , <EOL> existing_type = None , <EOL> existing_server_default = False , <EOL> existing_nullable = None , <EOL> existing_autoincrement = None , <EOL> schema = None <EOL> ) : <EOL> """<STR_LIT>""" <EOL> compiler = self . impl . dialect . statement_compiler ( <EOL> self . impl . dialect , <EOL> None <EOL> ) <EOL> def _count_constraint ( constraint ) : <EOL> return not isinstance ( constraint , sa_schema . PrimaryKeyConstraint ) and ( not constraint . _create_rule or <EOL> constraint . _create_rule ( compiler ) ) <EOL> if existing_type and type_ : <EOL> t = self . _table ( table_name , <EOL> sa_schema . Column ( column_name , existing_type ) , <EOL> schema = schema <EOL> ) <EOL> for constraint in t . constraints : <EOL> if _count_constraint ( constraint ) : <EOL> self . impl . drop_constraint ( constraint ) <EOL> self . impl . alter_column ( table_name , column_name , <EOL> nullable = nullable , <EOL> server_default = server_default , <EOL> name = new_column_name , <EOL> type_ = type_ , <EOL> schema = schema , <EOL> autoincrement = autoincrement , <EOL> existing_type = existing_type , <EOL> existing_server_default = existing_server_default , <EOL> existing_nullable = existing_nullable , <EOL> existing_autoincrement = existing_autoincrement <EOL> ) <EOL> if type_ : <EOL> t = self . _table ( table_name , <EOL> sa_schema . Column ( column_name , type_ ) , <EOL> schema = schema <EOL> ) <EOL> for constraint in t . constraints : <EOL> if _count_constraint ( constraint ) : <EOL> self . impl . add_constraint ( constraint ) <EOL> def f ( self , name ) : <EOL> """<STR_LIT>""" <EOL> if conv : <EOL> return conv ( name ) <EOL> else : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" ) <EOL> def add_column ( self , table_name , column , schema = None ) : <EOL> """<STR_LIT>""" <EOL> t = self . _table ( table_name , column , schema = schema ) <EOL> self . impl . add_column ( <EOL> table_name , <EOL> column , <EOL> schema = schema <EOL> ) <EOL> for constraint in t . constraints : <EOL> if not isinstance ( constraint , sa_schema . PrimaryKeyConstraint ) : <EOL> self . impl . add_constraint ( constraint ) <EOL> def drop_column ( self , table_name , column_name , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . impl . drop_column ( <EOL> table_name , <EOL> self . _column ( column_name , NULLTYPE ) , <EOL> ** kw <EOL> ) <EOL> def create_primary_key ( self , name , table_name , cols , schema = None ) : <EOL> """<STR_LIT>""" <EOL> self . impl . add_constraint ( <EOL> self . _primary_key_constraint ( name , table_name , cols , <EOL> schema ) <EOL> ) <EOL> def create_foreign_key ( self , name , source , referent , local_cols , <EOL> remote_cols , onupdate = None , ondelete = None , <EOL> deferrable = None , initially = None , match = None , <EOL> source_schema = None , referent_schema = None , <EOL> ** dialect_kw ) : <EOL> """<STR_LIT>""" <EOL> self . impl . add_constraint ( <EOL> self . _foreign_key_constraint ( name , source , referent , <EOL> local_cols , remote_cols , <EOL> onupdate = onupdate , ondelete = ondelete , <EOL> deferrable = deferrable , source_schema = source_schema , <EOL> referent_schema = referent_schema , <EOL> initially = initially , match = match , ** dialect_kw ) <EOL> ) <EOL> def create_unique_constraint ( self , name , source , local_cols , <EOL> schema = None , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . impl . add_constraint ( <EOL> self . _unique_constraint ( name , source , local_cols , <EOL> schema = schema , ** kw ) <EOL> ) <EOL> def create_check_constraint ( self , name , source , condition , <EOL> schema = None , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . impl . add_constraint ( <EOL> self . _check_constraint ( name , source , condition , schema = schema , ** kw ) <EOL> ) <EOL> def create_table ( self , name , * columns , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . impl . create_table ( <EOL> self . _table ( name , * columns , ** kw ) <EOL> ) <EOL> def drop_table ( self , name , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . impl . drop_table ( <EOL> self . _table ( name , ** kw ) <EOL> ) <EOL> def create_index ( self , name , table_name , columns , schema = None , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . impl . create_index ( <EOL> self . _index ( name , table_name , columns , schema = schema , ** kw ) <EOL> ) <EOL> @ util . _with_legacy_names ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> def drop_index ( self , name , table_name = None , schema = None ) : <EOL> """<STR_LIT>""" <EOL> self . impl . drop_index ( <EOL> self . _index ( name , table_name , [ '<STR_LIT:x>' ] , schema = schema ) <EOL> ) <EOL> @ util . _with_legacy_names ( [ ( "<STR_LIT:type>" , "<STR_LIT>" ) ] ) <EOL> def drop_constraint ( self , name , table_name , type_ = None , schema = None ) : <EOL> """<STR_LIT>""" <EOL> t = self . _table ( table_name , schema = schema ) <EOL> types = { <EOL> '<STR_LIT>' : lambda name : sa_schema . ForeignKeyConstraint ( <EOL> [ ] , [ ] , name = name ) , <EOL> '<STR_LIT>' : sa_schema . PrimaryKeyConstraint , <EOL> '<STR_LIT>' : sa_schema . UniqueConstraint , <EOL> '<STR_LIT>' : lambda name : sa_schema . CheckConstraint ( "<STR_LIT>" , name = name ) , <EOL> None : sa_schema . Constraint <EOL> } <EOL> try : <EOL> const = types [ type_ ] <EOL> except KeyError : <EOL> raise TypeError ( "<STR_LIT>" % <EOL> "<STR_LIT:U+002CU+0020>" . join ( sorted ( repr ( x ) for x in types ) ) ) <EOL> const = const ( name = name ) <EOL> t . append_constraint ( const ) <EOL> self . impl . drop_constraint ( const ) <EOL> def bulk_insert ( self , table , rows , multiinsert = True ) : <EOL> """<STR_LIT>""" <EOL> self . impl . bulk_insert ( table , rows , multiinsert = multiinsert ) <EOL> def inline_literal ( self , value , type_ = None ) : <EOL> """<STR_LIT>""" <EOL> return impl . _literal_bindparam ( None , value , type_ = type_ ) <EOL> def execute ( self , sql , execution_options = None ) : <EOL> """<STR_LIT>""" <EOL> self . migration_context . impl . execute ( sql , <EOL> execution_options = execution_options ) <EOL> def get_bind ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . migration_context . impl . bind </s>
