Application

class Admin(BaseAdminView):
    """Main entrypoint to admin interface.

    ???+ usage
        ```python
        from fastapi import FastAPI
        from sqladmin import Admin, ModelView

        from mymodels import User # SQLAlchemy model


        app = FastAPI()
        admin = Admin(app, engine)


        class UserAdmin(ModelView, model=User):
            column_list = [User.id, User.name]


        admin.add_view(UserAdmin)
        ```
    """

    def __init__(  # type: ignore[no-any-unimported]
        self,
        app: Starlette,
        engine: ENGINE_TYPE | None = None,
        session_maker: SESSION_MAKER | None = None,
        base_url: str = "/admin",
        title: str = "Admin",
        logo_url: str | None = None,
        logo_width: int = 64,
        logo_height: int = 64,
        favicon_url: str | None = None,
        middlewares: Sequence[Middleware] | None = None,
        debug: bool = False,
        templates_dir: str = "templates",
        authentication_backend: AuthenticationBackend | None = None,
        static_files_kwargs: dict[str, Any] | None = None,
        i18n_config: I18nConfig | None = None,
        audit_backend: AuditBackend | None = None,
    ) -> None:
        """
        Args:
            app: Starlette or FastAPI application.
            engine: SQLAlchemy engine instance.
            session_maker: SQLAlchemy sessionmaker instance.
            base_url: Base URL for Admin interface.
            title: Admin title.
            logo_url: URL of logo to be displayed instead of title.
            logo_width: Width of the logo image in pixels. Defaults to 64.
            logo_height: Height of the logo image in pixels. Defaults to 64.
            favicon_url: URL of favicon to be displayed.
            static_files_kwargs: Extra keyword arguments for Starlette StaticFiles.
            i18n_config: Internationalization configuration. When provided, the
                interface is translated per request and, if
                ``language_switcher`` is set, a language switcher is shown.
        """

        super().__init__(
            app=app,
            engine=engine,
            session_maker=session_maker,  # type: ignore[arg-type]
            base_url=base_url,
            title=title,
            logo_url=logo_url,
            logo_width=logo_width,
            logo_height=logo_height,
            favicon_url=favicon_url,
            templates_dir=templates_dir,
            middlewares=middlewares,
            authentication_backend=authentication_backend,
            i18n_config=i18n_config,
            audit_backend=audit_backend,
        )

        static_files_kwargs = {**(static_files_kwargs or {}), "packages": ["sqladmin"]}
        statics = StaticFiles(**static_files_kwargs)

        async def http_exception(
            request: Request, exc: Exception
        ) -> Response | Awaitable[Response]:
            if not isinstance(exc, HTTPException):
                raise TypeError(f"Expected HTTPException, got {type(exc)}")

            context = {
                "status_code": exc.status_code,
                "message": exc.detail,
            }
            return await self.templates.TemplateResponse(
                request, "sqladmin/error.html", context, status_code=exc.status_code
            )

        routes = [
            Mount("/statics", app=statics, name="statics"),
            Route("/", endpoint=self.index, name="index"),
            Route("/{identity}/list", endpoint=self.list, name="list"),
            Route(
                "/{identity}/details/{pk:path}", endpoint=self.details, name="details"
            ),
            Route(
                "/{identity}/delete",
                endpoint=self.delete,
                name="delete",
                methods=["DELETE"],
            ),
            Route(
                "/{identity}/create",
                endpoint=self.create,
                name="create",
                methods=["GET", "POST"],
            ),
            Route(
                "/{identity}/edit/{pk:path}",
                endpoint=self.edit,
                name="edit",
                methods=["GET", "POST"],
            ),
            Route(
                "/{identity}/export/{export_type}", endpoint=self.export, name="export"
            ),
            Route(
                "/{identity}/import",
                endpoint=self.import_endpoint,
                name="import",
                methods=["POST"],
            ),
            Route(
                "/{identity}/ajax/lookup", endpoint=self.ajax_lookup, name="ajax_lookup"
            ),
            Route("/login", endpoint=self.login, name="login", methods=["GET", "POST"]),
            Route("/logout", endpoint=self.logout, name="logout", methods=["GET"]),
            Route(
                "/{identity}/{pk:path}/{column_name}/download/",
                endpoint=self.file_download,
                name="file_download",
                methods=["GET"],
            ),
            Route(
                "/{identity}/{pk:path}/{column_name}/preview/",
                endpoint=self.file_preview,
                name="file_preview",
                methods=["GET"],
            ),
        ]

        self.admin.router.routes = routes
        self.admin.exception_handlers = {HTTPException: http_exception}
        self.admin.debug = debug
        self.app.mount(base_url, app=self.admin, name="admin")

    @login_required
    async def index(self, request: Request) -> Response:
        """Index route which can be overridden to create dashboards."""

        return await self.templates.TemplateResponse(request, "sqladmin/index.html")

    @login_required
    async def list(self, request: Request) -> Response:
        """List route to display paginated Model instances."""

        await self._list(request)

        model_view = self._find_model_view(request.path_params["identity"])
        pagination = await model_view.list(request)
        pagination.add_pagination_urls(request.url)

        request_page = model_view.validate_page_number(
            request.query_params.get("page"), 1
        )

        if request_page > pagination.page:
            return RedirectResponse(
                request.url.include_query_params(page=pagination.page),
                status_code=status.HTTP_302_FOUND,
            )

        context = {
            **await model_view.list_context(request),
            "model_view": model_view,
            "pagination": pagination,
            "can_import": await model_view.check_can_import(request),
        }

        if request.query_params.get("error"):
            context["error"] = request.query_params["error"]

        return await self.templates.TemplateResponse(
            request, model_view.list_template, context
        )

    @login_required
    async def details(self, request: Request) -> Response:
        """Details route."""

        await self._details(request)
        model_view = self._find_model_view(request.path_params["identity"])
        model = await model_view.get_object_for_details(request)

        if not model:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

        context = {
            **await model_view.details_context(request),
            "model_view": model_view,
            "model": model,
            "title": model_view.name,
        }

        return await self.templates.TemplateResponse(
            request, model_view.details_template, context
        )

    @login_required
    async def delete(self, request: Request) -> Response:
        """Delete route."""

        await self._delete(request)

        identity = request.path_params["identity"]
        model_view = self._find_model_view(identity)

        params = request.query_params.get("pks", "")
        pks = params.split(",") if params else []

        referer_url = URL(request.headers.get("referer", ""))
        referer_params = MultiDict(parse_qsl(referer_url.query))

        try:
            for pk in pks:
                model = await model_view.get_object_for_delete(pk)
                if not model:
                    raise HTTPException(
                        status_code=status.HTTP_404_NOT_FOUND, detail="Object not found"
                    )

                await model_view.delete_model(request, pk)
        except Exception as e:
            logger.exception(e)
            referer_params["error"] = str(e)

        url = URL(str(request.url_for("admin:list", identity=identity)))
        url = url.include_query_params(**referer_params)
        return PlainTextResponse(content=str(url))

    async def _resolve_after_change_response(
        self,
        request: Request,
        context: dict,
        obj: Any,
        template: str,
        identity: str,
    ) -> Response | None:
        """Return an override response from ``after_model_change`` or the
        one-time secret modal, if either is set on ``request.state``."""

        after_response = getattr(request.state, "_sqladmin_after_change_response", None)
        if isinstance(after_response, Response):
            return after_response

        if Secret.get(request) is not None:
            context["obj"] = obj
            context["secret_next_url"] = str(
                request.url_for("admin:list", identity=identity)
            )
            response = await self.templates.TemplateResponse(request, template, context)
            Secret.apply_no_store_headers(response)
            return response

        return None

    @login_required
    async def create(self, request: Request) -> Response:
        """Create model endpoint."""
        await self._create(request)

        identity = request.path_params["identity"]
        model_view = self._find_model_view(identity)

        Form = await model_view.scaffold_form(model_view._form_create_rules)

        create_context = await model_view.create_context(request)

        if request.method == "GET":
            form = Form()
            context = {
                **create_context,
                "model_view": model_view,
                "form": form,
            }
            return await self.templates.TemplateResponse(
                request, model_view.create_template, context
            )

        form_data = await self._handle_form_data(request)
        form = Form(form_data)

        context = {
            **create_context,
            "model_view": model_view,
            "form": form,
        }

        if not form.validate():
            return await self.templates.TemplateResponse(
                request,
                model_view.create_template,
                context,
                status_code=status.HTTP_400_BAD_REQUEST,
            )

        form_data_dict = self._denormalize_wtform_data(form.data, model_view.model)
        try:
            obj = await model_view.insert_model(request, form_data_dict)
        except Exception as e:
            logger.exception(e)
            context["error"] = str(e)
            return await self.templates.TemplateResponse(
                request,
                model_view.create_template,
                context,
                status_code=status.HTTP_400_BAD_REQUEST,
            )

        override = await self._resolve_after_change_response(
            request, context, obj, model_view.create_template, identity
        )
        if override is not None:
            return override

        url = self.get_save_redirect_url(
            request=request,
            form=form_data,
            obj=obj,
            model_view=model_view,
        )
        return RedirectResponse(url=url, status_code=status.HTTP_302_FOUND)

    @login_required
    async def edit(self, request: Request) -> Response:
        """Edit model endpoint."""

        await self._edit(request)

        identity = request.path_params["identity"]
        model_view = self._find_model_view(identity)

        Form = await model_view.scaffold_form(model_view._form_edit_rules)
        model = await model_view.get_object_for_edit(request)
        if not model:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
        initial_data = await model_view.get_form_data_for_edit(model)
        context = {
            **await model_view.edit_context(request),
            "obj": model,
            "model_view": model_view,
            "form": Form(data=initial_data),
        }

        if request.method == "GET":
            return await self.templates.TemplateResponse(
                request, model_view.edit_template, context
            )

        form_data = await self._handle_form_data(request, model)
        form = Form(form_data)
        context["form"] = form

        if not form.validate():
            return await self.templates.TemplateResponse(
                request,
                model_view.edit_template,
                context,
                status_code=status.HTTP_400_BAD_REQUEST,
            )

        form_data_dict = self._denormalize_wtform_data(form.data, model)
        try:
            if model_view.save_as and form_data.get("save") == "Save as new":
                obj = await model_view.insert_model(request, form_data_dict)
            else:
                obj = await model_view.update_model(
                    request, pk=request.path_params["pk"], data=form_data_dict
                )
        except Exception as e:
            logger.exception(e)
            context["error"] = str(e)
            return await self.templates.TemplateResponse(
                request,
                model_view.edit_template,
                context,
                status_code=status.HTTP_400_BAD_REQUEST,
            )

        override = await self._resolve_after_change_response(
            request, context, obj, model_view.edit_template, identity
        )
        if override is not None:
            return override

        url = self.get_save_redirect_url(
            request=request,
            form=form_data,
            obj=obj,
            model_view=model_view,
        )
        return RedirectResponse(url=url, status_code=status.HTTP_302_FOUND)

    @login_required
    async def export(self, request: Request) -> Response:
        """Export model endpoint."""

        await self._export(request)

        identity = request.path_params["identity"]
        export_type = request.path_params["export_type"]

        model_view = self._find_model_view(identity)
        rows = await model_view.get_model_objects(
            request=request, limit=model_view.export_max_rows
        )
        return await model_view.export_data(
            rows, export_type=export_type, request=request
        )

    @login_required
    async def import_endpoint(self, request: Request) -> Response:
        """Import model endpoint."""

        await self._import(request)

        identity = request.path_params["identity"]
        model_view = self._find_model_view(identity)

        upload = await handle_import_upload(request, model_view)
        if upload.error:
            return import_error_response(upload.error, upload.status_code)
        if not upload.content:
            return import_error_response(
                "No CSV file uploaded or file does not have a .csv extension."
            )

        return await import_csv(
            request,
            model_view,
            upload.content,
            upload.continue_on_error,
            self._denormalize_wtform_data,
        )

    async def login(self, request: Request) -> Response:
        if self.authentication_backend is None:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail="Authentication backend not configured.",
            )

        context = {}
        if request.method == "GET":
            return await self.templates.TemplateResponse(request, "sqladmin/login.html")

        response = await self.authentication_backend.login(request)
        if not response:
            context["error"] = "Invalid credentials."
            return await self.templates.TemplateResponse(
                request,
                "sqladmin/login.html",
                context,
                status_code=status.HTTP_400_BAD_REQUEST,
            )

        if isinstance(response, Response):
            return response

        return RedirectResponse(
            request.url_for("admin:index"), status_code=status.HTTP_302_FOUND
        )

    async def logout(self, request: Request) -> Response:
        if self.authentication_backend is None:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail="Authentication backend not configured.",
            )

        response = await self.authentication_backend.logout(request)

        if not response:
            return RedirectResponse(
                request.url_for("admin:login"), status_code=status.HTTP_302_FOUND
            )

        if isinstance(response, Response):
            return response

        return RedirectResponse(
            request.url_for("admin:index"), status_code=status.HTTP_302_FOUND
        )

    @login_required
    async def file_download(self, request: Request) -> Response:
        """Download file endpoint."""
        request_path = await self._get_file(request)
        return FileResponse(request_path, filename=request_path.name)

    @login_required
    async def file_preview(self, request: Request) -> Response:
        """Preview file inline."""
        request_path = await self._get_file(request)
        return FileResponse(
            request_path, filename=request_path.name, content_disposition_type="inline"
        )

    @login_required
    async def ajax_lookup(self, request: Request) -> Response:
        """Ajax lookup route."""

        identity = request.path_params["identity"]
        model_view = self._find_model_view(identity)

        if not model_view.is_accessible(request):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)

        name = request.query_params.get("name")
        term = request.query_params.get("term")

        if not name or not term:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)

        try:
            loader: QueryAjaxModelLoader = model_view._form_ajax_refs[name]
        except KeyError as exc:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST) from exc

        data = [loader.format(m) for m in await loader.get_list(request, term)]
        return JSONResponse({"results": data})

    @staticmethod
    def get_save_redirect_url(
        request: Request, form: FormData, model_view: ModelView, obj: Any
    ) -> str | URL:
        """
        Get the redirect URL after a save action
        which is triggered from create/edit page.
        """

        identity = request.path_params["identity"]
        identifier = get_object_identifier(obj)

        if form.get("save") == "Save":
            url = URL(str(request.url_for("admin:list", identity=identity)))
            if request.url.query:
                url = url.replace(query=request.url.query)
            return url

        if form.get("save") == "Save and continue editing" or (
            form.get("save") == "Save as new" and model_view.save_as_continue
        ):
            return request.url_for("admin:edit", identity=identity, pk=identifier)

        return request.url_for("admin:create", identity=identity)

    @staticmethod
    async def _handle_form_data(request: Request, obj: Any = None) -> FormData:
        """
        Handle form data and modify in case of UploadFile.
        This is needed since in edit page
        there's no way to show current file of object.
        """

        form = await request.form()
        form_data: list[tuple[str, str | UploadFile]] = []
        for key, value in form.multi_items():
            if not isinstance(value, UploadFile):
                form_data.append((key, value))
                continue

            should_clear = form.get(key + "_checkbox")
            empty_upload = len(await value.read(1)) != 1
            await value.seek(0)
            if should_clear:
                form_data.append((key, UploadFile(io.BytesIO(b""))))
            elif empty_upload and obj and getattr(obj, key):
                f = getattr(obj, key)  # In case of update, imitate UploadFile
                form_data.append((key, UploadFile(filename=f.name, file=f.open())))
            else:
                form_data.append((key, value))
        return FormData(form_data)

    @staticmethod
    def _normalize_wtform_data(obj: Any) -> dict:
        form_data = {}
        for field_name in WTFORMS_ATTRS:
            if value := getattr(obj, field_name, None):
                form_data[field_name + "_"] = value
        return form_data

    @staticmethod
    def _denormalize_wtform_data(form_data: dict, obj: Any) -> dict:
        data = form_data.copy()
        for field_name in WTFORMS_ATTRS_REVERSED:
            reserved_field_name = field_name[:-1]
            if (
                field_name in data
                and not hasattr(obj, field_name)
                and hasattr(obj, reserved_field_name)
            ):
                data[reserved_field_name] = data.pop(field_name)
        return data

