<s> import logging <EOL> import warnings <EOL> import inspect <EOL> from sqlalchemy . orm . attributes import InstrumentedAttribute <EOL> from sqlalchemy . orm import joinedload , aliased <EOL> from sqlalchemy . sql . expression import desc , ColumnElement <EOL> from sqlalchemy import Boolean , Table , func , or_ <EOL> from sqlalchemy . exc import IntegrityError <EOL> from sqlalchemy . sql . expression import cast <EOL> from sqlalchemy import Unicode <EOL> from flask import flash <EOL> from flask_admin . _compat import string_types , text_type <EOL> from flask_admin . babel import gettext , ngettext , lazy_gettext <EOL> from flask_admin . model import BaseModelView <EOL> from flask_admin . model . form import create_editable_list_form <EOL> from flask_admin . actions import action <EOL> from flask_admin . _backwards import ObsoleteAttr <EOL> from flask_admin . contrib . sqla import form , filters as sqla_filters , tools <EOL> from . typefmt import DEFAULT_FORMATTERS <EOL> from . ajax import create_ajax_loader <EOL> log = logging . getLogger ( "<STR_LIT>" ) <EOL> class ModelView ( BaseModelView ) : <EOL> """<STR_LIT>""" <EOL> column_auto_select_related = ObsoleteAttr ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> True ) <EOL> """<STR_LIT>""" <EOL> column_select_related_list = ObsoleteAttr ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> None ) <EOL> """<STR_LIT>""" <EOL> column_display_all_relations = ObsoleteAttr ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> False ) <EOL> """<STR_LIT>""" <EOL> column_searchable_list = ObsoleteAttr ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> None ) <EOL> """<STR_LIT>""" <EOL> column_filters = None <EOL> """<STR_LIT>""" <EOL> model_form_converter = form . AdminModelConverter <EOL> """<STR_LIT>""" <EOL> inline_model_form_converter = form . InlineModelConverter <EOL> """<STR_LIT>""" <EOL> filter_converter = sqla_filters . FilterConverter ( ) <EOL> """<STR_LIT>""" <EOL> fast_mass_delete = False <EOL> """<STR_LIT>""" <EOL> inline_models = None <EOL> """<STR_LIT>""" <EOL> column_type_formatters = DEFAULT_FORMATTERS <EOL> form_choices = None <EOL> """<STR_LIT>""" <EOL> form_optional_types = ( Boolean , ) <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , model , session , <EOL> name = None , category = None , endpoint = None , url = None , static_folder = None , <EOL> menu_class_name = None , menu_icon_type = None , menu_icon_value = None ) : <EOL> """<STR_LIT>""" <EOL> self . session = session <EOL> self . _search_fields = None <EOL> self . _filter_joins = dict ( ) <EOL> self . _sortable_joins = dict ( ) <EOL> if self . form_choices is None : <EOL> self . form_choices = { } <EOL> super ( ModelView , self ) . __init__ ( model , name , category , endpoint , url , static_folder , <EOL> menu_class_name = menu_class_name , <EOL> menu_icon_type = menu_icon_type , <EOL> menu_icon_value = menu_icon_value ) <EOL> self . _primary_key = self . scaffold_pk ( ) <EOL> if self . _primary_key is None : <EOL> raise Exception ( '<STR_LIT>' % self . model . __name__ ) <EOL> if not self . column_select_related_list : <EOL> self . _auto_joins = self . scaffold_auto_joins ( ) <EOL> else : <EOL> self . _auto_joins = self . column_select_related_list <EOL> def _get_model_iterator ( self , model = None ) : <EOL> """<STR_LIT>""" <EOL> if model is None : <EOL> model = self . model <EOL> return model . _sa_class_manager . mapper . iterate_properties <EOL> def _apply_path_joins ( self , query , joins , path , inner_join = True ) : <EOL> """<STR_LIT>""" <EOL> last = None <EOL> if path : <EOL> for item in path : <EOL> key = ( inner_join , item ) <EOL> alias = joins . get ( key ) <EOL> if key not in joins : <EOL> if not isinstance ( item , Table ) : <EOL> alias = aliased ( item . property . mapper . class_ ) <EOL> fn = query . join if inner_join else query . outerjoin <EOL> if last is None : <EOL> query = fn ( item ) if alias is None else fn ( alias , item ) <EOL> else : <EOL> prop = getattr ( last , item . key ) <EOL> query = fn ( prop ) if alias is None else fn ( alias , prop ) <EOL> joins [ key ] = alias <EOL> last = alias <EOL> return query , joins , last <EOL> def scaffold_pk ( self ) : <EOL> """<STR_LIT>""" <EOL> return tools . get_primary_key ( self . model ) <EOL> def get_pk_value ( self , model ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( self . _primary_key , tuple ) : <EOL> return tools . iterencode ( getattr ( model , attr ) for attr in self . _primary_key ) <EOL> else : <EOL> return tools . escape ( getattr ( model , self . _primary_key ) ) <EOL> def scaffold_list_columns ( self ) : <EOL> """<STR_LIT>""" <EOL> columns = [ ] <EOL> for p in self . _get_model_iterator ( ) : <EOL> if hasattr ( p , '<STR_LIT>' ) : <EOL> if self . column_display_all_relations or p . direction . name == '<STR_LIT>' : <EOL> columns . append ( p . key ) <EOL> elif hasattr ( p , '<STR_LIT>' ) : <EOL> if len ( p . columns ) > <NUM_LIT:1> : <EOL> filtered = tools . filter_foreign_columns ( self . model . __table__ , p . columns ) <EOL> if len ( filtered ) > <NUM_LIT:1> : <EOL> warnings . warn ( '<STR_LIT>' % ( self . model , p . key ) ) <EOL> continue <EOL> column = filtered [ <NUM_LIT:0> ] <EOL> else : <EOL> column = p . columns [ <NUM_LIT:0> ] <EOL> if column . foreign_keys : <EOL> continue <EOL> if not self . column_display_pk and column . primary_key : <EOL> continue <EOL> columns . append ( p . key ) <EOL> return columns <EOL> def scaffold_sortable_columns ( self ) : <EOL> """<STR_LIT>""" <EOL> columns = dict ( ) <EOL> for p in self . _get_model_iterator ( ) : <EOL> if hasattr ( p , '<STR_LIT>' ) : <EOL> if len ( p . columns ) > <NUM_LIT:1> : <EOL> continue <EOL> column = p . columns [ <NUM_LIT:0> ] <EOL> if column . foreign_keys : <EOL> continue <EOL> if not self . column_display_pk and column . primary_key : <EOL> continue <EOL> columns [ p . key ] = column <EOL> return columns <EOL> def get_sortable_columns ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _sortable_joins = dict ( ) <EOL> if self . column_sortable_list is None : <EOL> return self . scaffold_sortable_columns ( ) <EOL> else : <EOL> result = dict ( ) <EOL> for c in self . column_sortable_list : <EOL> if isinstance ( c , tuple ) : <EOL> column , path = tools . get_field_with_path ( self . model , c [ <NUM_LIT:1> ] ) <EOL> column_name = c [ <NUM_LIT:0> ] <EOL> else : <EOL> column , path = tools . get_field_with_path ( self . model , c ) <EOL> column_name = text_type ( c ) <EOL> if path and hasattr ( path [ <NUM_LIT:0> ] , '<STR_LIT>' ) : <EOL> self . _sortable_joins [ column_name ] = path <EOL> elif path : <EOL> raise Exception ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( c ) ) <EOL> else : <EOL> if getattr ( column , '<STR_LIT:key>' , None ) is not None : <EOL> column_name = column . key <EOL> else : <EOL> column_name = text_type ( c ) <EOL> result [ column_name ] = column <EOL> return result <EOL> def get_column_names ( self , only_columns , excluded_columns ) : <EOL> """<STR_LIT>""" <EOL> if excluded_columns : <EOL> only_columns = [ c for c in only_columns if c not in excluded_columns ] <EOL> formatted_columns = [ ] <EOL> for c in only_columns : <EOL> column , path = tools . get_field_with_path ( self . model , c ) <EOL> if path : <EOL> column_name = text_type ( c ) <EOL> else : <EOL> if getattr ( column , '<STR_LIT:key>' , None ) is not None : <EOL> column_name = column . key <EOL> else : <EOL> column_name = text_type ( c ) <EOL> visible_name = self . get_column_name ( column_name ) <EOL> formatted_columns . append ( ( column_name , visible_name ) ) <EOL> return formatted_columns <EOL> def init_search ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . column_searchable_list : <EOL> self . _search_fields = [ ] <EOL> for p in self . column_searchable_list : <EOL> attr , joins = tools . get_field_with_path ( self . model , p ) <EOL> if not attr : <EOL> raise Exception ( '<STR_LIT>' % p ) <EOL> for column in tools . get_columns_for_field ( attr ) : <EOL> self . _search_fields . append ( ( column , joins ) ) <EOL> return bool ( self . column_searchable_list ) <EOL> def scaffold_filters ( self , name ) : <EOL> """<STR_LIT>""" <EOL> attr , joins = tools . get_field_with_path ( self . model , name ) <EOL> if attr is None : <EOL> raise Exception ( '<STR_LIT>' % name ) <EOL> if hasattr ( attr , '<STR_LIT>' ) and hasattr ( attr . property , '<STR_LIT>' ) : <EOL> filters = [ ] <EOL> for p in self . _get_model_iterator ( attr . property . mapper . class_ ) : <EOL> if hasattr ( p , '<STR_LIT>' ) : <EOL> column = p . columns [ <NUM_LIT:0> ] <EOL> if column . foreign_keys or column . primary_key : <EOL> continue <EOL> visible_name = '<STR_LIT>' % ( self . get_column_name ( attr . prop . table . name ) , <EOL> self . get_column_name ( p . key ) ) <EOL> type_name = type ( column . type ) . __name__ <EOL> flt = self . filter_converter . convert ( type_name , <EOL> column , <EOL> visible_name ) <EOL> if flt : <EOL> table = column . table <EOL> if joins : <EOL> self . _filter_joins [ column ] = joins <EOL> elif tools . need_join ( self . model , table ) : <EOL> self . _filter_joins [ column ] = [ table ] <EOL> filters . extend ( flt ) <EOL> return filters <EOL> else : <EOL> is_hybrid_property = isinstance ( attr , ColumnElement ) <EOL> if is_hybrid_property : <EOL> column = attr <EOL> else : <EOL> columns = tools . 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 ( not is_hybrid_property and tools . need_join ( self . model , column . table ) and <EOL> name not in self . column_labels ) : <EOL> visible_name = '<STR_LIT>' % ( <EOL> self . get_column_name ( column . table . name ) , <EOL> self . get_column_name ( column . name ) <EOL> ) <EOL> else : <EOL> if not isinstance ( name , string_types ) : <EOL> visible_name = self . get_column_name ( name . property . key ) <EOL> else : <EOL> visible_name = self . get_column_name ( name ) <EOL> type_name = type ( column . type ) . __name__ <EOL> flt = self . filter_converter . convert ( <EOL> type_name , <EOL> column , <EOL> visible_name , <EOL> options = self . column_choices . get ( name ) , <EOL> ) <EOL> if joins : <EOL> self . _filter_joins [ column ] = joins <EOL> elif not is_hybrid_property and tools . need_join ( self . model , column . table ) : <EOL> self . _filter_joins [ column ] = [ column . table ] <EOL> return flt <EOL> def handle_filter ( self , filter ) : <EOL> if isinstance ( filter , sqla_filters . BaseSQLAFilter ) : <EOL> column = filter . column <EOL> if ( isinstance ( column , InstrumentedAttribute ) and <EOL> tools . need_join ( self . model , column . table ) ) : <EOL> self . _filter_joins [ column ] = [ column . table ] <EOL> return filter <EOL> def scaffold_form ( self ) : <EOL> """<STR_LIT>""" <EOL> converter = self . model_form_converter ( self . session , self ) <EOL> form_class = form . get_form ( self . model , converter , <EOL> base_class = self . form_base_class , <EOL> only = self . form_columns , <EOL> exclude = self . form_excluded_columns , <EOL> field_args = self . form_args , <EOL> extra_fields = self . form_extra_fields ) <EOL> if self . inline_models : <EOL> form_class = self . scaffold_inline_form_models ( form_class ) <EOL> return form_class <EOL> def scaffold_list_form ( self , widget = None , validators = None ) : <EOL> """<STR_LIT>""" <EOL> converter = self . model_form_converter ( self . session , self ) <EOL> form_class = form . get_form ( self . model , converter , <EOL> base_class = self . form_base_class , <EOL> only = self . column_editable_list , <EOL> field_args = validators ) <EOL> return create_editable_list_form ( self . form_base_class , form_class , <EOL> widget ) <EOL> def scaffold_inline_form_models ( self , form_class ) : <EOL> """<STR_LIT>""" <EOL> inline_converter = self . inline_model_form_converter ( self . session , <EOL> self , <EOL> self . model_form_converter ) <EOL> for m in self . inline_models : <EOL> form_class = inline_converter . contribute ( self . model , form_class , m ) <EOL> return form_class <EOL> def scaffold_auto_joins ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . column_auto_select_related : <EOL> return [ ] <EOL> relations = set ( ) <EOL> for p in self . _get_model_iterator ( ) : <EOL> if hasattr ( p , '<STR_LIT>' ) : <EOL> if p . mapper . class_ == self . model : <EOL> continue <EOL> if p . direction . name in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> relations . add ( p . key ) <EOL> joined = [ ] <EOL> for prop , name in self . _list_columns : <EOL> if prop in relations : <EOL> joined . append ( getattr ( self . model , prop ) ) <EOL> return joined <EOL> def _create_ajax_loader ( self , name , options ) : <EOL> return create_ajax_loader ( self . model , self . session , name , name , options ) <EOL> def get_query ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . session . query ( self . model ) <EOL> def get_count_query ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . session . query ( func . count ( '<STR_LIT:*>' ) ) . select_from ( self . model ) <EOL> def _order_by ( self , query , joins , sort_joins , sort_field , sort_desc ) : <EOL> """<STR_LIT>""" <EOL> if sort_field is not None : <EOL> query , joins , alias = self . _apply_path_joins ( query , joins , sort_joins , inner_join = False ) <EOL> column = sort_field if alias is None else getattr ( alias , sort_field . key ) <EOL> if sort_desc : <EOL> query = query . order_by ( desc ( column ) ) <EOL> else : <EOL> query = query . order_by ( column ) <EOL> return query , joins <EOL> def _get_default_order ( self ) : <EOL> order = super ( ModelView , self ) . _get_default_order ( ) <EOL> if order is not None : <EOL> field , direction = order <EOL> attr , joins = tools . get_field_with_path ( self . model , field ) <EOL> return attr , joins , direction <EOL> return None <EOL> def _apply_sorting ( self , query , joins , sort_column , sort_desc ) : <EOL> if sort_column is not None : <EOL> if sort_column in self . _sortable_columns : <EOL> sort_field = self . _sortable_columns [ sort_column ] <EOL> sort_joins = self . _sortable_joins . get ( sort_column ) <EOL> query , joins = self . _order_by ( query , joins , sort_joins , sort_field , sort_desc ) <EOL> else : <EOL> order = self . _get_default_order ( ) <EOL> if order : <EOL> sort_field , sort_joins , sort_desc = order <EOL> query , joins = self . _order_by ( query , joins , sort_joins , sort_field , sort_desc ) <EOL> return query , joins <EOL> def _apply_search ( self , query , count_query , joins , count_joins , search ) : <EOL> """<STR_LIT>""" <EOL> terms = search . split ( '<STR_LIT:U+0020>' ) <EOL> for term in terms : <EOL> if not term : <EOL> continue <EOL> stmt = tools . parse_like_term ( term ) <EOL> filter_stmt = [ ] <EOL> count_filter_stmt = [ ] <EOL> for field , path in self . _search_fields : <EOL> query , joins , alias = self . _apply_path_joins ( query , joins , path , inner_join = False ) <EOL> count_alias = None <EOL> if count_query is not None : <EOL> count_query , count_joins , count_alias = self . _apply_path_joins ( count_query , <EOL> count_joins , <EOL> path , <EOL> inner_join = False ) <EOL> column = field if alias is None else getattr ( alias , field . key ) <EOL> filter_stmt . append ( cast ( column , Unicode ) . ilike ( stmt ) ) <EOL> if count_filter_stmt is not None : <EOL> column = field if count_alias is None else getattr ( count_alias , field . key ) <EOL> count_filter_stmt . append ( cast ( column , Unicode ) . ilike ( stmt ) ) <EOL> query = query . filter ( or_ ( * filter_stmt ) ) <EOL> if count_query is not None : <EOL> count_query = count_query . filter ( or_ ( * count_filter_stmt ) ) <EOL> return query , count_query , joins , count_joins <EOL> def _apply_filters ( self , query , count_query , joins , count_joins , filters ) : <EOL> for idx , flt_name , value in filters : <EOL> flt = self . _filters [ idx ] <EOL> alias = None <EOL> count_alias = None <EOL> if isinstance ( flt , sqla_filters . BaseSQLAFilter ) : <EOL> path = self . _filter_joins . get ( flt . column , [ ] ) <EOL> query , joins , alias = self . _apply_path_joins ( query , joins , path , inner_join = False ) <EOL> if count_query is not None : <EOL> count_query , count_joins , count_alias = self . _apply_path_joins ( <EOL> count_query , <EOL> count_joins , <EOL> path , <EOL> inner_join = False ) <EOL> clean_value = flt . clean ( value ) <EOL> try : <EOL> query = flt . apply ( query , clean_value , alias ) <EOL> except TypeError : <EOL> spec = inspect . getargspec ( flt . apply ) <EOL> if len ( spec . args ) == <NUM_LIT:3> : <EOL> warnings . warn ( '<STR_LIT>' % repr ( flt ) ) <EOL> else : <EOL> raise <EOL> query = flt . apply ( query , clean_value ) <EOL> if count_query is not None : <EOL> try : <EOL> count_query = flt . apply ( count_query , clean_value , count_alias ) <EOL> except TypeError : <EOL> count_query = flt . apply ( count_query , clean_value ) <EOL> return query , count_query , joins , count_joins <EOL> def _apply_pagination ( self , query , page , page_size ) : <EOL> if page_size is None : <EOL> page_size = self . page_size <EOL> if page_size : <EOL> query = query . limit ( page_size ) <EOL> if page and page_size : <EOL> query = query . offset ( page * page_size ) <EOL> return query <EOL> def get_list ( self , page , sort_column , sort_desc , search , filters , <EOL> execute = True , page_size = None ) : <EOL> """<STR_LIT>""" <EOL> joins = { } <EOL> count_joins = { } <EOL> query = self . get_query ( ) <EOL> count_query = self . get_count_query ( ) if not self . simple_list_pager else None <EOL> if hasattr ( query , '<STR_LIT>' ) : <EOL> for entity in query . _join_entities : <EOL> for table in entity . tables : <EOL> joins [ table ] = None <EOL> if self . _search_supported and search : <EOL> query , count_query , joins , count_joins = self . _apply_search ( query , <EOL> count_query , <EOL> joins , <EOL> count_joins , <EOL> search ) <EOL> if filters and self . _filters : <EOL> query , count_query , joins , count_joins = self . _apply_filters ( query , <EOL> count_query , <EOL> joins , <EOL> count_joins , <EOL> filters ) <EOL> count = count_query . scalar ( ) if count_query else None <EOL> for j in self . _auto_joins : <EOL> query = query . options ( joinedload ( j ) ) <EOL> query , joins = self . _apply_sorting ( query , joins , sort_column , sort_desc ) <EOL> query = self . _apply_pagination ( query , page , page_size ) <EOL> if execute : <EOL> query = query . all ( ) <EOL> return count , query <EOL> def get_one ( self , id ) : <EOL> """<STR_LIT>""" <EOL> return self . session . query ( self . model ) . get ( tools . iterdecode ( id ) ) <EOL> def handle_view_exception ( self , exc ) : <EOL> if isinstance ( exc , IntegrityError ) : <EOL> flash ( gettext ( '<STR_LIT>' , message = text_type ( exc ) ) , '<STR_LIT:error>' ) <EOL> return True <EOL> return super ( ModelView , self ) . handle_view_exception ( exc ) <EOL> def create_model ( self , form ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> model = self . model ( ) <EOL> form . populate_obj ( model ) <EOL> self . session . add ( model ) <EOL> self . _on_model_change ( form , model , True ) <EOL> self . session . commit ( ) <EOL> except Exception as ex : <EOL> if not self . handle_view_exception ( ex ) : <EOL> flash ( gettext ( '<STR_LIT>' , error = str ( ex ) ) , '<STR_LIT:error>' ) <EOL> log . exception ( '<STR_LIT>' ) <EOL> self . session . rollback ( ) <EOL> return False <EOL> else : <EOL> self . after_model_change ( form , model , True ) <EOL> return model <EOL> def update_model ( self , form , model ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> form . populate_obj ( model ) <EOL> self . _on_model_change ( form , model , False ) <EOL> self . session . commit ( ) <EOL> except Exception as ex : <EOL> if not self . handle_view_exception ( ex ) : <EOL> flash ( gettext ( '<STR_LIT>' , error = str ( ex ) ) , '<STR_LIT:error>' ) <EOL> log . exception ( '<STR_LIT>' ) <EOL> self . session . rollback ( ) <EOL> return False <EOL> else : <EOL> self . after_model_change ( form , model , False ) <EOL> return True <EOL> def delete_model ( self , model ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . on_model_delete ( model ) <EOL> self . session . flush ( ) <EOL> self . session . delete ( model ) <EOL> self . session . commit ( ) <EOL> except Exception as ex : <EOL> if not self . handle_view_exception ( ex ) : <EOL> flash ( gettext ( '<STR_LIT>' , error = str ( ex ) ) , '<STR_LIT:error>' ) <EOL> log . exception ( '<STR_LIT>' ) <EOL> self . session . rollback ( ) <EOL> return False <EOL> else : <EOL> self . after_model_delete ( model ) <EOL> return True <EOL> def is_action_allowed ( self , name ) : <EOL> if name == '<STR_LIT>' and not self . can_delete : <EOL> return False <EOL> return super ( ModelView , self ) . is_action_allowed ( name ) <EOL> @ action ( '<STR_LIT>' , <EOL> lazy_gettext ( '<STR_LIT>' ) , <EOL> lazy_gettext ( '<STR_LIT>' ) ) <EOL> def action_delete ( self , ids ) : <EOL> try : <EOL> query = tools . get_query_for_ids ( self . get_query ( ) , self . model , ids ) <EOL> if self . fast_mass_delete : <EOL> count = query . delete ( synchronize_session = False ) <EOL> else : <EOL> count = <NUM_LIT:0> <EOL> for m in query . all ( ) : <EOL> if self . delete_model ( m ) : <EOL> count += <NUM_LIT:1> <EOL> self . session . commit ( ) <EOL> flash ( ngettext ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> count , <EOL> count = count ) ) <EOL> except Exception as ex : <EOL> if not self . handle_view_exception ( ex ) : <EOL> raise <EOL> flash ( gettext ( '<STR_LIT>' , error = str ( ex ) ) , '<STR_LIT:error>' ) </s>
