<s> import logging <EOL> from flask import request , flash , abort , Response <EOL> from flask_admin import expose <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 . _compat import iteritems , string_types <EOL> import mongoengine <EOL> import gridfs <EOL> from mongoengine . connection import get_db <EOL> from bson . objectid import ObjectId <EOL> from flask_admin . actions import action <EOL> from . filters import FilterConverter , BaseMongoEngineFilter <EOL> from . form import get_form , CustomModelConverter <EOL> from . typefmt import DEFAULT_FORMATTERS <EOL> from . tools import parse_like_term <EOL> from . helpers import format_error <EOL> from . ajax import process_ajax_references , create_ajax_loader <EOL> from . subdoc import convert_subdocuments <EOL> log = logging . getLogger ( "<STR_LIT>" ) <EOL> SORTABLE_FIELDS = set ( ( <EOL> mongoengine . StringField , <EOL> mongoengine . IntField , <EOL> mongoengine . FloatField , <EOL> mongoengine . BooleanField , <EOL> mongoengine . DateTimeField , <EOL> mongoengine . ComplexDateTimeField , <EOL> mongoengine . ObjectIdField , <EOL> mongoengine . DecimalField , <EOL> mongoengine . ReferenceField , <EOL> mongoengine . EmailField , <EOL> mongoengine . UUIDField , <EOL> mongoengine . URLField <EOL> ) ) <EOL> class ModelView ( BaseModelView ) : <EOL> """<STR_LIT>""" <EOL> column_filters = None <EOL> """<STR_LIT>""" <EOL> model_form_converter = CustomModelConverter <EOL> """<STR_LIT>""" <EOL> object_id_converter = ObjectId <EOL> """<STR_LIT>""" <EOL> filter_converter = FilterConverter ( ) <EOL> """<STR_LIT>""" <EOL> column_type_formatters = DEFAULT_FORMATTERS <EOL> """<STR_LIT>""" <EOL> allowed_search_types = ( mongoengine . StringField , <EOL> mongoengine . URLField , <EOL> mongoengine . EmailField ) <EOL> """<STR_LIT>""" <EOL> form_subdocuments = None <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , model , name = None , <EOL> 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 . _search_fields = [ ] <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> def _refresh_cache ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . form_subdocuments is None : <EOL> self . form_subdocuments = { } <EOL> self . _form_subdocuments = convert_subdocuments ( self . form_subdocuments ) <EOL> super ( ModelView , self ) . _refresh_cache ( ) <EOL> def _process_ajax_references ( self ) : <EOL> """<STR_LIT>""" <EOL> references = super ( ModelView , self ) . _process_ajax_references ( ) <EOL> return process_ajax_references ( references , self ) <EOL> def _get_model_fields ( self , model = None ) : <EOL> """<STR_LIT>""" <EOL> if model is None : <EOL> model = self . model <EOL> return sorted ( iteritems ( model . _fields ) , key = lambda n : n [ <NUM_LIT:1> ] . creation_counter ) <EOL> def scaffold_pk ( self ) : <EOL> return '<STR_LIT:id>' <EOL> def get_pk_value ( self , model ) : <EOL> """<STR_LIT>""" <EOL> return model . pk <EOL> def scaffold_list_columns ( self ) : <EOL> """<STR_LIT>""" <EOL> columns = [ ] <EOL> for n , f in self . _get_model_fields ( ) : <EOL> field_class = type ( f ) <EOL> if ( field_class == mongoengine . ListField and <EOL> isinstance ( f . field , mongoengine . EmbeddedDocumentField ) ) : <EOL> continue <EOL> if field_class == mongoengine . EmbeddedDocumentField : <EOL> continue <EOL> if self . column_display_pk or field_class != mongoengine . ObjectIdField : <EOL> columns . append ( n ) <EOL> return columns <EOL> def scaffold_sortable_columns ( self ) : <EOL> """<STR_LIT>""" <EOL> columns = { } <EOL> for n , f in self . _get_model_fields ( ) : <EOL> if type ( f ) in SORTABLE_FIELDS : <EOL> if self . column_display_pk or type ( f ) != mongoengine . ObjectIdField : <EOL> columns [ n ] = f <EOL> return columns <EOL> def init_search ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . column_searchable_list : <EOL> for p in self . column_searchable_list : <EOL> if isinstance ( p , string_types ) : <EOL> p = self . model . _fields . get ( p ) <EOL> if p is None : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> field_type = type ( p ) <EOL> if ( field_type not in self . allowed_search_types ) : <EOL> raise Exception ( '<STR_LIT>' + <EOL> '<STR_LIT>' % p ) <EOL> self . _search_fields . append ( p ) <EOL> return bool ( self . _search_fields ) <EOL> def scaffold_filters ( self , name ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( name , string_types ) : <EOL> attr = self . model . _fields . get ( name ) <EOL> else : <EOL> attr = name <EOL> if attr is None : <EOL> raise Exception ( '<STR_LIT>' % name ) <EOL> visible_name = None <EOL> if not isinstance ( name , string_types ) : <EOL> visible_name = self . get_column_name ( attr . name ) <EOL> if not visible_name : <EOL> visible_name = self . get_column_name ( name ) <EOL> type_name = type ( attr ) . __name__ <EOL> flt = self . filter_converter . convert ( type_name , <EOL> attr , <EOL> visible_name ) <EOL> return flt <EOL> def is_valid_filter ( self , filter ) : <EOL> """<STR_LIT>""" <EOL> return isinstance ( filter , BaseMongoEngineFilter ) <EOL> def scaffold_form ( self ) : <EOL> """<STR_LIT>""" <EOL> form_class = get_form ( self . model , <EOL> self . model_form_converter ( self ) , <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> return form_class <EOL> def scaffold_list_form ( self , widget = None , validators = None ) : <EOL> """<STR_LIT>""" <EOL> form_class = get_form ( self . model , <EOL> self . model_form_converter ( self ) , <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 _create_ajax_loader ( self , name , opts ) : <EOL> return create_ajax_loader ( self . model , name , name , opts ) <EOL> def get_query ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . model . objects <EOL> def _search ( self , query , search_term ) : <EOL> op , term = parse_like_term ( search_term ) <EOL> criteria = None <EOL> for field in self . _search_fields : <EOL> flt = { '<STR_LIT>' % ( field . name , op ) : term } <EOL> q = mongoengine . Q ( ** flt ) <EOL> if criteria is None : <EOL> criteria = q <EOL> else : <EOL> criteria |= q <EOL> return query . filter ( criteria ) <EOL> def get_list ( self , page , sort_column , sort_desc , search , filters , <EOL> execute = True , page_size = None ) : <EOL> """<STR_LIT>""" <EOL> query = self . get_query ( ) <EOL> if self . _filters : <EOL> for flt , flt_name , value in filters : <EOL> f = self . _filters [ flt ] <EOL> query = f . apply ( query , f . clean ( value ) ) <EOL> if self . _search_supported and search : <EOL> query = self . _search ( query , search ) <EOL> count = query . count ( ) if not self . simple_list_pager else None <EOL> if sort_column : <EOL> query = query . order_by ( '<STR_LIT>' % ( '<STR_LIT:->' if sort_desc else '<STR_LIT>' , sort_column ) ) <EOL> else : <EOL> order = self . _get_default_order ( ) <EOL> if order : <EOL> query = query . order_by ( '<STR_LIT>' % ( '<STR_LIT:->' if order [ <NUM_LIT:1> ] else '<STR_LIT>' , order [ <NUM_LIT:0> ] ) ) <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 . skip ( page * page_size ) <EOL> if execute : <EOL> query = query . all ( ) <EOL> return count , query <EOL> def get_one ( self , id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . get_query ( ) . filter ( pk = id ) . first ( ) <EOL> except mongoengine . ValidationError as ex : <EOL> flash ( gettext ( '<STR_LIT>' , <EOL> error = format_error ( ex ) ) , <EOL> '<STR_LIT:error>' ) <EOL> return None <EOL> def create_model ( self , form ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> model = self . model ( ) <EOL> form . populate_obj ( model ) <EOL> self . _on_model_change ( form , model , True ) <EOL> model . save ( ) <EOL> except Exception as ex : <EOL> if not self . handle_view_exception ( ex ) : <EOL> flash ( gettext ( '<STR_LIT>' , <EOL> error = format_error ( ex ) ) , <EOL> '<STR_LIT:error>' ) <EOL> log . exception ( '<STR_LIT>' ) <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> model . save ( ) <EOL> except Exception as ex : <EOL> if not self . handle_view_exception ( ex ) : <EOL> flash ( gettext ( '<STR_LIT>' , <EOL> error = format_error ( ex ) ) , <EOL> '<STR_LIT:error>' ) <EOL> log . exception ( '<STR_LIT>' ) <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> model . delete ( ) <EOL> except Exception as ex : <EOL> if not self . handle_view_exception ( ex ) : <EOL> flash ( gettext ( '<STR_LIT>' , <EOL> error = format_error ( ex ) ) , <EOL> '<STR_LIT:error>' ) <EOL> log . exception ( '<STR_LIT>' ) <EOL> return False <EOL> else : <EOL> self . after_model_delete ( model ) <EOL> return True <EOL> @ expose ( '<STR_LIT>' ) <EOL> def api_file_view ( self ) : <EOL> pk = request . args . get ( '<STR_LIT:id>' ) <EOL> coll = request . args . get ( '<STR_LIT>' ) <EOL> db = request . args . get ( '<STR_LIT>' , '<STR_LIT:default>' ) <EOL> if not pk or not coll or not db : <EOL> abort ( <NUM_LIT> ) <EOL> fs = gridfs . GridFS ( get_db ( db ) , coll ) <EOL> data = fs . get ( self . object_id_converter ( pk ) ) <EOL> if not data : <EOL> abort ( <NUM_LIT> ) <EOL> return Response ( data . read ( ) , <EOL> content_type = data . content_type , <EOL> headers = { <EOL> '<STR_LIT>' : data . length <EOL> } ) <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> count = <NUM_LIT:0> <EOL> all_ids = [ self . object_id_converter ( pk ) for pk in ids ] <EOL> for obj in self . get_query ( ) . in_bulk ( all_ids ) . values ( ) : <EOL> count += self . delete_model ( obj ) <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> flash ( gettext ( '<STR_LIT>' , error = str ( ex ) ) , <EOL> '<STR_LIT:error>' ) </s>