class BaseAdmin:
    """Base class for implementing Admin interface.

    Danger:
        This class should almost never be used directly.
    """

    def __init__(
        self,
        app: Starlette,
        engine: ENGINE_TYPE | None = None,
        session_maker: SESSION_MAKER | None = None,
        base_url: str = "/admin",
        title: str = "Admin",
        logo_url: str | None = None,
        logo_width: int = 64,
        logo_height: int = 64,
        favicon_url: str | None = None,
        templates_dir: str = "templates",
        middlewares: Sequence[Middleware] | None = None,
        authentication_backend: AuthenticationBackend | None = None,
        i18n_config: I18nConfig | None = None,
        audit_backend: AuditBackend | None = None,
    ) -> None:
        self.app = app
        self.audit_backend = audit_backend or NullAuditBackend()
        self.engine = engine
        self.base_url = base_url
        self.templates_dir = templates_dir
        self.title = title
        self.logo_url = logo_url
        self.logo_width = logo_width
        self.logo_height = logo_height
        self.favicon_url = favicon_url
        self.i18n_config = i18n_config
        if i18n_config is not None and not BABEL_INSTALLED:
            warnings.warn(
                "i18n_config was provided but the 'babel' package is not "
                "installed; the interface will not be translated. Install it "
                "with `pip install sqladmin[i18n]`.",
                UserWarning,
                stacklevel=3,
            )

        if session_maker:
            self.session_maker = session_maker
        elif isinstance(self.engine, Engine):
            self.session_maker = sessionmaker(bind=self.engine, class_=Session)
        else:
            self.session_maker = async_sessionmaker(
                bind=self.engine,
                class_=AsyncSession,
            )

        self.session_maker.configure(autoflush=False, autocommit=False)
        self.is_async = is_async_session_maker(self.session_maker)

        middlewares = list(middlewares or [])
        self.authentication_backend = authentication_backend
        if authentication_backend:
            middlewares.extend(authentication_backend.middlewares)

        if self.i18n_config is not None:
            middlewares.append(
                Middleware(
                    LocaleMiddleware,
                    i18n_config=self.i18n_config,
                    cookie_path=self.base_url or "/",
                )
            )

        self.admin = Starlette(middleware=middlewares)
        self.templates = self.init_templating_engine()
        self._views: list[BaseView | ModelView] = []
        self._menu = Menu()

    def init_templating_engine(self) -> Jinja2Templates:
        templates = Jinja2Templates("templates")
        loaders = [
            FileSystemLoader(self.templates_dir),
            PrefixLoader(
                {"sqladmin_original": PackageLoader("sqladmin", "templates/sqladmin")}
            ),
            PackageLoader("sqladmin", "templates"),
        ]

        templates.env.loader = ChoiceLoader(loaders)
        templates.env.globals["min"] = min
        templates.env.globals["zip"] = zip
        templates.env.globals["admin"] = self
        templates.env.globals["is_list"] = lambda x: isinstance(x, (list, set))
        templates.env.globals["get_object_identifier"] = get_object_identifier
        templates.env.globals["get_flashed_messages"] = get_flashed_messages
        templates.env.globals["Secret"] = Secret
        templates.env.globals["collect_form_media"] = collect_form_media

        templates.env.add_extension("jinja2.ext.i18n")
        if self.i18n_config is not None:
            templates.env.globals["i18n_config"] = self.i18n_config
            templates.env.globals["get_locale"] = get_locale
            templates.env.globals["get_locale_display_name"] = get_locale_display_name
            templates.env.install_gettext_callables(  # type: ignore[attr-defined]
                gettext, ngettext, newstyle=True
            )
        else:
            # No i18n configured: keep templates that use ``_("...")`` working
            # by installing identity (null) translation callables.
            templates.env.globals["i18n_config"] = I18nConfig()
            templates.env.install_null_translations(newstyle=True)  # type: ignore[attr-defined]

        return templates

    @property
    def views(self) -> list[BaseView | ModelView]:
        """Get list of ModelView and BaseView instances lazily.

        Returns:
            List of ModelView and BaseView instances added to Admin.
        """

        return self._views

    def _find_model_view(self, identity: str) -> ModelView:
        for view in self.views:
            if isinstance(view, ModelView) and view.identity == identity:
                return view

        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

    def add_view(self, view: type[ModelView] | type[BaseView]) -> None:
        """Add ModelView or BaseView classes to Admin.
        This is a shortcut that will handle both `add_model_view` and `add_base_view`.
        """

        if view.is_model:
            self.add_model_view(view)  # type: ignore
        else:
            self.add_base_view(view)

    @staticmethod
    def _find_decorated_funcs(
        view: type[BaseView | ModelView],
        view_instance: BaseView | ModelView,
        handle_fn: Callable[
            [MethodType, type[BaseView | ModelView], BaseView | ModelView],
            None,
        ],
    ) -> None:
        funcs = inspect.getmembers(view_instance, predicate=inspect.ismethod)

        for _, func in sorted(
            funcs,
            key=lambda x: inspect.getsourcelines(x[1])[1],
            reverse=True,
        ):
            handle_fn(func, view, view_instance)

    def _handle_action_decorated_func(
        self,
        func: MethodType,
        view: type[BaseView | ModelView],
        view_instance: BaseView | ModelView,
    ) -> None:
        if hasattr(func, "_action"):
            view_instance = cast(ModelView, view_instance)
            self.admin.add_route(
                route=func,
                path=f"/{view_instance.identity}/action/" + getattr(func, "_slug"),
                methods=["GET"],
                name=f"action-{view_instance.identity}-{getattr(func, '_slug')}",
                include_in_schema=getattr(func, "_include_in_schema"),
            )

            if getattr(func, "_add_in_list"):
                view_instance._custom_actions_in_list[getattr(func, "_slug")] = getattr(
                    func, "_label"
                )
            if getattr(func, "_add_in_detail"):
                view_instance._custom_actions_in_detail[getattr(func, "_slug")] = (
                    getattr(func, "_label")
                )

            if getattr(func, "_confirmation_message"):
                view_instance._custom_actions_confirmation[getattr(func, "_slug")] = (
                    getattr(func, "_confirmation_message")
                )

    def _handle_expose_decorated_func(
        self,
        func: MethodType,
        view: type[BaseView | ModelView],
        view_instance: BaseView | ModelView,
    ) -> None:
        if hasattr(func, "_exposed"):
            if view.is_model:
                path = f"/{view_instance.identity}" + getattr(func, "_path")
                name = f"view-{view_instance.identity}-{func.__name__}"
            else:
                view.identity = getattr(func, "_identity")
                path = getattr(func, "_path")
                name = f"view-{view.identity}"

            self.admin.add_route(
                route=func,
                path=path,
                methods=getattr(func, "_methods"),
                name=name,
                include_in_schema=getattr(func, "_include_in_schema"),
            )

    def add_model_view(self, view: type[ModelView]) -> None:
        """Add ModelView to the Admin.

        ???+ usage
            ```python
            from sqladmin import Admin, ModelView

            class UserAdmin(ModelView, model=User):
                pass

            admin.add_model_view(UserAdmin)
            ```
        """

        view._admin_ref = self
        # Set database engine from Admin instance
        view.session_maker = self.session_maker
        view.is_async = self.is_async
        view.ajax_lookup_url = urljoin(
            self.base_url + "/", f"{view.identity}/ajax/lookup"
        )
        view.templates = self.templates
        view_instance = view()

        self._find_decorated_funcs(
            view, view_instance, self._handle_action_decorated_func
        )

        self._find_decorated_funcs(
            view, view_instance, self._handle_expose_decorated_func
        )

        self._views.append(view_instance)
        self._build_menu(view_instance)

    def add_base_view(self, view: type[BaseView]) -> None:
        """Add BaseView to the Admin.

        ???+ usage
            ```python
            from sqladmin import BaseView, expose

            class CustomAdmin(BaseView):
                name = "Custom Page"
                icon = "fa-solid fa-chart-line"

                @expose("/custom", methods=["GET"])
                async def test_page(self, request: Request):
                    return await self.templates.TemplateResponse(request, "custom.html")

            admin.add_base_view(CustomAdmin)
            ```
        """

        view._admin_ref = self
        view.templates = self.templates
        view_instance = view()

        self._find_decorated_funcs(
            view, view_instance, self._handle_expose_decorated_func
        )
        self._views.append(view_instance)
        self._build_menu(view_instance)

    def _build_menu(self, view: ModelView | BaseView) -> None:
        if view.category:
            menu = CategoryMenu(name=view.category, icon=view.category_icon)
            menu.add_child(ViewMenu(view=view, name=view.name, icon=view.icon))
            self._menu.add(menu)
        else:
            self._menu.add(ViewMenu(view=view, icon=view.icon, name=view.name))

