classBaseAdmin:"""Base class for implementing Admin interface."""def__init__(self,title:str=_("Admin"),base_url:str="/admin",route_name:str="admin",logo_url:Optional[str]=None,login_logo_url:Optional[str]=None,templates_dir:str="templates",statics_dir:Optional[str]=None,index_view:Optional[CustomView]=None,auth_provider:Optional[BaseAuthProvider]=None,middlewares:Optional[Sequence[Middleware]]=None,debug:bool=False,i18n_config:Optional[I18nConfig]=None,favicon_url:Optional[str]=None,):""" Parameters: title: Admin title. base_url: Base URL for Admin interface. route_name: Mounted Admin name logo_url: URL of logo to be displayed instead of title. login_logo_url: If set, it will be used for login interface instead of logo_url. templates_dir: Templates dir for customisation statics_dir: Statics dir for customisation index_view: CustomView to use for index page. auth_provider: Authentication Provider middlewares: Starlette middlewares i18n_config: i18n configuration favicon_url: URL of favicon. """self.title=titleself.base_url=base_urlself.route_name=route_nameself.logo_url=logo_urlself.login_logo_url=login_logo_urlself.favicon_url=favicon_urlself.templates_dir=templates_dirself.statics_dir=statics_dirself.auth_provider=auth_providerself.middlewares=list(middlewares)ifmiddlewaresisnotNoneelse[]self.index_view=(index_viewif(index_viewisnotNone)elseCustomView("",add_to_menu=False))self._views:List[BaseView]=[]self._models:List[BaseModelView]=[]self.routes:List[Union[Route,Mount]]=[]self.debug=debugself.i18n_config=i18n_configself._setup_templates()self.init_locale()self.init_auth()self.init_routes()defadd_view(self,view:Union[Type[BaseView],BaseView])->None:""" Add View to the Admin interface. """view_instance=viewifisinstance(view,BaseView)elseview()self._views.append(view_instance)self.setup_view(view_instance)defcustom_render_js(self,request:Request)->Optional[str]:""" Override this function to provide a link to custom js to override the global `render` object in javascript which is use to render fields in list page. Args: request: Starlette Request """returnNonedefinit_locale(self)->None:ifself.i18n_configisnotNone:try:importbabel# noqaexceptImportErroraserr:raiseImportError("'babel' package is required to use i18n features.""Install it with `pip install starlette-admin[i18n]`")fromerrself.middlewares.insert(0,Middleware(LocaleMiddleware,i18n_config=self.i18n_config))definit_auth(self)->None:ifself.auth_providerisnotNone:self.auth_provider.setup_admin(self)definit_routes(self)->None:statics=StaticFiles(directory=self.statics_dir,packages=["starlette_admin"])self.routes.extend([Mount("/statics",app=statics,name="statics"),Route(self.index_view.path,self._render_custom_view(self.index_view),methods=self.index_view.methods,name="index",),Route("/api/{identity}",self._render_api,methods=["GET"],name="api",),Route("/api/{identity}/action",self.handle_action,methods=["GET","POST"],name="action",),Route("/api/{identity}/row-action",self.handle_row_action,methods=["GET","POST"],name="row-action",),Route("/{identity}/list",self._render_list,methods=["GET"],name="list",),Route("/{identity}/detail/{pk}",self._render_detail,methods=["GET"],name="detail",),Route("/{identity}/create",self._render_create,methods=["GET","POST"],name="create",),Route("/{identity}/edit/{pk}",self._render_edit,methods=["GET","POST"],name="edit",),])ifself.index_view.add_to_menu:self._views.append(self.index_view)def_setup_templates(self)->None:env=Environment(loader=ChoiceLoader([FileSystemLoader(self.templates_dir),PackageLoader("starlette_admin","templates"),PrefixLoader({"@starlette-admin":PackageLoader("starlette_admin","templates"),}),]),extensions=["jinja2.ext.i18n"],)templates=Jinja2Templates(env=env)# globalstemplates.env.globals["views"]=self._viewstemplates.env.globals["app_title"]=self.titletemplates.env.globals["is_auth_enabled"]=self.auth_providerisnotNonetemplates.env.globals["__name__"]=self.route_nametemplates.env.globals["logo_url"]=self.logo_urltemplates.env.globals["login_logo_url"]=self.login_logo_urltemplates.env.globals["favicon_url"]=self.favicon_urltemplates.env.globals["custom_render_js"]=lambdar:self.custom_render_js(r)templates.env.globals["get_locale"]=get_localetemplates.env.globals["get_locale_display_name"]=get_locale_display_nametemplates.env.globals["i18n_config"]=self.i18n_configorI18nConfig()# filterstemplates.env.filters["is_custom_view"]=lambdar:isinstance(r,CustomView)templates.env.filters["is_link"]=lambdares:isinstance(res,Link)templates.env.filters["is_model"]=lambdares:isinstance(res,BaseModelView)templates.env.filters["is_dropdown"]=lambdares:isinstance(res,DropDown)templates.env.filters["get_admin_user"]=(self.auth_provider.get_admin_userifself.auth_providerelseNone)templates.env.filters["get_admin_config"]=(self.auth_provider.get_admin_configifself.auth_providerelseNone)templates.env.filters["tojson"]=lambdadata:json.dumps(data,default=str)templates.env.filters["file_icon"]=get_file_icontemplates.env.filters["to_model"]=(lambdaidentity:self._find_model_from_identity(identity))templates.env.filters["is_iter"]=lambdav:isinstance(v,(list,tuple))templates.env.filters["is_str"]=lambdav:isinstance(v,str)templates.env.filters["is_dict"]=lambdav:isinstance(v,dict)templates.env.filters["ra"]=lambdaa:RequestAction(a)# install i18ntemplates.env.install_gettext_callables(gettext,ngettext,True)# type: ignoreself.templates=templatesdefsetup_view(self,view:BaseView)->None:ifisinstance(view,DropDown):forsub_viewinview.views:self.setup_view(sub_view)elifisinstance(view,CustomView):self.routes.insert(0,Route(view.path,endpoint=self._render_custom_view(view),methods=view.methods,name=view.name,),)elifisinstance(view,BaseModelView):view._find_foreign_model=self._find_model_from_identityself._models.append(view)def_find_model_from_identity(self,identity:Optional[str])->BaseModelView:ifidentityisnotNone:formodelinself._models:ifmodel.identity==identity:returnmodelraiseHTTPException(HTTP_404_NOT_FOUND,_("Model with identity %(identity)s not found")%{"identity":identity},)def_render_custom_view(self,custom_view:CustomView)->Callable[[Request],Awaitable[Response]]:asyncdefwrapper(request:Request)->Response:ifnotcustom_view.is_accessible(request):raiseHTTPException(HTTP_403_FORBIDDEN)returnawaitcustom_view.render(request,self.templates)returnwrapperasyncdef_render_api(self,request:Request)->Response:identity=request.path_params.get("identity")model=self._find_model_from_identity(identity)ifnotmodel.is_accessible(request):returnJSONResponse(None,status_code=HTTP_403_FORBIDDEN)skip=int(request.query_params.get("skip")or"0")limit=int(request.query_params.get("limit")or"100")order_by=request.query_params.getlist("order_by")where=request.query_params.get("where")pks=request.query_params.getlist("pks")select2="select2"inrequest.query_paramsrequest.state.action=RequestAction.APIifselect2elseRequestAction.LISTiflen(pks)>0:items=awaitmodel.find_by_pks(request,pks)total=len(items)else:ifwhereisnotNone:try:where=json.loads(where)exceptJSONDecodeError:where=str(where)items=awaitmodel.find_all(request=request,skip=skip,limit=limit,where=where,order_by=order_by,)total=awaitmodel.count(request=request,where=where)serialized_items=[(awaitmodel.serialize(item,request,RequestAction.APIifselect2elseRequestAction.LIST,include_relationships=notselect2,include_select2=select2,))foriteminitems]ifnotselect2:# Add row actions for datatablesrow_actions=awaitmodel.get_all_row_actions(request)assertmodel.pk_attrforserialized_iteminserialized_items:serialized_item["_meta"]["rowActions"]=self.templates.get_template("row-actions.html").render(_actions=row_actions,display_type=model.row_actions_display_type,pk=serialized_item[model.pk_attr],request=request,model=model,)returnJSONResponse({"items":serialized_items,"total":total,})asyncdefhandle_action(self,request:Request)->Response:request.state.action=RequestAction.ACTIONtry:identity=request.path_params.get("identity")pks=request.query_params.getlist("pks")name=not_none(request.query_params.get("name"))model=self._find_model_from_identity(identity)ifnotmodel.is_accessible(request):raiseActionFailed("Forbidden")handler_return=awaitmodel.handle_action(request,pks,name)ifisinstance(handler_return,Response):returnhandler_returnreturnJSONResponse({"msg":handler_return})exceptActionFailedasexc:returnJSONResponse({"msg":exc.msg},status_code=HTTP_400_BAD_REQUEST)asyncdefhandle_row_action(self,request:Request)->Response:request.state.action=RequestAction.ROW_ACTIONtry:identity=request.path_params.get("identity")pk=request.query_params.get("pk")name=not_none(request.query_params.get("name"))model=self._find_model_from_identity(identity)ifnotmodel.is_accessible(request):raiseActionFailed("Forbidden")handler_return=awaitmodel.handle_row_action(request,pk,name)ifisinstance(handler_return,Response):returnhandler_returnreturnJSONResponse({"msg":handler_return})exceptActionFailedasexc:returnJSONResponse({"msg":exc.msg},status_code=HTTP_400_BAD_REQUEST)asyncdef_render_list(self,request:Request)->Response:request.state.action=RequestAction.LISTidentity=request.path_params.get("identity")model=self._find_model_from_identity(identity)ifnotmodel.is_accessible(request):raiseHTTPException(HTTP_403_FORBIDDEN)returnself.templates.TemplateResponse(request=request,name=model.list_template,context={"model":model,"title":model.title(request),"_actions":awaitmodel.get_all_actions(request),"__js_model__":awaitmodel._configs(request),},)asyncdef_render_detail(self,request:Request)->Response:request.state.action=RequestAction.DETAILidentity=request.path_params.get("identity")model=self._find_model_from_identity(identity)ifnotmodel.is_accessible(request)ornotmodel.can_view_details(request):raiseHTTPException(HTTP_403_FORBIDDEN)pk=request.path_params.get("pk")obj=awaitmodel.find_by_pk(request,pk)ifobjisNone:raiseHTTPException(HTTP_404_NOT_FOUND)returnself.templates.TemplateResponse(request=request,name=model.detail_template,context={"title":model.title(request),"model":model,"raw_obj":obj,"_actions":awaitmodel.get_all_row_actions(request),"obj":awaitmodel.serialize(obj,request,RequestAction.DETAIL),},)asyncdef_render_create(self,request:Request)->Response:request.state.action=RequestAction.CREATEidentity=request.path_params.get("identity")model=self._find_model_from_identity(identity)config={"title":model.title(request),"model":model}ifnotmodel.is_accessible(request)ornotmodel.can_create(request):raiseHTTPException(HTTP_403_FORBIDDEN)ifrequest.method=="GET":returnself.templates.TemplateResponse(request=request,name=model.create_template,context=config,)form=awaitrequest.form()dict_obj=awaitself.form_to_dict(request,form,model,RequestAction.CREATE)try:obj=awaitmodel.create(request,dict_obj)exceptFormValidationErrorasexc:config.update({"errors":exc.errors,"obj":dict_obj,})returnself.templates.TemplateResponse(request=request,name=model.create_template,context=config,status_code=HTTP_422_UNPROCESSABLE_ENTITY,)pk=awaitmodel.get_pk_value(request,obj)url=request.url_for(self.route_name+":list",identity=model.identity)ifform.get("_continue_editing",None)isnotNone:url=request.url_for(self.route_name+":edit",identity=model.identity,pk=pk)elifform.get("_add_another",None)isnotNone:url=request.urlreturnRedirectResponse(url,status_code=HTTP_303_SEE_OTHER)asyncdef_render_edit(self,request:Request)->Response:request.state.action=RequestAction.EDITidentity=request.path_params.get("identity")model=self._find_model_from_identity(identity)ifnotmodel.is_accessible(request)ornotmodel.can_edit(request):raiseHTTPException(HTTP_403_FORBIDDEN)pk=request.path_params.get("pk")obj=awaitmodel.find_by_pk(request,pk)ifobjisNone:raiseHTTPException(HTTP_404_NOT_FOUND)config={"title":model.title(request),"model":model,"raw_obj":obj,"obj":awaitmodel.serialize(obj,request,RequestAction.EDIT),}ifrequest.method=="GET":returnself.templates.TemplateResponse(request=request,name=model.edit_template,context=config,)form=awaitrequest.form()dict_obj=awaitself.form_to_dict(request,form,model,RequestAction.EDIT)try:obj=awaitmodel.edit(request,pk,dict_obj)exceptFormValidationErrorasexc:config.update({"errors":exc.errors,"obj":dict_obj,})returnself.templates.TemplateResponse(request=request,name=model.edit_template,context=config,status_code=HTTP_422_UNPROCESSABLE_ENTITY,)pk=awaitmodel.get_pk_value(request,obj)url=request.url_for(self.route_name+":list",identity=model.identity)ifform.get("_continue_editing",None)isnotNone:url=request.url_for(self.route_name+":edit",identity=model.identity,pk=pk)elifform.get("_add_another",None)isnotNone:url=request.url_for(self.route_name+":create",identity=model.identity)returnRedirectResponse(url,status_code=HTTP_303_SEE_OTHER)asyncdef_render_error(self,request:Request,exc:Exception=HTTPException(status_code=HTTP_500_INTERNAL_SERVER_ERROR),)->Response:assertisinstance(exc,HTTPException)returnself.templates.TemplateResponse(request=request,name="error.html",context={"exc":exc},status_code=exc.status_code,)asyncdefform_to_dict(self,request:Request,form_data:FormData,model:BaseModelView,action:RequestAction,)->Dict[str,Any]:data={}forfieldinmodel.get_fields_list(request,action):data[field.name]=awaitfield.parse_form_data(request,form_data,action)returndatadefmount_to(self,app:Starlette,redirect_slashes:bool=True,)->None:admin_app=Starlette(routes=self.routes,middleware=self.middlewares,debug=self.debug,exception_handlers={HTTPException:self._render_error},)admin_app.state.ROUTE_NAME=self.route_nameapp.mount(self.base_url,app=admin_app,name=self.route_name,)admin_app.router.redirect_slashes=redirect_slashes
def__init__(self,title:str=_("Admin"),base_url:str="/admin",route_name:str="admin",logo_url:Optional[str]=None,login_logo_url:Optional[str]=None,templates_dir:str="templates",statics_dir:Optional[str]=None,index_view:Optional[CustomView]=None,auth_provider:Optional[BaseAuthProvider]=None,middlewares:Optional[Sequence[Middleware]]=None,debug:bool=False,i18n_config:Optional[I18nConfig]=None,favicon_url:Optional[str]=None,):""" Parameters: title: Admin title. base_url: Base URL for Admin interface. route_name: Mounted Admin name logo_url: URL of logo to be displayed instead of title. login_logo_url: If set, it will be used for login interface instead of logo_url. templates_dir: Templates dir for customisation statics_dir: Statics dir for customisation index_view: CustomView to use for index page. auth_provider: Authentication Provider middlewares: Starlette middlewares i18n_config: i18n configuration favicon_url: URL of favicon. """self.title=titleself.base_url=base_urlself.route_name=route_nameself.logo_url=logo_urlself.login_logo_url=login_logo_urlself.favicon_url=favicon_urlself.templates_dir=templates_dirself.statics_dir=statics_dirself.auth_provider=auth_providerself.middlewares=list(middlewares)ifmiddlewaresisnotNoneelse[]self.index_view=(index_viewif(index_viewisnotNone)elseCustomView("",add_to_menu=False))self._views:List[BaseView]=[]self._models:List[BaseModelView]=[]self.routes:List[Union[Route,Mount]]=[]self.debug=debugself.i18n_config=i18n_configself._setup_templates()self.init_locale()self.init_auth()self.init_routes()
defadd_view(self,view:Union[Type[BaseView],BaseView])->None:""" Add View to the Admin interface. """view_instance=viewifisinstance(view,BaseView)elseview()self._views.append(view_instance)self.setup_view(view_instance)
custom_render_js(request)
Override this function to provide a link to custom js to override the
global render object in javascript which is use to render fields in
list page.
defcustom_render_js(self,request:Request)->Optional[str]:""" Override this function to provide a link to custom js to override the global `render` object in javascript which is use to render fields in list page. Args: request: Starlette Request """returnNone