<s> """<STR_LIT>""" <EOL> import logging <EOL> import re <EOL> from sqlalchemy . engine . reflection import Inspector <EOL> from sqlalchemy . util import OrderedSet <EOL> from . compare import _compare_tables <EOL> from . render import _drop_table , _drop_column , _drop_index , _drop_constraint , _add_table , _add_column , _add_index , _add_constraint , _modify_col <EOL> from . . import util <EOL> log = logging . getLogger ( __name__ ) <EOL> def compare_metadata ( context , metadata ) : <EOL> """<STR_LIT>""" <EOL> autogen_context , connection = _autogen_context ( context , None ) <EOL> diffs = [ ] <EOL> _produce_net_changes ( connection , metadata , diffs , autogen_context ) <EOL> return diffs <EOL> def _produce_migration_diffs ( context , template_args , <EOL> imports , include_symbol = None , <EOL> include_object = None , <EOL> include_schemas = False ) : <EOL> opts = context . opts <EOL> metadata = opts [ '<STR_LIT>' ] <EOL> include_object = opts . get ( '<STR_LIT>' , include_object ) <EOL> include_symbol = opts . get ( '<STR_LIT>' , include_symbol ) <EOL> include_schemas = opts . get ( '<STR_LIT>' , include_schemas ) <EOL> object_filters = [ ] <EOL> if include_symbol : <EOL> def include_symbol_filter ( object , name , type_ , reflected , compare_to ) : <EOL> if type_ == "<STR_LIT>" : <EOL> return include_symbol ( name , object . schema ) <EOL> else : <EOL> return True <EOL> object_filters . append ( include_symbol_filter ) <EOL> if include_object : <EOL> object_filters . append ( include_object ) <EOL> if metadata is None : <EOL> raise util . CommandError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> context . script . env_py_location <EOL> ) ) <EOL> autogen_context , connection = _autogen_context ( context , imports ) <EOL> diffs = [ ] <EOL> _produce_net_changes ( connection , metadata , diffs , <EOL> autogen_context , object_filters , include_schemas ) <EOL> template_args [ opts [ '<STR_LIT>' ] ] = _indent ( _produce_upgrade_commands ( diffs , autogen_context ) ) <EOL> template_args [ opts [ '<STR_LIT>' ] ] = _indent ( _produce_downgrade_commands ( diffs , autogen_context ) ) <EOL> template_args [ '<STR_LIT>' ] = "<STR_LIT:\n>" . join ( sorted ( imports ) ) <EOL> def _autogen_context ( context , imports ) : <EOL> opts = context . opts <EOL> connection = context . bind <EOL> return { <EOL> '<STR_LIT>' : imports , <EOL> '<STR_LIT>' : connection , <EOL> '<STR_LIT>' : connection . dialect , <EOL> '<STR_LIT>' : context , <EOL> '<STR_LIT>' : opts <EOL> } , connection <EOL> def _indent ( text ) : <EOL> text = "<STR_LIT>" "<STR_LIT>" + text <EOL> text += "<STR_LIT>" <EOL> text = re . compile ( r'<STR_LIT>' , re . M ) . sub ( "<STR_LIT:U+0020>" , text ) . strip ( ) <EOL> return text <EOL> def _produce_net_changes ( connection , metadata , diffs , autogen_context , <EOL> object_filters = ( ) , <EOL> include_schemas = False ) : <EOL> inspector = Inspector . from_engine ( connection ) <EOL> conn_table_names = set ( ) <EOL> if include_schemas : <EOL> schemas = set ( inspector . get_schema_names ( ) ) <EOL> schemas . discard ( "<STR_LIT>" ) <EOL> schemas . add ( None ) <EOL> schemas . discard ( connection . dialect . default_schema_name ) <EOL> else : <EOL> schemas = [ None ] <EOL> for s in schemas : <EOL> tables = set ( inspector . get_table_names ( schema = s ) ) . difference ( [ '<STR_LIT>' ] ) <EOL> conn_table_names . update ( zip ( [ s ] * len ( tables ) , tables ) ) <EOL> metadata_table_names = OrderedSet ( [ ( table . schema , table . name ) <EOL> for table in metadata . sorted_tables ] ) <EOL> _compare_tables ( conn_table_names , metadata_table_names , <EOL> object_filters , <EOL> inspector , metadata , diffs , autogen_context ) <EOL> def _produce_upgrade_commands ( diffs , autogen_context ) : <EOL> buf = [ ] <EOL> for diff in diffs : <EOL> buf . append ( _invoke_command ( "<STR_LIT>" , diff , autogen_context ) ) <EOL> if not buf : <EOL> buf = [ "<STR_LIT>" ] <EOL> return "<STR_LIT:\n>" . join ( buf ) <EOL> def _produce_downgrade_commands ( diffs , autogen_context ) : <EOL> buf = [ ] <EOL> for diff in reversed ( diffs ) : <EOL> buf . append ( _invoke_command ( "<STR_LIT>" , diff , autogen_context ) ) <EOL> if not buf : <EOL> buf = [ "<STR_LIT>" ] <EOL> return "<STR_LIT:\n>" . join ( buf ) <EOL> def _invoke_command ( updown , args , autogen_context ) : <EOL> if isinstance ( args , tuple ) : <EOL> return _invoke_adddrop_command ( updown , args , autogen_context ) <EOL> else : <EOL> return _invoke_modify_command ( updown , args , autogen_context ) <EOL> def _invoke_adddrop_command ( updown , args , autogen_context ) : <EOL> cmd_type = args [ <NUM_LIT:0> ] <EOL> adddrop , cmd_type = cmd_type . split ( "<STR_LIT:_>" ) <EOL> cmd_args = args [ <NUM_LIT:1> : ] + ( autogen_context , ) <EOL> _commands = { <EOL> "<STR_LIT>" : ( _drop_table , _add_table ) , <EOL> "<STR_LIT>" : ( _drop_column , _add_column ) , <EOL> "<STR_LIT:index>" : ( _drop_index , _add_index ) , <EOL> "<STR_LIT>" : ( _drop_constraint , _add_constraint ) , <EOL> } <EOL> cmd_callables = _commands [ cmd_type ] <EOL> if ( <EOL> updown == "<STR_LIT>" and adddrop == "<STR_LIT>" <EOL> ) or ( <EOL> updown == "<STR_LIT>" and adddrop == "<STR_LIT>" <EOL> ) : <EOL> return cmd_callables [ <NUM_LIT:1> ] ( * cmd_args ) <EOL> else : <EOL> return cmd_callables [ <NUM_LIT:0> ] ( * cmd_args ) <EOL> def _invoke_modify_command ( updown , args , autogen_context ) : <EOL> sname , tname , cname = args [ <NUM_LIT:0> ] [ <NUM_LIT:1> : <NUM_LIT:4> ] <EOL> kw = { } <EOL> _arg_struct = { <EOL> "<STR_LIT>" : ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> "<STR_LIT>" : ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> "<STR_LIT>" : ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> } <EOL> for diff in args : <EOL> diff_kw = diff [ <NUM_LIT:4> ] <EOL> for arg in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> if arg in diff_kw : <EOL> kw . setdefault ( arg , diff_kw [ arg ] ) <EOL> old_kw , new_kw = _arg_struct [ diff [ <NUM_LIT:0> ] ] <EOL> if updown == "<STR_LIT>" : <EOL> kw [ new_kw ] = diff [ - <NUM_LIT:1> ] <EOL> kw [ old_kw ] = diff [ - <NUM_LIT:2> ] <EOL> else : <EOL> kw [ new_kw ] = diff [ - <NUM_LIT:2> ] <EOL> kw [ old_kw ] = diff [ - <NUM_LIT:1> ] <EOL> if "<STR_LIT>" in kw : <EOL> kw . pop ( "<STR_LIT>" , None ) <EOL> if "<STR_LIT>" in kw : <EOL> kw . pop ( "<STR_LIT>" , None ) <EOL> return _modify_col ( tname , cname , autogen_context , schema = sname , ** kw ) </s>