class FlashLevel(Enum):
    """Defines the standard severity levels for flash messages.

    These values map directly to Bootstrap contextual color classes
    used in the toast notification UI.

    Attributes:
        info: General informational message. Maps to Bootstrap `primary` (blue).
        error: High-severity error message. Maps to Bootstrap `danger` (red).
        warning: Cautionary message. Maps to Bootstrap `warning` (yellow).
        success: Successful operation message. Maps to Bootstrap `success` (green).
    """

    info = "primary"
    error = "danger"
    warning = "warning"
    success = "success"

class Flash:
    """Utility class providing class methods for session-based flash messages.

    Flash messages are stored in the session and displayed as Bootstrap toast
    notifications on the next page render. They are automatically cleared after
    being displayed.

    All methods are class methods and require the `request` object to access
    the user session. If no session is available, messages are silently ignored.
    """

    @classmethod
    def flash(
        cls,
        request: Request,
        message: str,
        level: FlashLevel = FlashLevel.info,
        title: str = "",
    ) -> bool:
        """Add a flash message with an explicitly defined severity level.

        This is the primary method. Use the convenience methods (`info`,
        `success`, `warning`, `error`) to avoid specifying the level manually.

        Args:
            request: The current incoming request object.
            message: The main text content of the message.
            level: The severity level. Defaults to `FlashLevel.info`.
            title: An optional title shown in the toast header. Defaults to `""`.

        Returns:
            `True` if the message was stored successfully,
            `False` if no session is available.

        Example:
            ```python
            from sqladmin import Flash, FlashLevel

            Flash.flash(
                request,
                "A crucial server process has started.",
                FlashLevel.warning,
                "System Alert",
            )
            ```
        """
        return flash(
            request,
            message,
            level.value,
            title,
        )

    @classmethod
    def info(cls, request: Request, message: str, title: str = "") -> bool:
        """Add an informational flash message.

        Args:
            request: The current incoming request object.
            message: The message content.
            title: An optional title shown in the toast header. Defaults to `""`.

        Returns:
            `True` if the message was stored successfully,
            `False` if no session is available.

        Example:
            ```python
            from sqladmin import Flash

            Flash.info(request, "Your export is being processed in the background.")
            ```
        """
        return cls.flash(request, message, FlashLevel.info, title)

    @classmethod
    def success(cls, request: Request, message: str, title: str = "") -> bool:
        """Add a success flash message.

        Args:
            request: The current incoming request object.
            message: The message content.
            title: An optional title shown in the toast header. Defaults to `""`.

        Returns:
            `True` if the message was stored successfully,
            `False` if no session is available.

        Example:
            ```python
            from sqladmin import Flash

            Flash.success(request, "Users approved successfully.", "Done")
            ```
        """
        return cls.flash(request, message, FlashLevel.success, title)

    @classmethod
    def warning(cls, request: Request, message: str, title: str = "") -> bool:
        """Add a warning flash message.

        Args:
            request: The current incoming request object.
            message: The message content.
            title: An optional title shown in the toast header. Defaults to `""`.

        Returns:
            `True` if the message was stored successfully,
            `False` if no session is available.

        Example:
            ```python
            from sqladmin import Flash

            Flash.warning(request, "Some records could not be processed.")
            ```
        """
        return cls.flash(request, message, FlashLevel.warning, title)

    @classmethod
    def error(cls, request: Request, message: str, title: str = "") -> bool:
        """Add an error flash message.

        Args:
            request: The current incoming request object.
            message: The message content.
            title: An optional title shown in the toast header. Defaults to `""`.

        Returns:
            `True` if the message was stored successfully,
            `False` if no session is available.

        Example:
            ```python
            from sqladmin import Flash

            Flash.error(request, "Access denied. Invalid credentials.", "Error")
            ```
        """
        return cls.flash(request, message, FlashLevel.error, title)