<s> import collections <EOL> import contextlib <EOL> import itertools <EOL> import logging <EOL> import re <EOL> from oslo_utils import timeutils <EOL> import six <EOL> import sqlalchemy <EOL> from sqlalchemy import Boolean <EOL> from sqlalchemy import CheckConstraint <EOL> from sqlalchemy import Column <EOL> from sqlalchemy . engine import Connectable <EOL> from sqlalchemy . engine import reflection <EOL> from sqlalchemy . engine import url as sa_url <EOL> from sqlalchemy import func <EOL> from sqlalchemy import Index <EOL> from sqlalchemy import inspect <EOL> from sqlalchemy import Integer <EOL> from sqlalchemy import MetaData <EOL> from sqlalchemy . sql . expression import literal_column <EOL> from sqlalchemy . sql import text <EOL> from sqlalchemy import String <EOL> from sqlalchemy import Table <EOL> from sqlalchemy . types import NullType <EOL> from oslo_db import exception <EOL> from oslo_db . _i18n import _ , _LI , _LW <EOL> from oslo_db . sqlalchemy import models <EOL> InvalidSortKey = exception . InvalidSortKey <EOL> ColumnError = exception . ColumnError <EOL> LOG = logging . getLogger ( __name__ ) <EOL> _DBURL_REGEX = re . compile ( r"<STR_LIT>" ) <EOL> _VALID_SORT_DIR = [ <EOL> "<STR_LIT:->" . join ( x ) for x in itertools . product ( [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> [ "<STR_LIT>" , "<STR_LIT>" ] ) ] <EOL> def sanitize_db_url ( url ) : <EOL> match = _DBURL_REGEX . match ( url ) <EOL> if match : <EOL> return '<STR_LIT>' % ( url [ : match . start ( <NUM_LIT:1> ) ] , url [ match . end ( <NUM_LIT:2> ) : ] ) <EOL> return url <EOL> def paginate_query ( query , model , limit , sort_keys , marker = None , <EOL> sort_dir = None , sort_dirs = None ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:id>' not in sort_keys : <EOL> LOG . warning ( _LW ( '<STR_LIT>' ) ) <EOL> assert ( not ( sort_dir and sort_dirs ) ) <EOL> if sort_dirs is None and sort_dir is None : <EOL> sort_dir = '<STR_LIT>' <EOL> if sort_dirs is None : <EOL> sort_dirs = [ sort_dir for _sort_key in sort_keys ] <EOL> assert ( len ( sort_dirs ) == len ( sort_keys ) ) <EOL> for current_sort_key , current_sort_dir in zip ( sort_keys , sort_dirs ) : <EOL> try : <EOL> inspect ( model ) . all_orm_descriptors [ current_sort_key ] <EOL> except KeyError : <EOL> raise exception . InvalidSortKey ( current_sort_key ) <EOL> else : <EOL> sort_key_attr = getattr ( model , current_sort_key ) <EOL> try : <EOL> main_sort_dir , __ , null_sort_dir = current_sort_dir . partition ( "<STR_LIT:->" ) <EOL> sort_dir_func = { <EOL> '<STR_LIT>' : sqlalchemy . asc , <EOL> '<STR_LIT>' : sqlalchemy . desc , <EOL> } [ main_sort_dir ] <EOL> null_order_by_stmt = { <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : sort_key_attr . is_ ( None ) , <EOL> "<STR_LIT>" : sort_key_attr . isnot ( None ) , <EOL> } [ null_sort_dir ] <EOL> except KeyError : <EOL> raise ValueError ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % <EOL> "<STR_LIT:U+002CU+0020>" . join ( _VALID_SORT_DIR ) ) <EOL> if null_order_by_stmt is not None : <EOL> query = query . order_by ( sqlalchemy . desc ( null_order_by_stmt ) ) <EOL> query = query . order_by ( sort_dir_func ( sort_key_attr ) ) <EOL> if marker is not None : <EOL> marker_values = [ ] <EOL> for sort_key in sort_keys : <EOL> v = getattr ( marker , sort_key ) <EOL> marker_values . append ( v ) <EOL> criteria_list = [ ] <EOL> for i in range ( len ( sort_keys ) ) : <EOL> crit_attrs = [ ] <EOL> for j in range ( i ) : <EOL> model_attr = getattr ( model , sort_keys [ j ] ) <EOL> crit_attrs . append ( ( model_attr == marker_values [ j ] ) ) <EOL> model_attr = getattr ( model , sort_keys [ i ] ) <EOL> if sort_dirs [ i ] . startswith ( '<STR_LIT>' ) : <EOL> crit_attrs . append ( ( model_attr < marker_values [ i ] ) ) <EOL> else : <EOL> crit_attrs . append ( ( model_attr > marker_values [ i ] ) ) <EOL> criteria = sqlalchemy . sql . and_ ( * crit_attrs ) <EOL> criteria_list . append ( criteria ) <EOL> f = sqlalchemy . sql . or_ ( * criteria_list ) <EOL> query = query . filter ( f ) <EOL> if limit is not None : <EOL> query = query . limit ( limit ) <EOL> return query <EOL> def to_list ( x , default = None ) : <EOL> if x is None : <EOL> return default <EOL> if not isinstance ( x , collections . Iterable ) or isinstance ( x , six . string_types ) : <EOL> return [ x ] <EOL> elif isinstance ( x , list ) : <EOL> return x <EOL> else : <EOL> return list ( x ) <EOL> def _read_deleted_filter ( query , db_model , deleted ) : <EOL> if '<STR_LIT>' not in db_model . __table__ . columns : <EOL> raise ValueError ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> % db_model . __name__ ) <EOL> default_deleted_value = db_model . __table__ . c . deleted . default . arg <EOL> if deleted : <EOL> query = query . filter ( db_model . deleted != default_deleted_value ) <EOL> else : <EOL> query = query . filter ( db_model . deleted == default_deleted_value ) <EOL> return query <EOL> def _project_filter ( query , db_model , project_id ) : <EOL> if '<STR_LIT>' not in db_model . __table__ . columns : <EOL> raise ValueError ( _ ( "<STR_LIT>" ) <EOL> % db_model . __name__ ) <EOL> if isinstance ( project_id , ( list , tuple , set ) ) : <EOL> query = query . filter ( db_model . project_id . in_ ( project_id ) ) <EOL> else : <EOL> query = query . filter ( db_model . project_id == project_id ) <EOL> return query <EOL> def model_query ( model , session , args = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not issubclass ( model , models . ModelBase ) : <EOL> raise TypeError ( _ ( "<STR_LIT>" ) ) <EOL> query = session . query ( model ) if not args else session . query ( * args ) <EOL> if '<STR_LIT>' in kwargs : <EOL> query = _read_deleted_filter ( query , model , kwargs [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in kwargs : <EOL> query = _project_filter ( query , model , kwargs [ '<STR_LIT>' ] ) <EOL> return query <EOL> def get_table ( engine , name ) : <EOL> """<STR_LIT>""" <EOL> metadata = MetaData ( ) <EOL> metadata . bind = engine <EOL> return Table ( name , metadata , autoload = True ) <EOL> class InsertFromSelect ( object ) : <EOL> """<STR_LIT>""" <EOL> def __new__ ( cls , table , select , cols = None ) : <EOL> if not cols : <EOL> cols = [ c . name for c in table . c ] <EOL> return table . insert ( inline = True ) . from_select ( cols , select ) <EOL> def __init__ ( self , table , select , cols = None ) : <EOL> pass <EOL> def _get_not_supported_column ( col_name_col_instance , column_name ) : <EOL> try : <EOL> column = col_name_col_instance [ column_name ] <EOL> except KeyError : <EOL> msg = _ ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> raise exception . ColumnError ( msg % column_name ) <EOL> if not isinstance ( column , Column ) : <EOL> msg = _ ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> raise exception . ColumnError ( msg % column_name ) <EOL> return column <EOL> def drop_old_duplicate_entries_from_table ( migrate_engine , table_name , <EOL> use_soft_delete , * uc_column_names ) : <EOL> """<STR_LIT>""" <EOL> meta = MetaData ( ) <EOL> meta . bind = migrate_engine <EOL> table = Table ( table_name , meta , autoload = True ) <EOL> columns_for_group_by = [ table . c [ name ] for name in uc_column_names ] <EOL> columns_for_select = [ func . max ( table . c . id ) ] <EOL> columns_for_select . extend ( columns_for_group_by ) <EOL> duplicated_rows_select = sqlalchemy . sql . select ( <EOL> columns_for_select , group_by = columns_for_group_by , <EOL> having = func . count ( table . c . id ) > <NUM_LIT:1> ) <EOL> for row in migrate_engine . execute ( duplicated_rows_select ) . fetchall ( ) : <EOL> delete_condition = table . c . id != row [ <NUM_LIT:0> ] <EOL> is_none = None <EOL> delete_condition &= table . c . deleted_at == is_none <EOL> for name in uc_column_names : <EOL> delete_condition &= table . c [ name ] == row [ name ] <EOL> rows_to_delete_select = sqlalchemy . sql . select ( <EOL> [ table . c . id ] ) . where ( delete_condition ) <EOL> for row in migrate_engine . execute ( rows_to_delete_select ) . fetchall ( ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , dict ( id = row [ <NUM_LIT:0> ] , table = table_name ) ) <EOL> if use_soft_delete : <EOL> delete_statement = table . update ( ) . where ( delete_condition ) . values ( { <EOL> '<STR_LIT>' : literal_column ( '<STR_LIT:id>' ) , <EOL> '<STR_LIT>' : literal_column ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : timeutils . utcnow ( ) <EOL> } ) <EOL> else : <EOL> delete_statement = table . delete ( ) . where ( delete_condition ) <EOL> migrate_engine . execute ( delete_statement ) <EOL> def _get_default_deleted_value ( table ) : <EOL> if isinstance ( table . c . id . type , Integer ) : <EOL> return <NUM_LIT:0> <EOL> if isinstance ( table . c . id . type , String ) : <EOL> return "<STR_LIT>" <EOL> raise exception . ColumnError ( _ ( "<STR_LIT>" ) ) <EOL> def _restore_indexes_on_deleted_columns ( migrate_engine , table_name , indexes ) : <EOL> table = get_table ( migrate_engine , table_name ) <EOL> insp = reflection . Inspector . from_engine ( migrate_engine ) <EOL> real_indexes = insp . get_indexes ( table_name ) <EOL> existing_index_names = dict ( <EOL> [ ( index [ '<STR_LIT:name>' ] , index [ '<STR_LIT>' ] ) for index in real_indexes ] ) <EOL> for index in indexes : <EOL> if '<STR_LIT>' not in index [ '<STR_LIT>' ] : <EOL> continue <EOL> name = index [ '<STR_LIT:name>' ] <EOL> if name in existing_index_names : <EOL> column_names = [ table . c [ c ] for c in existing_index_names [ name ] ] <EOL> old_index = Index ( name , * column_names , unique = index [ "<STR_LIT>" ] ) <EOL> old_index . drop ( migrate_engine ) <EOL> column_names = [ table . c [ c ] for c in index [ '<STR_LIT>' ] ] <EOL> new_index = Index ( index [ "<STR_LIT:name>" ] , * column_names , unique = index [ "<STR_LIT>" ] ) <EOL> new_index . create ( migrate_engine ) <EOL> def change_deleted_column_type_to_boolean ( migrate_engine , table_name , <EOL> ** col_name_col_instance ) : <EOL> if migrate_engine . name == "<STR_LIT>" : <EOL> return _change_deleted_column_type_to_boolean_sqlite ( <EOL> migrate_engine , table_name , ** col_name_col_instance ) <EOL> insp = reflection . Inspector . from_engine ( migrate_engine ) <EOL> indexes = insp . get_indexes ( table_name ) <EOL> table = get_table ( migrate_engine , table_name ) <EOL> old_deleted = Column ( '<STR_LIT>' , Boolean , default = False ) <EOL> old_deleted . create ( table , populate_default = False ) <EOL> table . update ( ) . where ( table . c . deleted == table . c . id ) . values ( old_deleted = True ) . execute ( ) <EOL> table . c . deleted . drop ( ) <EOL> table . c . old_deleted . alter ( name = "<STR_LIT>" ) <EOL> _restore_indexes_on_deleted_columns ( migrate_engine , table_name , indexes ) <EOL> def _change_deleted_column_type_to_boolean_sqlite ( migrate_engine , table_name , <EOL> ** col_name_col_instance ) : <EOL> insp = reflection . Inspector . from_engine ( migrate_engine ) <EOL> table = get_table ( migrate_engine , table_name ) <EOL> columns = [ ] <EOL> for column in table . columns : <EOL> column_copy = None <EOL> if column . name != "<STR_LIT>" : <EOL> if isinstance ( column . type , NullType ) : <EOL> column_copy = _get_not_supported_column ( col_name_col_instance , <EOL> column . name ) <EOL> else : <EOL> column_copy = column . copy ( ) <EOL> else : <EOL> column_copy = Column ( '<STR_LIT>' , Boolean , default = <NUM_LIT:0> ) <EOL> columns . append ( column_copy ) <EOL> constraints = [ constraint . copy ( ) for constraint in table . constraints ] <EOL> meta = table . metadata <EOL> new_table = Table ( table_name + "<STR_LIT>" , meta , <EOL> * ( columns + constraints ) ) <EOL> new_table . create ( ) <EOL> indexes = [ ] <EOL> for index in insp . get_indexes ( table_name ) : <EOL> column_names = [ new_table . c [ c ] for c in index [ '<STR_LIT>' ] ] <EOL> indexes . append ( Index ( index [ "<STR_LIT:name>" ] , * column_names , <EOL> unique = index [ "<STR_LIT>" ] ) ) <EOL> c_select = [ ] <EOL> for c in table . c : <EOL> if c . name != "<STR_LIT>" : <EOL> c_select . append ( c ) <EOL> else : <EOL> c_select . append ( table . c . deleted == table . c . id ) <EOL> ins = InsertFromSelect ( new_table , sqlalchemy . sql . select ( c_select ) ) <EOL> migrate_engine . execute ( ins ) <EOL> table . drop ( ) <EOL> for index in indexes : <EOL> index . create ( migrate_engine ) <EOL> new_table . rename ( table_name ) <EOL> new_table . update ( ) . where ( new_table . c . deleted == new_table . c . id ) . values ( deleted = True ) . execute ( ) <EOL> def change_deleted_column_type_to_id_type ( migrate_engine , table_name , <EOL> ** col_name_col_instance ) : <EOL> if migrate_engine . name == "<STR_LIT>" : <EOL> return _change_deleted_column_type_to_id_type_sqlite ( <EOL> migrate_engine , table_name , ** col_name_col_instance ) <EOL> insp = reflection . Inspector . from_engine ( migrate_engine ) <EOL> indexes = insp . get_indexes ( table_name ) <EOL> table = get_table ( migrate_engine , table_name ) <EOL> new_deleted = Column ( '<STR_LIT>' , table . c . id . type , <EOL> default = _get_default_deleted_value ( table ) ) <EOL> new_deleted . create ( table , populate_default = True ) <EOL> deleted = True <EOL> table . update ( ) . where ( table . c . deleted == deleted ) . values ( new_deleted = table . c . id ) . execute ( ) <EOL> table . c . deleted . drop ( ) <EOL> table . c . new_deleted . alter ( name = "<STR_LIT>" ) <EOL> _restore_indexes_on_deleted_columns ( migrate_engine , table_name , indexes ) <EOL> def _change_deleted_column_type_to_id_type_sqlite ( migrate_engine , table_name , <EOL> ** col_name_col_instance ) : <EOL> insp = reflection . Inspector . from_engine ( migrate_engine ) <EOL> meta = MetaData ( bind = migrate_engine ) <EOL> table = Table ( table_name , meta , autoload = True ) <EOL> default_deleted_value = _get_default_deleted_value ( table ) <EOL> columns = [ ] <EOL> for column in table . columns : <EOL> column_copy = None <EOL> if column . name != "<STR_LIT>" : <EOL> if isinstance ( column . type , NullType ) : <EOL> column_copy = _get_not_supported_column ( col_name_col_instance , <EOL> column . name ) <EOL> else : <EOL> column_copy = column . copy ( ) <EOL> else : <EOL> column_copy = Column ( '<STR_LIT>' , table . c . id . type , <EOL> default = default_deleted_value ) <EOL> columns . append ( column_copy ) <EOL> def is_deleted_column_constraint ( constraint ) : <EOL> if not isinstance ( constraint , CheckConstraint ) : <EOL> return False <EOL> sqltext = str ( constraint . sqltext ) <EOL> constraint_markers = ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ) <EOL> return any ( sqltext . endswith ( marker ) for marker in constraint_markers ) <EOL> constraints = [ ] <EOL> for constraint in table . constraints : <EOL> if not is_deleted_column_constraint ( constraint ) : <EOL> constraints . append ( constraint . copy ( ) ) <EOL> new_table = Table ( table_name + "<STR_LIT>" , meta , <EOL> * ( columns + constraints ) ) <EOL> new_table . create ( ) <EOL> indexes = [ ] <EOL> for index in insp . get_indexes ( table_name ) : <EOL> column_names = [ new_table . c [ c ] for c in index [ '<STR_LIT>' ] ] <EOL> indexes . append ( Index ( index [ "<STR_LIT:name>" ] , * column_names , <EOL> unique = index [ "<STR_LIT>" ] ) ) <EOL> ins = InsertFromSelect ( new_table , table . select ( ) ) <EOL> migrate_engine . execute ( ins ) <EOL> table . drop ( ) <EOL> for index in indexes : <EOL> index . create ( migrate_engine ) <EOL> new_table . rename ( table_name ) <EOL> deleted = True <EOL> new_table . update ( ) . where ( new_table . c . deleted == deleted ) . values ( deleted = new_table . c . id ) . execute ( ) <EOL> deleted = False <EOL> new_table . update ( ) . where ( new_table . c . deleted == deleted ) . values ( deleted = default_deleted_value ) . execute ( ) <EOL> def get_connect_string ( backend , database , user = None , passwd = None , <EOL> host = '<STR_LIT:localhost>' ) : <EOL> """<STR_LIT>""" <EOL> args = { '<STR_LIT>' : backend , <EOL> '<STR_LIT:user>' : user , <EOL> '<STR_LIT>' : passwd , <EOL> '<STR_LIT:host>' : host , <EOL> '<STR_LIT>' : database } <EOL> if backend == '<STR_LIT>' : <EOL> template = '<STR_LIT>' <EOL> else : <EOL> template = "<STR_LIT>" <EOL> return template % args <EOL> def is_backend_avail ( backend , database , user = None , passwd = None ) : <EOL> """<STR_LIT>""" <EOL> from oslo_db . sqlalchemy import provision <EOL> connect_uri = get_connect_string ( backend = backend , <EOL> database = database , <EOL> user = user , <EOL> passwd = passwd ) <EOL> try : <EOL> eng = provision . Backend . _ensure_backend_available ( connect_uri ) <EOL> eng . dispose ( ) <EOL> except exception . BackendNotAvailable : <EOL> return False <EOL> else : <EOL> return True <EOL> def get_db_connection_info ( conn_pieces ) : <EOL> database = conn_pieces . path . strip ( '<STR_LIT:/>' ) <EOL> loc_pieces = conn_pieces . netloc . split ( '<STR_LIT:@>' ) <EOL> host = loc_pieces [ <NUM_LIT:1> ] <EOL> auth_pieces = loc_pieces [ <NUM_LIT:0> ] . split ( '<STR_LIT::>' ) <EOL> user = auth_pieces [ <NUM_LIT:0> ] <EOL> password = "<STR_LIT>" <EOL> if len ( auth_pieces ) > <NUM_LIT:1> : <EOL> password = auth_pieces [ <NUM_LIT:1> ] . strip ( ) <EOL> return ( user , password , database , host ) <EOL> def index_exists ( migrate_engine , table_name , index_name ) : <EOL> """<STR_LIT>""" <EOL> inspector = reflection . Inspector . from_engine ( migrate_engine ) <EOL> indexes = inspector . get_indexes ( table_name ) <EOL> index_names = [ index [ '<STR_LIT:name>' ] for index in indexes ] <EOL> return index_name in index_names <EOL> def add_index ( migrate_engine , table_name , index_name , idx_columns ) : <EOL> """<STR_LIT>""" <EOL> table = get_table ( migrate_engine , table_name ) <EOL> if not index_exists ( migrate_engine , table_name , index_name ) : <EOL> index = Index ( <EOL> index_name , * [ getattr ( table . c , col ) for col in idx_columns ] <EOL> ) <EOL> index . create ( ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % index_name ) <EOL> def drop_index ( migrate_engine , table_name , index_name ) : <EOL> """<STR_LIT>""" <EOL> table = get_table ( migrate_engine , table_name ) <EOL> for index in table . indexes : <EOL> if index . name == index_name : <EOL> index . drop ( ) <EOL> break <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % index_name ) <EOL> def change_index_columns ( migrate_engine , table_name , index_name , new_columns ) : <EOL> """<STR_LIT>""" <EOL> drop_index ( migrate_engine , table_name , index_name ) <EOL> add_index ( migrate_engine , table_name , index_name , new_columns ) <EOL> def column_exists ( engine , table_name , column ) : <EOL> """<STR_LIT>""" <EOL> t = get_table ( engine , table_name ) <EOL> return column in t . c <EOL> class DialectFunctionDispatcher ( object ) : <EOL> @ classmethod <EOL> def dispatch_for_dialect ( cls , expr , multiple = False ) : <EOL> """<STR_LIT>""" <EOL> if multiple : <EOL> cls = DialectMultiFunctionDispatcher <EOL> else : <EOL> cls = DialectSingleFunctionDispatcher <EOL> return cls ( ) . dispatch_for ( expr ) <EOL> _db_plus_driver_reg = re . compile ( r'<STR_LIT>' ) <EOL> def dispatch_for ( self , expr ) : <EOL> def decorate ( fn ) : <EOL> dbname , driver = self . _parse_dispatch ( expr ) <EOL> if fn is self : <EOL> fn = fn . _last <EOL> self . _last = fn <EOL> self . _register ( expr , dbname , driver , fn ) <EOL> return self <EOL> return decorate <EOL> def _parse_dispatch ( self , text ) : <EOL> m = self . _db_plus_driver_reg . match ( text ) <EOL> if not m : <EOL> raise ValueError ( "<STR_LIT>" % text ) <EOL> return m . group ( <NUM_LIT:1> ) or '<STR_LIT:*>' , m . group ( <NUM_LIT:2> ) or '<STR_LIT:*>' <EOL> def __call__ ( self , * arg , ** kw ) : <EOL> target = arg [ <NUM_LIT:0> ] <EOL> return self . _dispatch_on ( <EOL> self . _url_from_target ( target ) , target , arg , kw ) <EOL> def _url_from_target ( self , target ) : <EOL> if isinstance ( target , Connectable ) : <EOL> return target . engine . url <EOL> elif isinstance ( target , six . string_types ) : <EOL> if "<STR_LIT>" not in target : <EOL> target_url = sa_url . make_url ( "<STR_LIT>" % target ) <EOL> else : <EOL> target_url = sa_url . make_url ( target ) <EOL> return target_url <EOL> elif isinstance ( target , sa_url . URL ) : <EOL> return target <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % target ) <EOL> def dispatch_on_drivername ( self , drivername ) : <EOL> """<STR_LIT>""" <EOL> dbname , driver = self . _db_plus_driver_reg . match ( drivername ) . group ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> def go ( * arg , ** kw ) : <EOL> return self . _dispatch_on_db_driver ( dbname , "<STR_LIT:*>" , arg , kw ) <EOL> return go <EOL> def _dispatch_on ( self , url , target , arg , kw ) : <EOL> dbname , driver = self . _db_plus_driver_reg . match ( <EOL> url . drivername ) . group ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> if not driver : <EOL> driver = url . get_dialect ( ) . driver <EOL> return self . _dispatch_on_db_driver ( dbname , driver , arg , kw ) <EOL> def _invoke_fn ( self , fn , arg , kw ) : <EOL> return fn ( * arg , ** kw ) <EOL> class DialectSingleFunctionDispatcher ( DialectFunctionDispatcher ) : <EOL> def __init__ ( self ) : <EOL> self . reg = collections . defaultdict ( dict ) <EOL> def _register ( self , expr , dbname , driver , fn ) : <EOL> fn_dict = self . reg [ dbname ] <EOL> if driver in fn_dict : <EOL> raise TypeError ( "<STR_LIT>" % expr ) <EOL> fn_dict [ driver ] = fn <EOL> def _matches ( self , dbname , driver ) : <EOL> for db in ( dbname , '<STR_LIT:*>' ) : <EOL> subdict = self . reg [ db ] <EOL> for drv in ( driver , '<STR_LIT:*>' ) : <EOL> if drv in subdict : <EOL> return subdict [ drv ] <EOL> else : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> ( "<STR_LIT>" % ( dbname , driver ) ) ) <EOL> def _dispatch_on_db_driver ( self , dbname , driver , arg , kw ) : <EOL> fn = self . _matches ( dbname , driver ) <EOL> return self . _invoke_fn ( fn , arg , kw ) <EOL> class DialectMultiFunctionDispatcher ( DialectFunctionDispatcher ) : <EOL> def __init__ ( self ) : <EOL> self . reg = collections . defaultdict ( <EOL> lambda : collections . defaultdict ( list ) ) <EOL> def _register ( self , expr , dbname , driver , fn ) : <EOL> self . reg [ dbname ] [ driver ] . append ( fn ) <EOL> def _matches ( self , dbname , driver ) : <EOL> if driver != '<STR_LIT:*>' : <EOL> drivers = ( driver , '<STR_LIT:*>' ) <EOL> else : <EOL> drivers = ( '<STR_LIT:*>' , ) <EOL> for db in ( dbname , '<STR_LIT:*>' ) : <EOL> subdict = self . reg [ db ] <EOL> for drv in drivers : <EOL> for fn in subdict [ drv ] : <EOL> yield fn <EOL> def _dispatch_on_db_driver ( self , dbname , driver , arg , kw ) : <EOL> for fn in self . _matches ( dbname , driver ) : <EOL> if self . _invoke_fn ( fn , arg , kw ) is not None : <EOL> raise TypeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> dispatch_for_dialect = DialectFunctionDispatcher . dispatch_for_dialect <EOL> def get_non_innodb_tables ( connectable , skip_tables = ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) ) : <EOL> """<STR_LIT>""" <EOL> query_str = """<STR_LIT>""" <EOL> params = { } <EOL> if skip_tables : <EOL> params = dict ( <EOL> ( '<STR_LIT>' % i , table_name ) <EOL> for i , table_name in enumerate ( skip_tables ) <EOL> ) <EOL> placeholders = '<STR_LIT:U+002CU+0020>' . join ( '<STR_LIT::>' + p for p in params ) <EOL> query_str += '<STR_LIT>' % placeholders <EOL> params [ '<STR_LIT>' ] = connectable . engine . url . database <EOL> query = text ( query_str ) <EOL> noninnodb = connectable . execute ( query , ** params ) <EOL> return [ i [ <NUM_LIT:0> ] for i in noninnodb ] <EOL> class NonCommittingConnectable ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , connection ) : <EOL> self . connection = connection <EOL> self . _trans = connection . begin ( ) <EOL> self . _restart_nested ( ) <EOL> def _restart_nested ( self ) : <EOL> self . _nested_trans = self . connection . begin_nested ( ) <EOL> def _dispose ( self ) : <EOL> if not self . connection . closed : <EOL> self . _nested_trans . rollback ( ) <EOL> self . _trans . rollback ( ) <EOL> self . connection . close ( ) <EOL> def execute ( self , obj , * multiparams , ** params ) : <EOL> """<STR_LIT>""" <EOL> return self . connection . execute ( obj , * multiparams , ** params ) <EOL> def scalar ( self , obj , * multiparams , ** params ) : <EOL> """<STR_LIT>""" <EOL> return self . connection . scalar ( obj , * multiparams , ** params ) <EOL> class NonCommittingEngine ( NonCommittingConnectable ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def url ( self ) : <EOL> return self . connection . engine . url <EOL> @ property <EOL> def engine ( self ) : <EOL> return self <EOL> def connect ( self ) : <EOL> return NonCommittingConnection ( self . connection ) <EOL> @ contextlib . contextmanager <EOL> def begin ( self ) : <EOL> conn = self . connect ( ) <EOL> trans = conn . begin ( ) <EOL> try : <EOL> yield conn <EOL> except Exception : <EOL> trans . rollback ( ) <EOL> else : <EOL> trans . commit ( ) <EOL> class NonCommittingConnection ( NonCommittingConnectable ) : <EOL> """<STR_LIT>""" <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def begin ( self ) : <EOL> return NonCommittingTransaction ( self , self . connection . begin ( ) ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , * arg ) : <EOL> pass <EOL> class NonCommittingTransaction ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , provisioned , transaction ) : <EOL> self . provisioned = provisioned <EOL> self . transaction = transaction <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , type , value , traceback ) : <EOL> if type is None : <EOL> try : <EOL> self . commit ( ) <EOL> except Exception : <EOL> self . rollback ( ) <EOL> raise <EOL> else : <EOL> self . rollback ( ) <EOL> def commit ( self ) : <EOL> self . transaction . commit ( ) <EOL> def rollback ( self ) : <EOL> self . transaction . rollback ( ) <EOL> self . provisioned . _restart_nested ( ) </s>
