<s> from sqlalchemy import tuple_ , or_ , and_ <EOL> from sqlalchemy . sql . operators import eq <EOL> from sqlalchemy . exc import DBAPIError <EOL> from sqlalchemy . orm . attributes import InstrumentedAttribute <EOL> from flask_admin . _compat import filter_list , string_types <EOL> from flask_admin . tools import iterencode , iterdecode , escape <EOL> def parse_like_term ( term ) : <EOL> if term . startswith ( '<STR_LIT>' ) : <EOL> stmt = '<STR_LIT>' % term [ <NUM_LIT:1> : ] <EOL> elif term . startswith ( '<STR_LIT:=>' ) : <EOL> stmt = term [ <NUM_LIT:1> : ] <EOL> else : <EOL> stmt = '<STR_LIT>' % term <EOL> return stmt <EOL> def filter_foreign_columns ( base_table , columns ) : <EOL> """<STR_LIT>""" <EOL> return filter_list ( lambda c : c . table == base_table , columns ) <EOL> def get_primary_key ( model ) : <EOL> """<STR_LIT>""" <EOL> mapper = model . _sa_class_manager . mapper <EOL> pks = [ mapper . get_property_by_column ( c ) . key for c in mapper . primary_key ] <EOL> if len ( pks ) == <NUM_LIT:1> : <EOL> return pks [ <NUM_LIT:0> ] <EOL> elif len ( pks ) > <NUM_LIT:1> : <EOL> return tuple ( pks ) <EOL> else : <EOL> return None <EOL> def has_multiple_pks ( model ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( model , '<STR_LIT>' ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> return len ( model . _sa_class_manager . mapper . primary_key ) > <NUM_LIT:1> <EOL> def tuple_operator_in ( model_pk , ids ) : <EOL> """<STR_LIT>""" <EOL> l = [ ] <EOL> for id in ids : <EOL> k = [ ] <EOL> for i in range ( len ( model_pk ) ) : <EOL> k . append ( eq ( model_pk [ i ] , id [ i ] ) ) <EOL> l . append ( and_ ( * k ) ) <EOL> if len ( l ) >= <NUM_LIT:1> : <EOL> return or_ ( * l ) <EOL> else : <EOL> return None <EOL> def get_query_for_ids ( modelquery , model , ids ) : <EOL> """<STR_LIT>""" <EOL> if has_multiple_pks ( model ) : <EOL> decoded_ids = [ iterdecode ( v ) for v in ids ] <EOL> model_pk = [ getattr ( model , name ) for name in get_primary_key ( model ) ] <EOL> try : <EOL> query = modelquery . filter ( tuple_ ( * model_pk ) . in_ ( decoded_ids ) ) <EOL> query . all ( ) <EOL> except DBAPIError : <EOL> query = modelquery . filter ( tuple_operator_in ( model_pk , decoded_ids ) ) <EOL> else : <EOL> model_pk = getattr ( model , get_primary_key ( model ) ) <EOL> query = modelquery . filter ( model_pk . in_ ( ids ) ) <EOL> return query <EOL> def get_columns_for_field ( field ) : <EOL> if ( not field or <EOL> not hasattr ( field , '<STR_LIT>' ) or <EOL> not hasattr ( field . property , '<STR_LIT>' ) or <EOL> not field . property . columns ) : <EOL> raise Exception ( '<STR_LIT>' % field ) <EOL> return field . property . columns <EOL> def need_join ( model , table ) : <EOL> """<STR_LIT>""" <EOL> return table not in model . _sa_class_manager . mapper . tables <EOL> def get_field_with_path ( model , name ) : <EOL> """<STR_LIT>""" <EOL> path = [ ] <EOL> if isinstance ( name , string_types ) : <EOL> current_model = model <EOL> for attribute in name . split ( '<STR_LIT:.>' ) : <EOL> value = getattr ( current_model , attribute ) <EOL> if ( hasattr ( value , '<STR_LIT>' ) and <EOL> hasattr ( value . property , '<STR_LIT>' ) ) : <EOL> current_model = value . property . mapper . class_ <EOL> table = current_model . __table__ <EOL> if need_join ( model , table ) : <EOL> path . append ( value ) <EOL> attr = value <EOL> else : <EOL> attr = name <EOL> if isinstance ( attr , InstrumentedAttribute ) : <EOL> columns = get_columns_for_field ( attr ) <EOL> if len ( columns ) > <NUM_LIT:1> : <EOL> raise Exception ( '<STR_LIT>' % name ) <EOL> column = columns [ <NUM_LIT:0> ] <EOL> if need_join ( model , column . table ) : <EOL> path . append ( column . table ) <EOL> return attr , path </s>
