Admin

fastapi_admin_kit.Admin

Main admin interface. Register models and mount to your FastAPI app.

Uses component-based architecture with: - config: AdminConfig (UI, auth, audit, behavior, storage, nav settings) - database: AdminDatabase (engine, table creation, role seeding) - router: AdminRouter (routing, static files, Jinja) - template: AdminTemplate (branding, sidebar context)

Source code in fastapi_admin_kit/admin/core.py

class Admin:
    """Main admin interface. Register models and mount to your FastAPI app.

    Uses component-based architecture with:
    - config: AdminConfig (UI, auth, audit, behavior, storage, nav settings)
    - database: AdminDatabase (engine, table creation, role seeding)
    - router: AdminRouter (routing, static files, Jinja)
    - template: AdminTemplate (branding, sidebar context)
    """

    def __init__(
        self,
        app: FastAPI | None = None,
        engine: Any | None = None,
        database_config: DatabaseConfig | None = None,
        *,
        # Component instances (new API)
        config: AdminConfig | None = None,
        database: AdminDatabase | None = None,
        router: AdminRouter | None = None,
        template: AdminTemplate | None = None,
        backend: Any | None = None,
        # Legacy kwargs for backward compatibility
        base: type | None = None,
        title: str = "FastAPI Admin Kit",
        logo_url: str | None = None,
        favicon_url: str | None = None,
        primary_color: str = "#0ea5e9",
        primary_color_dark: str = "#0284c7",
        dark_mode_default: bool = False,
        per_page_default: int = 25,
        session_ttl: int = 28800,
        audit_retention_days: int = 365,
        dashboard_stats: list[str] | None = None,
        dashboard_charts: bool = True,
        admin_path: str = "/admin",
        secret_key: str = "",
        auth_model: type | None = None,
        auth_backend: AuthBackend | None = BuiltinAuthBackend(),
        session_cookie_name: str = "admin_session",
        session_secure: bool = True,
        session_samesite: str = "strict",
        access_token_ttl: int = 600,
        api_token_middleware: bool = True,
        api_token_strict: bool = False,
        trusted_proxies: list[str] | None = None,
        seed_roles: list[SeedRole] | None = None,
        seed_roles_overwrite: bool = False,
        superuser_emails: list[str] | None = None,
        storage: StorageBackend | None = None,
        uploads_url: str = "/uploads",
        auto_discover: bool = True,
        skip_models: list[str] | None = None,
        nav_groups: list[NavGroupConfig] | None = None,
        sidebar_builder: SidebarBuilder | None = None,
        require_tags: bool = False,
        theme: ThemeConfig | None = None,
        # UI component config
        sidebar_style: str = "default",
        sidebar_position: str = "left",
        table_style: str = "default",
        table_row_height: str = "normal",
        form_layout: str = "two-column",
        form_spacing: str = "normal",
        dashboard_grid: str = "auto",
        dashboard_card_style: str = "default",
        dashboard_stat_size: str = "normal",
        content_width: str = "default",
        topbar_style: str = "default",
        custom_css: str = "",
        custom_css_url: str = "",
        custom_js: str = "",
        custom_js_url: str = "",
        show_history: bool = True,
        show_view_on_site: bool = True,
        environment_label: str | None = None,
        environment_color: str = "info",
        mobile_sidebar: str = "overlay",
        dashboard_permission: str | None = None,
        settings_permission: str | None = None,
        # AI
        ai: Any = None,
        ai_enabled: bool = False,
        is_development: bool = False,
        sidebar_bottom_links: list[dict[str, str]] | None = None,
        # Notifications
        enable_notification: bool = True,
        notification_service: Any | None = None,
        notifications_api_path: str | None = None,
        notifications_list_path: str | None = None,
        # AI chat file attachments
        ai_chat_max_file_size_mb: int = 10,
        ai_chat_allowed_extensions: list[str] | None = None,
        # Optional Redis-backed caching (opt-in)
        cache_enabled: bool | None = None,
        cache_ttl: int | None = None,
    ):
        self.registry = AdminRegistry()
        self._app: FastAPI | None = app

        # Expose the instance on app.state immediately so plugins (e.g.
        # configure_notifications) can reach the admin before admin.setup()
        # runs — _wire_app_state() overwrites the same slot at setup time.
        if app is not None:
            app.state.admin = self

        # Add CSRF middleware early (must be before app starts)
        if app is not None:
            from fastapi_admin_kit.auth.csrf import (
                CSRFMiddleware,
                auth_redirect_handler,
                forbidden_handler,
            )

            app.add_exception_handler(401, auth_redirect_handler)
            app.add_exception_handler(403, forbidden_handler)
            app.add_middleware(CSRFMiddleware)
            self._csrf_middleware_added = True

            # API bearer-token pre-validation. Added BEFORE SessionMiddleware
            # so it runs inside it (Starlette: last added = outermost) and can
            # use the per-request DB session to resolve the live user.
            from fastapi_admin_kit.api.middleware import AccessTokenMiddleware

            app.add_middleware(AccessTokenMiddleware)
            self._api_token_middleware_added = True

            # Register the per-request session + audit-context middlewares here
            # (at construction time) rather than in ``setup()``. Starlette builds
            # ``app.middleware_stack`` on the *first* scope it receives — which is
            # the lifespan startup event that fires *before* ``setup()`` runs. If
            # these middlewares were only added in ``setup()``, the stack would
            # already be frozen and ``add_middleware`` would raise ``RuntimeError``
            # (silently swallowed), leaving the session middleware out of the live
            # stack. Without it, DB writes are flushed but never committed.
            from fastapi_admin_kit.audit.middleware import (
                AuditContextMiddleware,
            )
            from fastapi_admin_kit.db import SessionMiddleware

            app.add_middleware(SessionMiddleware)
            self._session_middleware_added = True
            app.add_middleware(AuditContextMiddleware)
            self._audit_middleware_added = True
        else:
            self._csrf_middleware_added = False
            self._api_token_middleware_added = False
            self._session_middleware_added = False
            self._audit_middleware_added = False

        # Default auth backend if none provided
        if auth_backend is None:
            from fastapi_admin_kit.auth.backend import BuiltinAuthBackend

            auth_backend = BuiltinAuthBackend()

        # Build components from legacy kwargs if components not provided
        if config is None:
            config = AdminConfig(
                ui=UIConfig(
                    title=title,
                    logo_url=logo_url,
                    favicon_url=favicon_url,
                    primary_color=primary_color,
                    primary_color_dark=primary_color_dark,
                    dark_mode_default=dark_mode_default,
                    per_page_default=per_page_default,
                    theme=theme,
                    sidebar_style=sidebar_style,
                    sidebar_position=sidebar_position,
                    table_style=table_style,
                    table_row_height=table_row_height,
                    form_layout=form_layout,
                    form_spacing=form_spacing,
                    dashboard_grid=dashboard_grid,
                    dashboard_card_style=dashboard_card_style,
                    dashboard_stat_size=dashboard_stat_size,
                    content_width=content_width,
                    topbar_style=topbar_style,
                    custom_css=custom_css,
                    custom_css_url=custom_css_url,
                    custom_js=custom_js,
                    custom_js_url=custom_js_url,
                    show_history=show_history,
                    show_view_on_site=show_view_on_site,
                    environment_label=environment_label,
                    environment_color=environment_color,
                    mobile_sidebar=mobile_sidebar,
                ),
                auth=AuthConfig(
                    auth_model=auth_model,
                    auth_backend=auth_backend,
                    session_ttl=session_ttl,
                    session_cookie_name=session_cookie_name,
                    session_secure=session_secure,
                    superuser_emails=superuser_emails,
                    session_samesite=session_samesite,
                    access_token_ttl=access_token_ttl,
                    api_token_middleware=api_token_middleware,
                    api_token_strict=api_token_strict,
                    trusted_proxies=trusted_proxies,
                ),
                audit=AuditConfig(audit_retention_days=audit_retention_days),
                behavior=BehaviorConfig(
                    auto_discover=auto_discover,
                    skip_models=skip_models,
                    dashboard_stats=dashboard_stats or [],
                    dashboard_charts=dashboard_charts,
                ),
                storage=StorageConfig(storage=storage, uploads_url=uploads_url),
                nav=NavConfig(
                    nav_groups=nav_groups or [],
                    sidebar_builder=sidebar_builder,
                    require_tags=require_tags,
                    dashboard_permission=dashboard_permission,
                    settings_permission=settings_permission,
                    sidebar_bottom_links=sidebar_bottom_links,
                ),
                ai_chat=AIChatConfig(
                    max_file_size_mb=ai_chat_max_file_size_mb,
                    allowed_extensions=ai_chat_allowed_extensions
                    or [
                        ".pdf",
                        ".xlsx",
                        ".xls",
                        ".docx",
                        ".doc",
                        ".csv",
                        ".png",
                        ".jpg",
                        ".jpeg",
                        ".gif",
                        ".webp",
                    ],
                ),
                cache=CacheConfig(enabled=cache_enabled, ttl=cache_ttl),
            )
        else:
            config = _merge_legacy_kwargs_into_config(
                config,
                ui=dict(
                    title=title,
                    logo_url=logo_url,
                    favicon_url=favicon_url,
                    primary_color=primary_color,
                    primary_color_dark=primary_color_dark,
                    dark_mode_default=dark_mode_default,
                    per_page_default=per_page_default,
                    theme=theme,
                    sidebar_style=sidebar_style,
                    sidebar_position=sidebar_position,
                    table_style=table_style,
                    table_row_height=table_row_height,
                    form_layout=form_layout,
                    form_spacing=form_spacing,
                    dashboard_grid=dashboard_grid,
                    dashboard_card_style=dashboard_card_style,
                    dashboard_stat_size=dashboard_stat_size,
                    content_width=content_width,
                    topbar_style=topbar_style,
                    custom_css=custom_css,
                    custom_css_url=custom_css_url,
                    custom_js=custom_js,
                    custom_js_url=custom_js_url,
                    show_history=show_history,
                    show_view_on_site=show_view_on_site,
                    environment_label=environment_label,
                    environment_color=environment_color,
                    mobile_sidebar=mobile_sidebar,
                ),
                auth=dict(
                    auth_model=auth_model,
                    auth_backend=auth_backend,
                    session_ttl=session_ttl,
                    session_cookie_name=session_cookie_name,
                    session_secure=session_secure,
                    superuser_emails=superuser_emails,
                    session_samesite=session_samesite,
                    access_token_ttl=access_token_ttl,
                    api_token_middleware=api_token_middleware,
                    api_token_strict=api_token_strict,
                    trusted_proxies=trusted_proxies,
                ),
                audit=dict(audit_retention_days=audit_retention_days),
                behavior=dict(
                    auto_discover=auto_discover,
                    skip_models=skip_models,
                    dashboard_stats=dashboard_stats or [],
                    dashboard_charts=dashboard_charts,
                ),
                storage=dict(storage=storage, uploads_url=uploads_url),
                nav=dict(
                    nav_groups=nav_groups or [],
                    sidebar_builder=sidebar_builder,
                    require_tags=require_tags,
                    dashboard_permission=dashboard_permission,
                    settings_permission=settings_permission,
                    sidebar_bottom_links=sidebar_bottom_links,
                ),
                cache=dict(enabled=cache_enabled, ttl=cache_ttl),
            )
            # The merge helper compares against CacheConfig's *constructor*
            # defaults, but the default instance already resolved the env
            # flag at construction — so apply explicit enabled/ttl overrides.
            if cache_enabled is not None:
                config.cache.enabled = cache_enabled
            if cache_ttl is not None:
                config.cache.ttl = cache_ttl

        if database is None:
            database = AdminDatabase(
                engine=engine,
                base=base,
                database_config=database_config,
                use_alembic=config.behavior.use_alembic,
            )

        if router is None:
            router = AdminRouter(
                admin_path=admin_path,
                secret_key=secret_key or os.environ.get("SECRET_KEY", ""),
            )

        if template is None:
            template = AdminTemplate(
                title=config.ui.title,
                logo_url=config.ui.logo_url,
                favicon_url=config.ui.favicon_url,
                primary_color=config.ui.primary_color,
                primary_color_dark=config.ui.primary_color_dark,
                dark_mode_default=config.ui.dark_mode_default,
                dashboard_permission=config.nav.dashboard_permission,
                settings_permission=config.nav.settings_permission,
                sidebar_bottom_links=config.nav.sidebar_bottom_links,
                template_dirs=config.template_dirs,
            )

        self.config = config
        self.database = database
        self.router = router
        self.template = template
        self.cache_config = config.cache

        # Redis availability flags (populated by _setup_redis).
        self.redis_enabled = False
        self.redis_configured = False
        self._redis_wired = False

        # Store notification paths on config for template access
        default_notifications_path = f"{self.router.admin_path}/notifications"
        default_notifications_list = f"{self.router.admin_path}/admin_notifications/"
        self.config.notifications_api_path = notifications_api_path or default_notifications_path
        self.config.notifications_list_path = notifications_list_path or default_notifications_list

        # Notifications: auto-wire a service in setup() unless the user has
        # configured one already (e.g. via configure_notifications).
        self._enable_notification = enable_notification
        self._notification_service = notification_service

        # Backend: defaults to composed SqlAlchemyBackend
        if backend is None:
            from fastapi_admin_kit.backends.sqlalchemy import SqlAlchemyBackend

            backend = SqlAlchemyBackend.from_admin_database(database)
        self.backend = backend

        # Wire the AdminDatabase (engine/base/config) into the backend's
        # DatabaseBackend adapter.  When a backend is supplied standalone
        # (e.g. ``backend=SqlAlchemyBackend()``) its ``database`` adapter has no
        # engine reference, so ``create_connection()``/``create_session_factory()``
        # would fail.  ``from_admin_database`` already sets this; we only fill it
        # in when it is missing so a user-provided backend still works.
        backend_database = getattr(self.backend, "database", None)
        if (
            backend_database is not None
            and getattr(backend_database, "_admin_database", None) is None
        ):
            backend_database._admin_database = database

        # Inject backend into the auth backend so BuiltinAuthBackend can build
        # queries via QueryBackend instead of importing sqlalchemy directly.
        _auth_backend = getattr(getattr(self, "config", None), "auth", None)
        _auth_backend = getattr(_auth_backend, "auth_backend", None) if _auth_backend else None
        if _auth_backend is not None:
            for attr, value in (
                ("_backend", self.backend),
                ("backend", self.backend),
                ("_query_backend", getattr(self.backend, "query", None)),
                ("query_backend", getattr(self.backend, "query", None)),
            ):
                try:
                    setattr(_auth_backend, attr, value)
                except Exception:
                    pass

        # Inject backend's introspection adapter into the registry's ModelInspector
        self.registry.inspector._adapter = self.backend.introspection

        # RBAC
        self.seed_roles = seed_roles if seed_roles is not None else DEFAULT_SEED_ROLES
        self.seed_roles_overwrite = seed_roles_overwrite

        # Built sidebar (populated during setup)
        self._nav_groups_built: list[Any] = []

        # AI
        if ai_enabled and ai is None:
            from fastapi_admin_kit.ai.config import AIConfig

            ai = AIConfig()
        self._ai_config = ai
        self._ai_enabled = ai_enabled
        self.is_development = is_development

        # Internal state (populated during setup)
        self._session_backend: Any = None
        self._jinja_env: Environment | None = None
        self._router_built: bool = False

        # Wire Redis now (before startup) so the SDK can wrap the lifespan
        # before it begins. Degrades gracefully when Redis is unavailable.
        if app is not None:
            self._setup_redis(app)

        if app is not None and engine is not None:
            # Deferred setup — user will call await admin.setup() via lifespan
            pass

    # ------------------------------------------------------------------
    # Backward-compatible property accessors
    # ------------------------------------------------------------------

    @property
    def title(self) -> str:
        return self.config.ui.title

    @property
    def logo_url(self) -> str | None:
        return self.config.ui.logo_url

    @property
    def favicon_url(self) -> str | None:
        return self.config.ui.favicon_url

    @property
    def primary_color(self) -> str:
        return self.config.ui.primary_color

    @property
    def primary_color_dark(self) -> str:
        return self.config.ui.primary_color_dark

    @property
    def dark_mode_default(self) -> bool:
        return self.config.ui.dark_mode_default

    @property
    def per_page_default(self) -> int:
        return self.config.ui.per_page_default

    @property
    def admin_path(self) -> str:
        return self.router.admin_path

    @property
    def secret_key(self) -> str:
        return self.router.secret_key

    @property
    def engine(self) -> Any | None:
        return self.database.engine

    @property
    def base(self) -> type | None:
        return self.database.base

    @property
    def session_ttl(self) -> int:
        return self.config.auth.session_ttl

    @property
    def audit_retention_days(self) -> int:
        return self.config.audit.audit_retention_days

    @property
    def dashboard_stats(self) -> list[str]:
        return self.config.behavior.dashboard_stats

    @property
    def dashboard_charts(self) -> bool:
        return self.config.behavior.dashboard_charts

    @property
    def auth_model(self) -> type | None:
        return self.config.auth.auth_model

    @property
    def auth_backend(self) -> AuthBackend | None:
        return self.config.auth.auth_backend

    @property
    def session_cookie_name(self) -> str:
        return self.config.auth.session_cookie_name

    @property
    def session_secure(self) -> bool:
        return self.config.auth.session_secure

    @property
    def superuser_emails(self) -> list[str]:
        return self.config.auth.superuser_emails

    @property
    def storage(self) -> StorageBackend | None:
        return self.config.storage.storage

    @property
    def uploads_url(self) -> str:
        return self.config.storage.uploads_url

    @property
    def auto_discover(self) -> bool:
        return self.config.behavior.auto_discover

    @property
    def nav_groups(self) -> list[NavGroupConfig]:
        return self.config.nav.nav_groups

    @property
    def sidebar_builder(self) -> SidebarBuilder | None:
        return self.config.nav.sidebar_builder

    @property
    def require_tags(self) -> bool:
        return self.config.nav.require_tags

    # ------------------------------------------------------------------
    # Setup (async)
    # ------------------------------------------------------------------

    async def setup(self, app: FastAPI | None = None) -> None:
        """Run all startup wiring: create tables, seed roles, mount assets.

        This must be called once during application lifespan, typically via
        the :meth:`lifespan` context manager.
        """
        if app is not None:
            self._app = app

        if self._app is None:
            raise ConfigError(
                "Admin requires a FastAPI app instance. Pass app= or call setup(app=)."
            )

        self.database._ensure_engine()

        if self.database.engine is None:
            raise ConfigError(
                "Admin requires a SQLAlchemy engine. Pass engine= or database_config= to Admin()."
            )

        app = self._app

        # Wire Redis when the app was supplied after construction (e.g. via
        # ``await admin.setup(app)`` inside a manual lifespan).
        if not getattr(self, "_redis_wired", False):
            self._setup_redis(app)

        # Add CSRF middleware if not already added in __init__
        if not getattr(self, "_csrf_middleware_added", False):
            from fastapi_admin_kit.auth.csrf import (
                CSRFMiddleware,
                auth_redirect_handler,
                forbidden_handler,
            )

            try:
                app.add_exception_handler(401, auth_redirect_handler)
                app.add_exception_handler(403, forbidden_handler)
                app.add_middleware(CSRFMiddleware)
            except RuntimeError:
                pass  # Already started — middleware was added in __init__

        # Add API bearer-token middleware if not already added in __init__
        if not getattr(self, "_api_token_middleware_added", False):
            from fastapi_admin_kit.api.middleware import AccessTokenMiddleware

            try:
                app.add_middleware(AccessTokenMiddleware)
                self._api_token_middleware_added = True
            except RuntimeError:
                app.middleware_stack = None
                app.add_middleware(AccessTokenMiddleware)
                self._api_token_middleware_added = True

        # Add per-request session middleware
        if app is not None and not getattr(self, "_session_middleware_added", False):
            from fastapi_admin_kit.db import SessionMiddleware

            try:
                app.add_middleware(SessionMiddleware)
                self._session_middleware_added = True
            except RuntimeError:
                # Stack already built (e.g. the lifespan startup scope). Force a
                # rebuild on the next request so the middleware is included.
                app.middleware_stack = None
                app.add_middleware(SessionMiddleware)
                self._session_middleware_added = True

        # Add audit context middleware
        if app is not None and not getattr(self, "_audit_middleware_added", False):
            from fastapi_admin_kit.audit.middleware import (
                AuditContextMiddleware,
            )

            try:
                app.add_middleware(AuditContextMiddleware)
                self._audit_middleware_added = True
            except RuntimeError:
                app.middleware_stack = None
                app.add_middleware(AuditContextMiddleware)
                self._audit_middleware_added = True

        # 0. Validate secret_key strength
        if not self.router.secret_key:
            raise ConfigError(
                "Admin secret_key is required. Pass a strong secret (≥32 chars) "
                "via Admin(secret_key=...) or the SECRET_KEY environment variable."
            )
        if len(self.router.secret_key) < 32:
            raise ConfigError(
                f"Admin secret_key is too short ({len(self.router.secret_key)} chars). "
                "Must be at least 32 characters for secure signing."
            )

        # 1. Validate auth_model satisfies AdminUserProtocol
        self.config.auth.validate_auth_model()

        # 2. Database tables should be created via Alembic migrations
        #    ``create_tables()`` is the public entry point that projects can
        #    also call from their lifespan before ``admin.setup(app)`` —
        #    using it here keeps both code paths in sync (and is the only
        #    way the custom-auth_model skip is applied).
        await self.create_tables()

        # 2.1 Preflight: if AI is enabled but the tables are genuinely missing
        # (Alembic / SKIP_CREATE_TABLES mode), warn loudly but never block boot.
        if self._ai_enabled:
            try:
                missing = await self.database._missing_tables(
                    self._ai_enabled, list(AI_TABLE_NAMES)
                )
            except Exception:  # pragma: no cover - inspector failures must not block boot
                missing = set()
            if missing:
                logger.warning(
                    "ai_enabled=True but these tables are missing: %s. "
                    "AI chat/usage tracking will fail until the schema is migrated — run "
                    "`alembic revision --autogenerate && alembic upgrade head` "
                    "(or `fak migrate admin_ai_conversations` in dev mode).",
                    ", ".join(sorted(missing)),
                )

        # 3. Seed default roles
        await self.database._seed_roles(self.seed_roles, self.seed_roles_overwrite)

        # 4. Create and store session backend
        self._session_backend = self.database._init_session_backend(
            secret_key=self.router.secret_key,
            session_ttl=self.config.auth.session_ttl,
            cookie_name=self.config.auth.session_cookie_name,
            secure=self.config.auth.session_secure,
        )

        # 5. Store backends and config on app.state
        self._wire_app_state(app)

        # 5.1 Auto-wire the notification system (mount router + service on
        # app.state) unless the user configured it manually already.
        self._setup_notifications(app)

        # 6. Mount static files
        self._mount_static(app)

        # 7. Initialise Jinja2
        self._init_jinja(app)

        # 8. Auto-register built-in admin models (before auto_discover)
        self._register_builtin_models()

        # 8.1 Auto-discover user models
        if self.config.behavior.auto_discover:
            self.registry.auto_discover(exclude_tables=self._excluded_builtin_tables())

        # 8.2 Apply skip_models — mark listed models to hide from admin
        skip_models = self.config.behavior.skip_models
        # Built-in internal models are always hidden from admin
        # Note: model class names match table names (e.g., admin_refresh_tokens)
        default_skip = {"admin_refresh_tokens", "admin_user_permissions", "admin_user_totp"}
        all_skip = default_skip | skip_models | self._excluded_builtin_tables()
        skip_lower = {s.lower() for s in all_skip}
        for registered in self.registry.all():
            model_name = getattr(registered.model, "__name__", "").lower()
            if model_name in skip_lower:
                registered.admin.skip_auto_routes = True

        # 8.3 Attach audit event listeners (after registry is populated)
        session_factory = getattr(app.state, "admin_session_factory", None)
        if session_factory is not None:
            self.backend.audit.attach_listeners(session_factory, self.registry)

        # 9. Validate require_tags
        if self.config.nav.require_tags:
            self._validate_tags()

        # 9.1 Add AI nav group before building sidebar
        if self._ai_enabled:
            self._add_ai_nav_group()

        # 10. Build sidebar structure (once at startup)
        self._nav_groups_built = self._build_sidebar()
        self.template._nav_groups_built = self._nav_groups_built
        if self._jinja_env:
            self._jinja_env.env.globals["nav_groups"] = self._nav_groups_built

        # 11. Build and mount routers
        self._build_router(app)

        # 12. Setup AI routes if enabled
        if self._ai_enabled:
            self._setup_ai_routes(app)

    # ------------------------------------------------------------------
    # Register
    # ------------------------------------------------------------------

    def register(
        self,
        model: type,
        admin_class: type[ModelAdmin] | None = None,
    ) -> _RegistrationProxy | RegisteredModel:
        """Register a model with the admin.

        Usage::

            admin.register(Product)

            @admin.register(Product)
            class ProductAdmin(ModelAdmin):
                list_display = ["name", "price"]
        """
        if admin_class is not None:
            registered = self.registry.register(model, admin_class)
        else:
            registered = self.registry.register(model)
        if self._jinja_env:
            self._jinja_env.env.globals["registered_models"] = self.registry.all()
            if self._nav_groups_built:
                self._nav_groups_built = self._build_sidebar()
                self.template._nav_groups_built = self._nav_groups_built
                self._jinja_env.env.globals["nav_groups"] = self._nav_groups_built
        if admin_class is not None:
            return registered
        return _RegistrationProxy(self, registered)

    # ------------------------------------------------------------------
    # Lifespan
    # ------------------------------------------------------------------

    @asynccontextmanager
    async def lifespan(self, app: FastAPI) -> AsyncIterator[None]:
        """FastAPI lifespan context manager.

        Usage::

            app = FastAPI(lifespan=admin.lifespan)
        """
        await self.setup(app)
        yield

    # ------------------------------------------------------------------
    # Query helpers
    # ------------------------------------------------------------------

    def get_registered(self, table_name: str) -> RegisteredModel | None:
        """Get a registered model by table name."""
        return self.registry.get(table_name)

    def all_registered(self) -> list[RegisteredModel]:
        """Get all registered models."""
        return self.registry.all()

    def unregister(self, model: type) -> None:
        """Unregister a model so it can be re-registered with a custom admin class.

        Useful for overriding built-in admin models::

            from fastapi_admin_kit.auth.models import User
            from fastapi_admin_kit.admin.builtin_models import UserAdmin

            class MyUserAdmin(UserAdmin):
                list_display = ["id", "email", "full_name"]

            admin.unregister(User)
            admin.register(User, MyUserAdmin)
        """
        table_name = model.__tablename__
        self.registry._models.pop(table_name, None)

    # ------------------------------------------------------------------
    # Internal wiring
    # ------------------------------------------------------------------

    def _setup_redis(self, app: FastAPI) -> None:
        """Wire the optional Redis-backed caching/rate-limiting integration.

        Checks ``REDIS_URL`` at startup: when present (and the
        ``fastapi-redis-sdk`` is installed) the SDK's lifespan wrapper plus
        caching/rate-limiting support are registered on the app. Otherwise the
        admin falls back to the existing in-memory rate limiter and no cache
        middleware — full backward compatibility.
        """
        if self._redis_wired:
            return
        self._redis_wired = True

        from fastapi_admin_kit.redis import (
            redis_configured,
            redis_enabled,
            setup_redis,
        )

        self.redis_configured = redis_configured()
        self.redis_enabled = redis_enabled()

        if not self.redis_enabled:
            return

        setup_redis(
            app,
            cache_enabled=self.cache_config.enabled,
            cache_ttl=self.cache_config.ttl,
            rate_limiting=True,
        )

    def _validate_auth_model(self) -> None:
        """Validate that auth_model satisfies AdminUserProtocol."""
        self.config.auth.validate_auth_model()

    def _wire_app_state(self, app: FastAPI) -> None:
        """Store backends and configuration on app.state as typed AdminState."""
        from fastapi_admin_kit.admin.state import AdminState

        admin_config = {
            "title": self.config.ui.title,
            "logo_url": self.config.ui.logo_url,
            "favicon_url": self.config.ui.favicon_url,
            "primary_color": self.config.ui.primary_color,
            "primary_color_dark": self.config.ui.primary_color_dark,
            "dark_mode_default": self.config.ui.dark_mode_default,
            "per_page_default": self.config.ui.per_page_default,
            "session_ttl": self.config.auth.session_ttl,
            "access_token_ttl": self.config.auth.access_token_ttl,
            "api_token_middleware": self.config.auth.api_token_middleware,
            "api_token_strict": self.config.auth.api_token_strict,
            "trusted_proxies": self.config.auth.trusted_proxies,
            "audit_retention_days": self.config.audit.audit_retention_days,
            "dashboard_stats": self.config.behavior.dashboard_stats,
            "dashboard_charts": self.config.behavior.dashboard_charts,
            "admin_path": self.router.admin_path,
            "superuser_emails": self.config.auth.superuser_emails,
            "ui_config": self.config.ui.apply_to_template_context(),
        }
        if self.config.ui.theme:
            admin_config.update(self.config.ui.theme.to_context())

        # Create session factory if engine is available
        db_session = None
        session_factory = None
        connection = None
        engine = self.database.engine
        if engine is not None:
            connection = self.backend.database.create_connection()
            session_factory = self.backend.database.create_session_factory(connection)
            # Legacy fallback — a single session for backward compat. It is a
            # backend-agnostic SessionBackend, exactly like the per-request one.
            db_session = session_factory()

        # Inject auth_model and backend into the auth backend for ORM-agnostic queries.
        # BuiltinAuthBackend uses the QueryBackend (select/where/options) and the
        # DatabaseBackend's session_adapter_class via as_session_backend, so it
        # must not import sqlalchemy directly.
        if self.config.auth.auth_backend is not None:
            if self.config.auth.auth_model is not None:
                try:
                    self.config.auth.auth_backend._auth_model = self.config.auth.auth_model
                except AttributeError:
                    pass
            # Wire the composite backend and its query adapter — supports both
            # BuiltinAuthBackend (stores _backend/_query_backend) and any
            # custom backend that exposes the same attributes.
            for attr, value in (
                ("_backend", self.backend),
                ("backend", self.backend),
                ("_query_backend", getattr(self.backend, "query", None)),
                ("query_backend", getattr(self.backend, "query", None)),
            ):
                try:
                    setattr(self.config.auth.auth_backend, attr, value)
                except Exception:
                    pass

        state = AdminState(
            engine=engine,
            session_backend=self._session_backend,
            auth_backend=self.config.auth.auth_backend,
            storage=self.config.storage.storage,
            registry=self.registry,
            db_session=db_session,
            config=admin_config,
            jinja_env=self._jinja_env,
            admin_instance=self,
            secret_key=self.router.secret_key,
            session_samesite=self.config.auth.session_samesite,
            backend=self.backend,
        )

        # Store typed state as single attribute
        app.state.admin_state = state

        # Also store individual attributes for backward compatibility
        app.state.admin = self  # Admin instance (backward compat)
        app.state.admin_engine = state.engine
        app.state.admin_session_backend = state.session_backend
        app.state.admin_auth_backend = state.auth_backend
        app.state.admin_storage = state.storage
        app.state.admin_registry = state.registry
        app.state.admin_db_session = state.db_session
        app.state.admin_session_factory = session_factory
        app.state.admin_config = state.config
        app.state.admin_jinja_env = state.jinja_env
        # Unified signing-key source for sessions, CSRF, and JWT (see AdminState).
        app.state.admin_secret_key = state.secret_key
        # Multi-ORM backend: store composed backend and derive individual adapters
        app.state.admin_backend = self.backend
        app.state.admin_connection = connection
        app.state.admin_session_backend_class = self.backend.database.session_adapter_class
        app.state.admin_query_adapter = self.backend.query
        app.state.admin_introspection_adapter = self.backend.introspection
        app.state.admin_audit_backend = self.backend.audit

        # Wire the password hasher to the User model
        from fastapi_admin_kit.auth.models import User

        User.set_hasher(self.config.auth.get_hasher())

    def _mount_static(self, app: FastAPI) -> None:
        """Mount the static files directory and uploads directory."""
        static_dir = Path(__file__).parent.parent / "static"
        if static_dir.is_dir():
            app.mount(
                "/static",
                StaticFiles(directory=str(static_dir)),
                name="admin_static",
            )

        # Mount uploads directory if using LocalStorageBackend
        from fastapi_admin_kit.storage.local import LocalStorageBackend

        if isinstance(self.config.storage.storage, LocalStorageBackend):
            self.config.storage.storage.ensure_dir()
            app.mount(
                self.config.storage.uploads_url,
                StaticFiles(directory=str(self.config.storage.storage.upload_dir)),
                name="admin_uploads",
            )

    def _init_jinja(self, app: FastAPI) -> None:
        """Initialise the Jinja2 template environment.

        User-provided template dirs (from AdminConfig.template_dirs) are
        prepended so custom templates override built-in ones.
        """
        from starlette.templating import Jinja2Templates

        templates_dir = Path(__file__).parent.parent / "templates"
        user_dirs = list(getattr(self.template, "template_dirs", None) or [])
        all_dirs = [str(d) for d in user_dirs] + [str(templates_dir)]
        self._jinja_env = Jinja2Templates(directory=all_dirs)

        # Enable autoescape for XSS protection
        self._jinja_env.env.autoescape = True

        def slugify(s: str) -> str:
            return re.sub(r"[^\w]", "-", s, flags=re.A).strip("-").lower()

        def _attr(obj: Any, name: str) -> Any:
            return getattr(obj, name, "")

        self._jinja_env.env.filters["slugify"] = slugify

        def file_url(path: Any) -> str:
            """Map a stored file path to its public URL for ``src``/``href``.

            Storage keeps relative paths (``"documents/uuid.jpg"``) so that
            ``upload_dir / path`` and ``delete(path)`` work. Browsers need
            an absolute public URL (``"/uploads/documents/uuid.jpg"``), so
            templates must pipe stored values through this filter instead
            of rendering them raw.
            """
            if not path:
                return ""
            s = str(path)
            if s.startswith(("http://", "https://", "data:", "blob:")):
                return s
            storage = getattr(getattr(self.config, "storage", None), "storage", None)
            if storage is not None and hasattr(storage, "url"):
                try:
                    return storage.url(s)
                except Exception:
                    pass
            storage_cfg = getattr(self.config, "storage", None)
            base = (getattr(storage_cfg, "uploads_url", None) or "/uploads").rstrip("/")
            return f"{base}/{s.lstrip('/')}"

        self._jinja_env.env.filters["file_url"] = file_url
        self._jinja_env.env.globals["attr"] = _attr
        from fastapi_admin_kit.inspection import model_display_name

        self._jinja_env.env.globals["model_display_name"] = model_display_name
        self._jinja_env.env.globals["registered_models"] = self.registry.all()
        self._jinja_env.env.globals["admin_path"] = self.router.admin_path
        self._jinja_env.env.globals["notifications_api_path"] = getattr(
            self.config, "notifications_api_path", f"{self.router.admin_path}/notifications"
        )
        self._jinja_env.env.globals["notifications_list_path"] = getattr(
            self.config, "notifications_list_path", f"{self.router.admin_path}/admin_notifications/"
        )
        self._jinja_env.env.globals["notifications_enabled"] = self._enable_notification
        self._jinja_env.env.globals["nav_groups"] = self._nav_groups_built

        # CSRF token helper — reads from request.state (set by CSRFMiddleware)
        def _get_csrf_token(request) -> str:
            return getattr(request.state, "csrf_token", "")

        self._jinja_env.env.globals["get_csrf_token"] = _get_csrf_token

        # Flash messages helper (reads from session cookie directly)
        def _get_flash_messages(request) -> list[dict[str, str]]:
            try:
                session_backend = request.app.state.admin_session_backend
                cookie_name = getattr(session_backend, "cookie_name", "admin_session")
                raw = request.cookies.get(cookie_name)
                if not raw or not hasattr(session_backend, "load"):
                    return []
                data = session_backend.load(raw)
                if not isinstance(data, dict):
                    return []
                return data.pop("admin_flash", []) if "admin_flash" in data else []
            except Exception:
                return []

        self._jinja_env.env.globals["get_flash_messages"] = _get_flash_messages

        # Material Symbols icon helper
        _icon_map = {
            "home": "home",
            "chart-bar": "bar_chart",
            "clock": "schedule",
            "shield-check": "verified_user",
            "users": "group",
            "folder": "folder",
            "cube": "inventory_2",
            "shopping-cart": "shopping_cart",
            "magnifying-glass": "search",
            "chevron-right": "chevron_right",
            "chevron-left": "chevron_left",
            "chevron-up": "expand_less",
            "chevron-down": "expand_more",
            "ellipsis-vertical": "more_vert",
            "pencil": "edit",
            "trash": "delete",
            "x-mark": "close",
            "x-circle": "cancel",
            "check-circle": "check_circle",
            "check": "check",
            "plus": "add",
            "eye": "visibility",
            "bell": "notifications",
            "sun": "light_mode",
            "moon": "dark_mode",
            "bars-": "menu",
            "bars-3": "menu",
            "arrow-down-tray": "download",
            "arrow-up-tray": "upload",
            "arrow-path": "refresh",
            "paper-airplane": "send",
            "exclamation-triangle": "warning",
            "information-circle": "info",
            "document-text": "description",
            "arrow-down": "arrow_downward",
            "arrow-up": "arrow_upward",
            "bolt": "bolt",
            "cog-": "settings",
            "cog-6-tooth": "settings",
            "smart_toy": "smart_toy",
            "monitoring": "monitoring",
            "build": "build",
            "sparkles": "auto_awesome",
            "robot": "smart_toy",
        }

        def _icon(name: str, size: str = "", **kwargs) -> str:
            from markupsafe import Markup

            ms_name = _icon_map.get(name, name)
            css_class = kwargs.get("class", kwargs.get("css_class", ""))
            size_style = f' style="font-size: {size};"' if size else ""
            cls = f"material-symbols-outlined {css_class}".strip()
            return Markup(f'<span class="{cls}"{size_style}>{ms_name}</span>')

        self._jinja_env.env.globals["icon"] = _icon

        # Admin config global (used by templates for branding, dark mode, etc.)
        admin_cfg = {
            "title": self.config.ui.title,
            "logo_url": self.config.ui.logo_url,
            "favicon_url": self.config.ui.favicon_url,
            "primary_color": self.config.ui.primary_color,
            "primary_color_dark": self.config.ui.primary_color_dark,
            "dark_mode_default": self.config.ui.dark_mode_default,
            "admin_path": self.router.admin_path,
            "ai_enabled": self._ai_enabled,
        }
        self._jinja_env.env.globals["admin_config"] = admin_cfg

        # Static file cache-busting version hash
        import hashlib
        from pathlib import Path as _Path

        _static_dir = _Path(__file__).parent.parent / "static"
        _hash_data = b""
        for _f in (
            "css/style.css",
            "css/dist/tailwind.css",
            "js/admin.js",
        ):
            _fp = _static_dir / _f
            if _fp.is_file():
                _hash_data += _fp.read_bytes()
        _static_version = hashlib.md5(_hash_data).hexdigest()[:12] if _hash_data else "dev"
        self._jinja_env.env.globals["static_version"] = _static_version

        # Theme config globals
        self._jinja_env.env.globals["theme_preset"] = "editorial"
        if self.config.ui.theme:
            self._jinja_env.env.globals["theme_css"] = self.config.ui.theme.to_css_variables()
            self._jinja_env.env.globals["theme_font_import_url"] = (
                self.config.ui.theme.font_import_url
            )
            self._jinja_env.env.globals["theme_preset"] = self.config.ui.theme.preset
        self._jinja_env.env.globals["ui_config"] = self.config.ui.apply_to_template_context()

        app.state.admin_jinja_env = self._jinja_env

    def _setup_notifications(self, app: FastAPI) -> None:
        """Auto-wire the notification system into the admin.

        When ``enable_notification=True`` (the default) the admin creates a
        default :class:`NotificationService`, registers it on ``app.state``
        and mounts the notification router at the configured API path — so
        users do **not** need to call ``configure_notifications`` themselves.

        A service already configured by the user (via
        ``configure_notifications()`` or ``Admin(notification_service=...)``)
        is respected and never double-mounted.
        """
        if not self._enable_notification:
            return
        if getattr(app.state, "notification_service", None) is not None:
            return

        from fastapi_admin_kit.notifications.plugin import configure_notifications
        from fastapi_admin_kit.notifications.service import NotificationService

        service = self._notification_service
        if service is None:
            session_factory = getattr(app.state, "admin_session_factory", None)
            service = NotificationService(session_factory=session_factory)
            self._notification_service = service

        configure_notifications(app, service, prefix=self.config.notifications_api_path)

    def _build_router(self, app: FastAPI) -> None:
        """Build and mount routers for all registered models."""
        if self._router_built:
            return

        from fastapi_admin_kit.auth.router import router as auth_router
        from fastapi_admin_kit.router import build_model_router
        from fastapi_admin_kit.views.audit import router as audit_router
        from fastapi_admin_kit.views.profile import router as profile_router
        from fastapi_admin_kit.views.roles import router as roles_router
        from fastapi_admin_kit.views.settings import router as settings_router
        from fastapi_admin_kit.views.totp import router as totp_router
        from fastapi_admin_kit.views.users import router as users_router

        for registered in self.registry.all():
            if getattr(registered.admin, "skip_auto_routes", False):
                continue
            # API-only models (export_endpoint="api") get no admin HTML router.
            if getattr(registered.admin, "export_endpoint", None) == "api":
                continue
            model_router = build_model_router(registered, cache_config=self.cache_config)
            if model_router is None:
                continue
            app.include_router(model_router, prefix=self.router.admin_path)

        # Auth routes (login/logout)
        app.include_router(auth_router, prefix=self.router.admin_path)

        # Global search API
        from fastapi_admin_kit.api.search import router as search_api_router

        app.include_router(search_api_router, prefix=self.router.admin_path)

        # Audit, role management, settings, user management, profile, and 2FA routes
        app.include_router(audit_router, prefix=self.router.admin_path)
        app.include_router(roles_router, prefix=self.router.admin_path)
        app.include_router(settings_router, prefix=self.router.admin_path)
        app.include_router(users_router, prefix=self.router.admin_path)
        app.include_router(profile_router, prefix=self.router.admin_path)
        app.include_router(totp_router, prefix=self.router.admin_path)

        # Dashboard route
        from fastapi_admin_kit.views.dashboard import dashboard_view_factory

        dashboard_view = dashboard_view_factory(self)
        app.add_api_route(
            self.router.admin_path,
            dashboard_view,
            methods=["GET"],
            tags=["admin"],
            include_in_schema=False,
        )

        # JSON API for external frontend apps
        from fastapi_admin_kit.api import AdminAPIRouter

        api_router = AdminAPIRouter(registry=self.registry)
        app.include_router(api_router.build_router())

        self._router_built = True

    # ------------------------------------------------------------------
    # Built-in model registration
    # ------------------------------------------------------------------

    def _register_builtin_models(self) -> None:
        """Auto-register built-in admin models with default admin classes."""
        from fastapi_admin_kit.admin.builtin_models import (
            # UserPermissionAdmin,
            # UserTOTPAdmin,
            AuditLogAdmin,
            LoginAttemptAdmin,
            NotificationAdmin,
            NotificationLogAdmin,
            NotificationPreferenceAdmin,
            PermissionAdmin,
            RoleAdmin,
            UserAdmin,
        )
        from fastapi_admin_kit.migrations.models import (
            AuditLog,
            LoginAttempt,
            Notification,
            NotificationLog,
            NotificationPreference,
            Permission,
            Role,
            User,
        )

        if self._ai_enabled:
            from fastapi_admin_kit.admin.builtin_models import (
                AIConversationAdmin,
                AIMessageAdmin,
                AIUsageLogAdmin,
            )
            from fastapi_admin_kit.migrations.models import (
                AIConversation,
                AIMessage,
                AIUsageLog,
            )

        builtin_models = [
            (Role, RoleAdmin),
            # (RefreshToken, RefreshTokenAdmin),
            (Permission, PermissionAdmin),
            # (UserPermission, UserPermissionAdmin),
            # (UserTOTP, UserTOTPAdmin),
            (LoginAttempt, LoginAttemptAdmin),
            (AuditLog, AuditLogAdmin),
        ]

        # When a custom auth_model is provided, the built-in ``User`` model is
        # not registered: the project supplies its own user model and
        # ``UserAdmin`` (the built-in CRUD) does not match it. Skipping here
        # also keeps the registry consistent with the migration metadata, from
        # which the built-in ``admin_users`` table is removed in
        # ``_adapt_builtin_user_id_columns`` when a custom auth_model is set.
        from fastapi_admin_kit.migrations.models import User as BuiltinUser

        if self.config.auth.auth_model is None or self.config.auth.auth_model is BuiltinUser:
            builtin_models.insert(0, (User, UserAdmin))

        # Notification models are exposed under the "notifications" sidebar
        # group. Register them only when notifications are enabled so the
        # group never appears when enable_notification=False.
        if self._enable_notification:
            builtin_models += [
                (Notification, NotificationAdmin),
                (NotificationPreference, NotificationPreferenceAdmin),
                (NotificationLog, NotificationLogAdmin),
            ]

        for model, admin_class in builtin_models:
            if model.__tablename__ not in self.registry._models:
                self.registry.register(model, admin_class)

        if self._ai_enabled:
            ai_builtin_models = [
                (AIConversation, AIConversationAdmin),
                (AIMessage, AIMessageAdmin),
                (AIUsageLog, AIUsageLogAdmin),
            ]
            for model, admin_class in ai_builtin_models:
                if model.__tablename__ not in self.registry._models:
                    self.registry.register(model, admin_class)

    def _builtin_user_tables_to_skip(self) -> tuple[str, ...]:
        """Return the names of built-in tables that should NOT be created when
        a custom ``auth_model`` is configured.

        When the project supplies its own user model (``auth_model=``), the
        built-in ``admin_users`` table is skipped — the custom auth_model
        is the source of truth for user identity. ``admin_user_roles`` is
        kept and its ``user_id`` foreign key is retargeted to the custom
        auth_model's table at create time (see
        ``SqlAlchemyDatabaseBackend.adapt_auth_model``) so the junction
        links roles to the project's own user rows.

        The role/permission system (``admin_roles``,
        ``admin_role_permissions``, ``admin_permissions``) is kept so the
        admin can still manage granular per-table access.

        Returns an empty tuple when the built-in ``User`` is in use (default
        installation) or when no ``auth_model`` is configured.
        """
        from fastapi_admin_kit.migrations.models import User as BuiltinUser

        auth_model = self.config.auth.auth_model
        if auth_model is None or auth_model is BuiltinUser:
            return ()
        return ("admin_users",)

    async def create_tables(self) -> None:
        """Create all admin database tables, correctly handling a custom
        ``auth_model``.

        This is the public API projects should use in their ``lifespan`` (or
        startup hook) instead of calling
        ``AdminBase.metadata.create_all`` directly. Calling
        ``AdminBase.metadata.create_all`` directly bypasses the
        custom-auth_model logic and will create the default
        ``admin_users`` / ``admin_user_roles`` tables even when a project
        supplies its own user model.

        What this method does, in order:

        1. Validate the configured ``auth_model`` (raises ``ConfigError`` if
           it does not satisfy :class:`AdminUserProtocol`).
        2. Mirror the auth model's primary-key type onto the built-in
           log-pattern ``user_id`` columns (e.g. a UUID PK becomes a UUID
           ``user_id`` so the column type matches across tables).
        3. Call ``AdminDatabase._create_tables(extra_exclude_tables=...)``
           with the built-in user tables filtered out when a custom
           ``auth_model`` is configured. Honors ``SKIP_CREATE_TABLES=true``
           and the AI / notification feature flags exactly like
           ``Admin.setup()`` does.
        """
        self.config.auth.validate_auth_model()
        skip_create_tables = os.environ.get("SKIP_CREATE_TABLES", "false").lower() == "true"
        if skip_create_tables:
            logger.info("SKIP_CREATE_TABLES=true: skipping admin table creation")
            return
        self._adapt_builtin_user_id_columns()
        # Ask the configured backend to retarget the built-in user
        # relations (admin_user_roles FK, Role.users M2M, etc.) at the
        # custom auth_model. Each backend implements this against its own
        # ORM primitives; non-SQLA backends may no-op.
        #
        # Use ``self.database._backend`` (the backend that actually performs
        # DDL in ``_create_tables`` below) rather than the composite
        # ``self.backend.database`` so tests that swap
        # ``admin.database._backend`` with a fake do not mutate the global
        # SQLAlchemy mapper state. A backend without ``adapt_auth_model``
        # (e.g. a test fake or memory backend) is treated as no-op.
        database_backend = getattr(self.database, "_backend", None)
        adapt = getattr(database_backend, "adapt_auth_model", None)
        if adapt is not None and self.config.auth.auth_model is not None:
            try:
                from fastapi_admin_kit.migrations.models import User as BuiltinUser

                if self.config.auth.auth_model is not BuiltinUser:
                    adapt(self.config.auth.auth_model)
            except Exception as exc:  # pragma: no cover - defensive
                logger.warning("adapt_auth_model failed: %s", exc)
        extra_exclude = list(self._builtin_user_tables_to_skip())
        if extra_exclude:
            logger.info(
                "Custom auth_model=%s configured: skipping built-in admin "
                "tables %s at create_all (project supplies its own user "
                "table).",
                self.config.auth.auth_model,
                extra_exclude,
            )
        await self.database._create_tables(
            include_ai_tables=self._ai_enabled,
            extra_exclude_tables=extra_exclude or None,
        )

    def _adapt_builtin_user_id_columns(self) -> None:
        """
        Adapt the built-in log-pattern ``user_id`` columns in ``AdminBase.metadata``
        to match the primary-key type of the configured ``auth_model`` (see
        ``schemas/builtin.py``). When a project supplies a custom
        ``auth_model`` whose primary key differs from the built-in ``User``
        (e.g. a ``UUID`` PK), these columns are retyped to match so that
        ``metadata.create_all`` emits the correct DDL.

        This is a no-op for the default installation (``auth_model is None``
        or the built-in ``User``), which preserves backward compatibility
        and never alters existing schemas/migrations.

        When a custom ``auth_model`` is provided, the built-in ``admin_users``
        table is also removed from ``AdminBase.metadata`` (and the
        ``admin_user_roles`` junction) so that ``create_all`` does not emit
        DDL for the default schema. The custom user model — which must
        already be registered on a ``Base`` and present in the project's
        metadata — becomes the sole source of truth for the user table.
        """
        from fastapi_admin_kit.migrations.models import User as BuiltinUser

        auth_model = self.config.auth.auth_model
        if auth_model is None:
            return
        if auth_model is BuiltinUser:
            return

        metadata = BuiltinUser.__table__.metadata

        from sqlalchemy import inspect as sa_inspect

        pk_cols = sa_inspect(auth_model).primary_key
        if not pk_cols:
            return
        pk_col = pk_cols[0]
        new_type = pk_col.type
        logger.debug(
            "Adapting builtin user_id columns: auth_model=%s pk_col=%s pk_type=%r",
            auth_model,
            pk_col.name,
            new_type,
        )

        # user_email columns intentionally stay string (they store emails).
        log_tables = [
            "admin_audit_log",
            "admin_user_permissions",
            "admin_refresh_tokens",
            "admin_user_totp",
            "admin_notifications",
            "admin_notification_preferences",
            "admin_notification_logs",
            "admin_ai_usage_log",
            "admin_ai_conversations",
        ]
        for table_name in log_tables:
            table = metadata.tables.get(table_name)
            if table is None or "user_id" not in table.c:
                continue
            col = table.c["user_id"]
            logger.debug(
                "  %s.user_id: %r (%s) -> %r (%s)",
                table_name,
                col.type,
                type(col.type).__name__,
                new_type,
                type(new_type).__name__,
            )
            col.type = new_type

        # Note: the built-in ``admin_users`` table is excluded from
        # ``create_all`` at the call site in ``setup()`` via
        # ``AdminDatabase._create_tables(extra_exclude_tables=...)`` — we do
        # NOT remove it from ``AdminBase.metadata`` here because the
        # ``FacadeDict`` exposed by ``Base.metadata`` is immutable.
        #
        # FK retargeting on ``admin_user_roles`` and the ``Role.users`` M2M
        # re-binding are delegated to the configured backend via
        # ``backend.adapt_auth_model(auth_model)`` — this keeps ``core.py``
        # ORM-agnostic. Backends (SQLAlchemy) implement the actual column
        # and relationship mutations; memory/no-op backends can ignore.

        # Note: the built-in ``admin_users`` and ``admin_user_roles`` tables
        # are excluded from ``create_all`` at the call site in ``setup()`` via
        # ``AdminDatabase._create_tables(extra_exclude_tables=...)`` — we do
        # NOT remove them from ``AdminBase.metadata`` here because the
        # ``FacadeDict`` exposed by ``Base.metadata`` is immutable.

    # ------------------------------------------------------------------
    # AI Setup
    # ------------------------------------------------------------------

    def _excluded_builtin_tables(self) -> frozenset[str]:
        """Tables to hide from auto-discovery / default-skip lists.

        Always excludes internal tables (refresh tokens, user permissions,
        TOTP secrets, AI attachments). When AI is disabled, also excludes the
        three user-facing AI tables so they never leak into the sidebar/routes.
        When notifications are disabled, also excludes the notification tables
        so the "notifications" sidebar group never appears.

        When a custom ``auth_model`` is configured, also excludes the built-in
        ``admin_users`` model from auto-discovery (it is never created when a
        custom auth_model is set — see ``_builtin_user_tables_to_skip``). The
        ``admin_user_roles`` junction table is NOT excluded: it is kept and its
        ``user_id`` foreign key is retargeted to the custom auth_model's table
        so role relationships still work end-to-end.
        """
        excluded = set(INTERNAL_TABLE_NAMES)  # incl. admin_ai_attachments
        if not self._ai_enabled:
            excluded |= AI_TABLE_NAMES
        if not self._enable_notification:
            excluded |= NOTIFICATION_TABLE_NAMES
        if self._builtin_user_tables_to_skip():
            excluded |= {"admin_users"}
        return frozenset(excluded)

    def _add_ai_nav_group(self) -> None:
        """Add the AI nav group to nav_groups before sidebar build.

        Idempotent: skips if an ``ai`` group already exists (setup can run
        more than once, e.g. across tests).
        """
        from fastapi_admin_kit.nav import NavGroupConfig, NavItemConfig

        if any(g.tag == "ai" for g in self.config.nav.nav_groups):
            return

        ai_nav = NavGroupConfig(
            tag="ai",
            label="AI",
            icon="smart_toy",
            order=900,
            collapsed_by_default=False,
            extra_items=[
                NavItemConfig(
                    label="Chat",
                    url="/admin/ai/chat",
                    icon="chat",
                    order=1,
                ),
                NavItemConfig(
                    label="Dashboard",
                    url="/admin/ai/dashboard",
                    icon="monitoring",
                    order=2,
                ),
                NavItemConfig(
                    label="Logs",
                    url="/admin/ai/logs",
                    icon="description",
                    order=3,
                ),
                NavItemConfig(
                    label="Tools",
                    url="/admin/ai/tools",
                    icon="build",
                    order=4,
                ),
                NavItemConfig(
                    label="Agents",
                    url="/admin/ai/agents",
                    icon="smart_toy",
                    order=5,
                ),
            ],
        )
        self.config.nav.nav_groups.append(ai_nav)

    def _setup_ai_routes(self, app: FastAPI) -> None:
        """Initialize AI agents and mount AI routes."""
        from fastapi_admin_kit.ai.plugin import AIPlugin

        plugin = AIPlugin(agents=self._ai_config.agents if self._ai_config else [])
        plugin.on_startup(self)

        if self._ai_config and self._ai_config.dashboard_enabled:
            app.include_router(plugin.get_routes(), prefix=self.router.admin_path)

    # ------------------------------------------------------------------
    # Tags validation
    # ------------------------------------------------------------------

    def _validate_tags(self) -> None:
        """Raise ConfigError if any registered model has no tag (when require_tags=True)."""
        untagged: list[str] = []
        for registered in self.registry.all():
            admin = registered.admin
            tags = getattr(admin, "tags", None)
            tag = getattr(admin, "tag", None)
            if not tags and not tag:
                untagged.append(registered.table_name)
        if untagged:
            raise ConfigError(
                "require_tags=True but the following models have no tag: "
                + ", ".join(sorted(untagged))
            )

    # ------------------------------------------------------------------
    # Sidebar
    # ------------------------------------------------------------------

    def _build_sidebar(self) -> list:
        """Build the sidebar group structure once at startup."""
        from fastapi_admin_kit.nav import DefaultSidebarBuilder

        builder = self.config.nav.sidebar_builder or DefaultSidebarBuilder()
        return builder.build(
            self.registry.all(),
            self.config.nav.nav_groups,
            admin_path=self.router.admin_path,
        )

    async def build_sidebar_context(
        self,
        request: Any,
        user: Any = None,
        permissions_map: dict | None = None,
    ) -> dict:
        """Build per-request sidebar context (RBAC filter + permissions map)."""
        return await self.template.build_sidebar_context(
            request, user=user, permissions_map=permissions_map
        )

    async def sidebar_template_kwargs(self, request: Any) -> dict[str, Any]:
        """Thin wrapper — returns sidebar kwargs for TemplateResponse contexts."""
        return await self.template.sidebar_template_kwargs(request)

    async def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
        """Inject nav_groups + permissions_map into a template context dict."""
        return await self.template.apply_sidebar_context(request, user, context)

setup(app=None) async

Run all startup wiring: create tables, seed roles, mount assets.

This must be called once during application lifespan, typically via the :meth:lifespan context manager.

Source code in fastapi_admin_kit/admin/core.py

async def setup(self, app: FastAPI | None = None) -> None:
    """Run all startup wiring: create tables, seed roles, mount assets.

    This must be called once during application lifespan, typically via
    the :meth:`lifespan` context manager.
    """
    if app is not None:
        self._app = app

    if self._app is None:
        raise ConfigError(
            "Admin requires a FastAPI app instance. Pass app= or call setup(app=)."
        )

    self.database._ensure_engine()

    if self.database.engine is None:
        raise ConfigError(
            "Admin requires a SQLAlchemy engine. Pass engine= or database_config= to Admin()."
        )

    app = self._app

    # Wire Redis when the app was supplied after construction (e.g. via
    # ``await admin.setup(app)`` inside a manual lifespan).
    if not getattr(self, "_redis_wired", False):
        self._setup_redis(app)

    # Add CSRF middleware if not already added in __init__
    if not getattr(self, "_csrf_middleware_added", False):
        from fastapi_admin_kit.auth.csrf import (
            CSRFMiddleware,
            auth_redirect_handler,
            forbidden_handler,
        )

        try:
            app.add_exception_handler(401, auth_redirect_handler)
            app.add_exception_handler(403, forbidden_handler)
            app.add_middleware(CSRFMiddleware)
        except RuntimeError:
            pass  # Already started — middleware was added in __init__

    # Add API bearer-token middleware if not already added in __init__
    if not getattr(self, "_api_token_middleware_added", False):
        from fastapi_admin_kit.api.middleware import AccessTokenMiddleware

        try:
            app.add_middleware(AccessTokenMiddleware)
            self._api_token_middleware_added = True
        except RuntimeError:
            app.middleware_stack = None
            app.add_middleware(AccessTokenMiddleware)
            self._api_token_middleware_added = True

    # Add per-request session middleware
    if app is not None and not getattr(self, "_session_middleware_added", False):
        from fastapi_admin_kit.db import SessionMiddleware

        try:
            app.add_middleware(SessionMiddleware)
            self._session_middleware_added = True
        except RuntimeError:
            # Stack already built (e.g. the lifespan startup scope). Force a
            # rebuild on the next request so the middleware is included.
            app.middleware_stack = None
            app.add_middleware(SessionMiddleware)
            self._session_middleware_added = True

    # Add audit context middleware
    if app is not None and not getattr(self, "_audit_middleware_added", False):
        from fastapi_admin_kit.audit.middleware import (
            AuditContextMiddleware,
        )

        try:
            app.add_middleware(AuditContextMiddleware)
            self._audit_middleware_added = True
        except RuntimeError:
            app.middleware_stack = None
            app.add_middleware(AuditContextMiddleware)
            self._audit_middleware_added = True

    # 0. Validate secret_key strength
    if not self.router.secret_key:
        raise ConfigError(
            "Admin secret_key is required. Pass a strong secret (≥32 chars) "
            "via Admin(secret_key=...) or the SECRET_KEY environment variable."
        )
    if len(self.router.secret_key) < 32:
        raise ConfigError(
            f"Admin secret_key is too short ({len(self.router.secret_key)} chars). "
            "Must be at least 32 characters for secure signing."
        )

    # 1. Validate auth_model satisfies AdminUserProtocol
    self.config.auth.validate_auth_model()

    # 2. Database tables should be created via Alembic migrations
    #    ``create_tables()`` is the public entry point that projects can
    #    also call from their lifespan before ``admin.setup(app)`` —
    #    using it here keeps both code paths in sync (and is the only
    #    way the custom-auth_model skip is applied).
    await self.create_tables()

    # 2.1 Preflight: if AI is enabled but the tables are genuinely missing
    # (Alembic / SKIP_CREATE_TABLES mode), warn loudly but never block boot.
    if self._ai_enabled:
        try:
            missing = await self.database._missing_tables(
                self._ai_enabled, list(AI_TABLE_NAMES)
            )
        except Exception:  # pragma: no cover - inspector failures must not block boot
            missing = set()
        if missing:
            logger.warning(
                "ai_enabled=True but these tables are missing: %s. "
                "AI chat/usage tracking will fail until the schema is migrated — run "
                "`alembic revision --autogenerate && alembic upgrade head` "
                "(or `fak migrate admin_ai_conversations` in dev mode).",
                ", ".join(sorted(missing)),
            )

    # 3. Seed default roles
    await self.database._seed_roles(self.seed_roles, self.seed_roles_overwrite)

    # 4. Create and store session backend
    self._session_backend = self.database._init_session_backend(
        secret_key=self.router.secret_key,
        session_ttl=self.config.auth.session_ttl,
        cookie_name=self.config.auth.session_cookie_name,
        secure=self.config.auth.session_secure,
    )

    # 5. Store backends and config on app.state
    self._wire_app_state(app)

    # 5.1 Auto-wire the notification system (mount router + service on
    # app.state) unless the user configured it manually already.
    self._setup_notifications(app)

    # 6. Mount static files
    self._mount_static(app)

    # 7. Initialise Jinja2
    self._init_jinja(app)

    # 8. Auto-register built-in admin models (before auto_discover)
    self._register_builtin_models()

    # 8.1 Auto-discover user models
    if self.config.behavior.auto_discover:
        self.registry.auto_discover(exclude_tables=self._excluded_builtin_tables())

    # 8.2 Apply skip_models — mark listed models to hide from admin
    skip_models = self.config.behavior.skip_models
    # Built-in internal models are always hidden from admin
    # Note: model class names match table names (e.g., admin_refresh_tokens)
    default_skip = {"admin_refresh_tokens", "admin_user_permissions", "admin_user_totp"}
    all_skip = default_skip | skip_models | self._excluded_builtin_tables()
    skip_lower = {s.lower() for s in all_skip}
    for registered in self.registry.all():
        model_name = getattr(registered.model, "__name__", "").lower()
        if model_name in skip_lower:
            registered.admin.skip_auto_routes = True

    # 8.3 Attach audit event listeners (after registry is populated)
    session_factory = getattr(app.state, "admin_session_factory", None)
    if session_factory is not None:
        self.backend.audit.attach_listeners(session_factory, self.registry)

    # 9. Validate require_tags
    if self.config.nav.require_tags:
        self._validate_tags()

    # 9.1 Add AI nav group before building sidebar
    if self._ai_enabled:
        self._add_ai_nav_group()

    # 10. Build sidebar structure (once at startup)
    self._nav_groups_built = self._build_sidebar()
    self.template._nav_groups_built = self._nav_groups_built
    if self._jinja_env:
        self._jinja_env.env.globals["nav_groups"] = self._nav_groups_built

    # 11. Build and mount routers
    self._build_router(app)

    # 12. Setup AI routes if enabled
    if self._ai_enabled:
        self._setup_ai_routes(app)

register(model, admin_class=None)

Register a model with the admin.

Usage::

admin.register(Product)

@admin.register(Product)
class ProductAdmin(ModelAdmin):
    list_display = ["name", "price"]

Source code in fastapi_admin_kit/admin/core.py

def register(
    self,
    model: type,
    admin_class: type[ModelAdmin] | None = None,
) -> _RegistrationProxy | RegisteredModel:
    """Register a model with the admin.

    Usage::

        admin.register(Product)

        @admin.register(Product)
        class ProductAdmin(ModelAdmin):
            list_display = ["name", "price"]
    """
    if admin_class is not None:
        registered = self.registry.register(model, admin_class)
    else:
        registered = self.registry.register(model)
    if self._jinja_env:
        self._jinja_env.env.globals["registered_models"] = self.registry.all()
        if self._nav_groups_built:
            self._nav_groups_built = self._build_sidebar()
            self.template._nav_groups_built = self._nav_groups_built
            self._jinja_env.env.globals["nav_groups"] = self._nav_groups_built
    if admin_class is not None:
        return registered
    return _RegistrationProxy(self, registered)

lifespan(app) async

FastAPI lifespan context manager.

Usage::

app = FastAPI(lifespan=admin.lifespan)

Source code in fastapi_admin_kit/admin/core.py

@asynccontextmanager
async def lifespan(self, app: FastAPI) -> AsyncIterator[None]:
    """FastAPI lifespan context manager.

    Usage::

        app = FastAPI(lifespan=admin.lifespan)
    """
    await self.setup(app)
    yield

get_registered(table_name)

Get a registered model by table name.

Source code in fastapi_admin_kit/admin/core.py

def get_registered(self, table_name: str) -> RegisteredModel | None:
    """Get a registered model by table name."""
    return self.registry.get(table_name)

all_registered()

Get all registered models.

Source code in fastapi_admin_kit/admin/core.py

def all_registered(self) -> list[RegisteredModel]:
    """Get all registered models."""
    return self.registry.all()

unregister(model)

Unregister a model so it can be re-registered with a custom admin class.

Useful for overriding built-in admin models::

from fastapi_admin_kit.auth.models import User
from fastapi_admin_kit.admin.builtin_models import UserAdmin

class MyUserAdmin(UserAdmin):
    list_display = ["id", "email", "full_name"]

admin.unregister(User)
admin.register(User, MyUserAdmin)

Source code in fastapi_admin_kit/admin/core.py

def unregister(self, model: type) -> None:
    """Unregister a model so it can be re-registered with a custom admin class.

    Useful for overriding built-in admin models::

        from fastapi_admin_kit.auth.models import User
        from fastapi_admin_kit.admin.builtin_models import UserAdmin

        class MyUserAdmin(UserAdmin):
            list_display = ["id", "email", "full_name"]

        admin.unregister(User)
        admin.register(User, MyUserAdmin)
    """
    table_name = model.__tablename__
    self.registry._models.pop(table_name, None)

create_tables() async

Create all admin database tables, correctly handling a custom auth_model.

This is the public API projects should use in their lifespan (or startup hook) instead of calling AdminBase.metadata.create_all directly. Calling AdminBase.metadata.create_all directly bypasses the custom-auth_model logic and will create the default admin_users / admin_user_roles tables even when a project supplies its own user model.

What this method does, in order:

1. Validate the configured auth_model (raises ConfigError if it does not satisfy :class:AdminUserProtocol).
2. Mirror the auth model's primary-key type onto the built-in log-pattern user_id columns (e.g. a UUID PK becomes a UUID user_id so the column type matches across tables).
3. Call AdminDatabase._create_tables(extra_exclude_tables=...) with the built-in user tables filtered out when a custom auth_model is configured. Honors SKIP_CREATE_TABLES=true and the AI / notification feature flags exactly like Admin.setup() does.

Source code in fastapi_admin_kit/admin/core.py

async def create_tables(self) -> None:
    """Create all admin database tables, correctly handling a custom
    ``auth_model``.

    This is the public API projects should use in their ``lifespan`` (or
    startup hook) instead of calling
    ``AdminBase.metadata.create_all`` directly. Calling
    ``AdminBase.metadata.create_all`` directly bypasses the
    custom-auth_model logic and will create the default
    ``admin_users`` / ``admin_user_roles`` tables even when a project
    supplies its own user model.

    What this method does, in order:

    1. Validate the configured ``auth_model`` (raises ``ConfigError`` if
       it does not satisfy :class:`AdminUserProtocol`).
    2. Mirror the auth model's primary-key type onto the built-in
       log-pattern ``user_id`` columns (e.g. a UUID PK becomes a UUID
       ``user_id`` so the column type matches across tables).
    3. Call ``AdminDatabase._create_tables(extra_exclude_tables=...)``
       with the built-in user tables filtered out when a custom
       ``auth_model`` is configured. Honors ``SKIP_CREATE_TABLES=true``
       and the AI / notification feature flags exactly like
       ``Admin.setup()`` does.
    """
    self.config.auth.validate_auth_model()
    skip_create_tables = os.environ.get("SKIP_CREATE_TABLES", "false").lower() == "true"
    if skip_create_tables:
        logger.info("SKIP_CREATE_TABLES=true: skipping admin table creation")
        return
    self._adapt_builtin_user_id_columns()
    # Ask the configured backend to retarget the built-in user
    # relations (admin_user_roles FK, Role.users M2M, etc.) at the
    # custom auth_model. Each backend implements this against its own
    # ORM primitives; non-SQLA backends may no-op.
    #
    # Use ``self.database._backend`` (the backend that actually performs
    # DDL in ``_create_tables`` below) rather than the composite
    # ``self.backend.database`` so tests that swap
    # ``admin.database._backend`` with a fake do not mutate the global
    # SQLAlchemy mapper state. A backend without ``adapt_auth_model``
    # (e.g. a test fake or memory backend) is treated as no-op.
    database_backend = getattr(self.database, "_backend", None)
    adapt = getattr(database_backend, "adapt_auth_model", None)
    if adapt is not None and self.config.auth.auth_model is not None:
        try:
            from fastapi_admin_kit.migrations.models import User as BuiltinUser

            if self.config.auth.auth_model is not BuiltinUser:
                adapt(self.config.auth.auth_model)
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning("adapt_auth_model failed: %s", exc)
    extra_exclude = list(self._builtin_user_tables_to_skip())
    if extra_exclude:
        logger.info(
            "Custom auth_model=%s configured: skipping built-in admin "
            "tables %s at create_all (project supplies its own user "
            "table).",
            self.config.auth.auth_model,
            extra_exclude,
        )
    await self.database._create_tables(
        include_ai_tables=self._ai_enabled,
        extra_exclude_tables=extra_exclude or None,
    )

build_sidebar_context(request, user=None, permissions_map=None) async

Build per-request sidebar context (RBAC filter + permissions map).

Source code in fastapi_admin_kit/admin/core.py

async def build_sidebar_context(
    self,
    request: Any,
    user: Any = None,
    permissions_map: dict | None = None,
) -> dict:
    """Build per-request sidebar context (RBAC filter + permissions map)."""
    return await self.template.build_sidebar_context(
        request, user=user, permissions_map=permissions_map
    )

sidebar_template_kwargs(request) async

Thin wrapper — returns sidebar kwargs for TemplateResponse contexts.

Source code in fastapi_admin_kit/admin/core.py

async def sidebar_template_kwargs(self, request: Any) -> dict[str, Any]:
    """Thin wrapper — returns sidebar kwargs for TemplateResponse contexts."""
    return await self.template.sidebar_template_kwargs(request)

apply_sidebar_context(request, user, context) async

Inject nav_groups + permissions_map into a template context dict.

Source code in fastapi_admin_kit/admin/core.py

async def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
    """Inject nav_groups + permissions_map into a template context dict."""
    return await self.template.apply_sidebar_context(request, user, context)

AdminConfig

fastapi_admin_kit.admin.AdminConfig

Orchestrates all configuration classes with validation.

Source code in fastapi_admin_kit/admin/admin_config.py

class AdminConfig:
    """Orchestrates all configuration classes with validation."""

    def __init__(
        self,
        ui: UIConfig | None = None,
        auth: AuthConfig | None = None,
        audit: AuditConfig | None = None,
        behavior: BehaviorConfig | None = None,
        storage: StorageConfig | None = None,
        nav: NavConfig | None = None,
        template_dirs: list[str] | None = None,
        ai_chat: AIChatConfig | None = None,
        cache: CacheConfig | None = None,
    ):
        self.ui = ui or UIConfig()
        self.auth = auth or AuthConfig()
        self.audit = audit or AuditConfig()
        self.behavior = behavior or BehaviorConfig()
        self.storage = storage or StorageConfig()
        self.nav = nav or NavConfig()
        self.template_dirs = template_dirs or []
        self.ai_chat = ai_chat or AIChatConfig()
        self.cache = cache or CacheConfig()

    def validate_all(self) -> None:
        """Validate all configuration components."""
        self.auth.validate_auth_model()
        self.audit.validate_audit_config()
        self.storage.validate_storage_config()
        self.nav.validate_nav_config()
        self.cache.validate_cache_config()

    def get_ui_context(self) -> dict:
        """Get UI configuration for template context."""
        return self.ui.apply_to_template_context()

    def get_branding_config(self) -> dict:
        """Get branding configuration."""
        return {
            "title": self.ui.title,
            "logo_url": self.ui.logo_url,
            "favicon_url": self.ui.favicon_url,
            "primary_color": self.ui.primary_color,
            "primary_color_dark": self.ui.primary_color_dark,
            "dark_mode_default": self.ui.dark_mode_default,
        }

    def get_session_config(self) -> dict:
        """Get session configuration."""
        return {
            "session_ttl": self.auth.session_ttl,
            "session_cookie_name": self.auth.session_cookie_name,
            "session_secure": self.auth.session_secure,
        }

    def get_audit_config(self) -> dict:
        """Get audit configuration."""
        return {
            "audit_retention_days": self.audit.audit_retention_days,
        }

    def get_storage_config(self) -> dict:
        """Get storage configuration."""
        return {
            "uploads_url": self.storage.uploads_url,
        }

    def get_behavior_config(self) -> dict:
        """Get behavior configuration."""
        return {
            "auto_discover": self.behavior.auto_discover,
            "dashboard_stats": self.behavior.dashboard_stats,
            "dashboard_charts": self.behavior.dashboard_charts,
        }

    def get_nav_config(self) -> dict:
        """Get navigation configuration."""
        return {
            "nav_groups": self.nav.nav_groups,
            "sidebar_builder": self.nav.sidebar_builder,
            "require_tags": self.nav.require_tags,
            "dashboard_permission": self.nav.dashboard_permission,
            "settings_permission": self.nav.settings_permission,
        }

validate_all()

Validate all configuration components.

Source code in fastapi_admin_kit/admin/admin_config.py

def validate_all(self) -> None:
    """Validate all configuration components."""
    self.auth.validate_auth_model()
    self.audit.validate_audit_config()
    self.storage.validate_storage_config()
    self.nav.validate_nav_config()
    self.cache.validate_cache_config()

get_ui_context()

Get UI configuration for template context.

Source code in fastapi_admin_kit/admin/admin_config.py

def get_ui_context(self) -> dict:
    """Get UI configuration for template context."""
    return self.ui.apply_to_template_context()

get_branding_config()

Get branding configuration.

Source code in fastapi_admin_kit/admin/admin_config.py

def get_branding_config(self) -> dict:
    """Get branding configuration."""
    return {
        "title": self.ui.title,
        "logo_url": self.ui.logo_url,
        "favicon_url": self.ui.favicon_url,
        "primary_color": self.ui.primary_color,
        "primary_color_dark": self.ui.primary_color_dark,
        "dark_mode_default": self.ui.dark_mode_default,
    }

get_session_config()

Get session configuration.

Source code in fastapi_admin_kit/admin/admin_config.py

def get_session_config(self) -> dict:
    """Get session configuration."""
    return {
        "session_ttl": self.auth.session_ttl,
        "session_cookie_name": self.auth.session_cookie_name,
        "session_secure": self.auth.session_secure,
    }

get_audit_config()

Get audit configuration.

Source code in fastapi_admin_kit/admin/admin_config.py

def get_audit_config(self) -> dict:
    """Get audit configuration."""
    return {
        "audit_retention_days": self.audit.audit_retention_days,
    }

get_storage_config()

Get storage configuration.

Source code in fastapi_admin_kit/admin/admin_config.py

def get_storage_config(self) -> dict:
    """Get storage configuration."""
    return {
        "uploads_url": self.storage.uploads_url,
    }

get_behavior_config()

Get behavior configuration.

Source code in fastapi_admin_kit/admin/admin_config.py

def get_behavior_config(self) -> dict:
    """Get behavior configuration."""
    return {
        "auto_discover": self.behavior.auto_discover,
        "dashboard_stats": self.behavior.dashboard_stats,
        "dashboard_charts": self.behavior.dashboard_charts,
    }

get_nav_config()

Get navigation configuration.

Source code in fastapi_admin_kit/admin/admin_config.py

def get_nav_config(self) -> dict:
    """Get navigation configuration."""
    return {
        "nav_groups": self.nav.nav_groups,
        "sidebar_builder": self.nav.sidebar_builder,
        "require_tags": self.nav.require_tags,
        "dashboard_permission": self.nav.dashboard_permission,
        "settings_permission": self.nav.settings_permission,
    }

AdminRegistry

fastapi_admin_kit.AdminRegistry

Singleton registry for admin models.

Uses dependency injection for ModelInspector and ModelValidator, making the registry testable and separable from inspection/validation concerns.

Source code in fastapi_admin_kit/registry/core.py

class AdminRegistry:
    """Singleton registry for admin models.

    Uses dependency injection for ModelInspector and ModelValidator,
    making the registry testable and separable from inspection/validation concerns.
    """

    _instance: AdminRegistry | None = None
    _models: dict[str, RegisteredModel] = {}

    def __new__(cls) -> AdminRegistry:
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._models = {}
        return cls._instance

    def __init__(self) -> None:
        """Initialize the registry with default inspector and validator."""
        if not hasattr(self, "_inspector"):
            self._inspector = ModelInspector()
        if not hasattr(self, "_validator"):
            self._validator = ModelValidator(self)

    @property
    def inspector(self) -> ModelInspector:
        """Get the model inspector."""
        return self._inspector

    @inspector.setter
    def inspector(self, value: ModelInspector) -> None:
        """Set the model inspector."""
        self._inspector = value

    @property
    def validator(self) -> ModelValidator:
        """Get the model validator."""
        return self._validator

    @validator.setter
    def validator(self, value: ModelValidator) -> None:
        """Set the model validator."""
        self._validator = value

    def register(
        self,
        model: type,
        admin_class: type[ModelAdmin] | None = None,
    ) -> RegisteredModel:
        """Register a model with the admin.

        Args:
            model: A SQLAlchemy declarative model class.
            admin_class: Optional ModelAdmin subclass for the model.

        Returns:
            The registered model configuration.

        Raises:
            ValueError: If the model is not a valid SQLAlchemy model.
        """
        from fastapi_admin_kit.views import ModelAdmin

        # Validate using the injected validator
        self._validator.validate_model_registration(model, admin_class)

        admin = admin_class() if admin_class else ModelAdmin()
        registered = build_registered_model(model, admin)

        self._models[registered.table_name] = registered
        return registered

    def get(self, table_name: str) -> RegisteredModel | None:
        """Get a registered model by table name.

        Args:
            table_name: The table name to look up.

        Returns:
            The registered model, or None if not found.
        """
        return self._models.get(table_name)

    def all(self) -> list[RegisteredModel]:
        """Get all registered models.

        Returns:
            A list of all registered models.
        """
        return list(self._models.values())

    def auto_discover(
        self, exclude_tables: set[str] | frozenset[str] | None = None
    ) -> list[RegisteredModel]:
        """Scan all subclasses of DeclarativeBase and register unregistered ones.

        Also discovers SQLModel subclasses if SQLModel is installed.

        Args:
            exclude_tables: Optional set/frozenset of ``__tablename__`` values to
                skip during discovery (e.g. built-in internal or gated tables).

        Returns:
            A list of newly registered models.
        """
        from sqlalchemy.orm import DeclarativeBase

        exclude_tables = exclude_tables or set()
        discovered: list[RegisteredModel] = []
        seen: set[type] = set()

        def _maybe_register(cls: type) -> None:
            if cls in seen:
                return
            seen.add(cls)
            table_name = getattr(cls, "__tablename__", None)
            if table_name is None or table_name in self._models:
                return
            if table_name in exclude_tables:
                return
            discovered.append(self.register(cls))

        # Discover SQLAlchemy DeclarativeBase subclasses
        for subclass in _all_declarative_subclasses(DeclarativeBase):
            if hasattr(subclass, "registry"):
                for mapper in subclass.registry.mappers:
                    _maybe_register(mapper.class_)

        # Discover SQLModel subclasses (if installed)
        try:
            from sqlmodel import SQLModel

            for subclass in _all_declarative_subclasses(SQLModel):
                if hasattr(subclass, "registry"):
                    for mapper in subclass.registry.mappers:
                        _maybe_register(mapper.class_)
        except ImportError:
            pass

        return discovered

    def clear(self) -> None:
        """Clear all registrations (useful for testing)."""
        self._models.clear()

inspector property writable

Get the model inspector.

validator property writable

Get the model validator.

__init__()

Initialize the registry with default inspector and validator.

Source code in fastapi_admin_kit/registry/core.py

def __init__(self) -> None:
    """Initialize the registry with default inspector and validator."""
    if not hasattr(self, "_inspector"):
        self._inspector = ModelInspector()
    if not hasattr(self, "_validator"):
        self._validator = ModelValidator(self)

register(model, admin_class=None)

Register a model with the admin.

Parameters:

Name	Type	Description	Default
model	type	A SQLAlchemy declarative model class.	required
admin_class	type[ModelAdmin] | None	Optional ModelAdmin subclass for the model.	None

Returns:

Type	Description
RegisteredModel	The registered model configuration.

Raises:

Type	Description
ValueError	If the model is not a valid SQLAlchemy model.

Source code in fastapi_admin_kit/registry/core.py

def register(
    self,
    model: type,
    admin_class: type[ModelAdmin] | None = None,
) -> RegisteredModel:
    """Register a model with the admin.

    Args:
        model: A SQLAlchemy declarative model class.
        admin_class: Optional ModelAdmin subclass for the model.

    Returns:
        The registered model configuration.

    Raises:
        ValueError: If the model is not a valid SQLAlchemy model.
    """
    from fastapi_admin_kit.views import ModelAdmin

    # Validate using the injected validator
    self._validator.validate_model_registration(model, admin_class)

    admin = admin_class() if admin_class else ModelAdmin()
    registered = build_registered_model(model, admin)

    self._models[registered.table_name] = registered
    return registered

get(table_name)

Get a registered model by table name.

Parameters:

Name	Type	Description	Default
table_name	str	The table name to look up.	required

Returns:

Type	Description
RegisteredModel | None	The registered model, or None if not found.

Source code in fastapi_admin_kit/registry/core.py

def get(self, table_name: str) -> RegisteredModel | None:
    """Get a registered model by table name.

    Args:
        table_name: The table name to look up.

    Returns:
        The registered model, or None if not found.
    """
    return self._models.get(table_name)

all()

Get all registered models.

Returns:

Type	Description
list[RegisteredModel]	A list of all registered models.

Source code in fastapi_admin_kit/registry/core.py

def all(self) -> list[RegisteredModel]:
    """Get all registered models.

    Returns:
        A list of all registered models.
    """
    return list(self._models.values())

auto_discover(exclude_tables=None)

Scan all subclasses of DeclarativeBase and register unregistered ones.

Also discovers SQLModel subclasses if SQLModel is installed.

Parameters:

Name	Type	Description	Default
exclude_tables	set[str] | frozenset[str] | None	Optional set/frozenset of __tablename__ values to skip during discovery (e.g. built-in internal or gated tables).	None

Returns:

Type	Description
list[RegisteredModel]	A list of newly registered models.

Source code in fastapi_admin_kit/registry/core.py

def auto_discover(
    self, exclude_tables: set[str] | frozenset[str] | None = None
) -> list[RegisteredModel]:
    """Scan all subclasses of DeclarativeBase and register unregistered ones.

    Also discovers SQLModel subclasses if SQLModel is installed.

    Args:
        exclude_tables: Optional set/frozenset of ``__tablename__`` values to
            skip during discovery (e.g. built-in internal or gated tables).

    Returns:
        A list of newly registered models.
    """
    from sqlalchemy.orm import DeclarativeBase

    exclude_tables = exclude_tables or set()
    discovered: list[RegisteredModel] = []
    seen: set[type] = set()

    def _maybe_register(cls: type) -> None:
        if cls in seen:
            return
        seen.add(cls)
        table_name = getattr(cls, "__tablename__", None)
        if table_name is None or table_name in self._models:
            return
        if table_name in exclude_tables:
            return
        discovered.append(self.register(cls))

    # Discover SQLAlchemy DeclarativeBase subclasses
    for subclass in _all_declarative_subclasses(DeclarativeBase):
        if hasattr(subclass, "registry"):
            for mapper in subclass.registry.mappers:
                _maybe_register(mapper.class_)

    # Discover SQLModel subclasses (if installed)
    try:
        from sqlmodel import SQLModel

        for subclass in _all_declarative_subclasses(SQLModel):
            if hasattr(subclass, "registry"):
                for mapper in subclass.registry.mappers:
                    _maybe_register(mapper.class_)
    except ImportError:
        pass

    return discovered

clear()

Clear all registrations (useful for testing).

Source code in fastapi_admin_kit/registry/core.py

def clear(self) -> None:
    """Clear all registrations (useful for testing)."""
    self._models.clear()

RegisteredModel

fastapi_admin_kit.RegisteredModel dataclass

Central dataclass holding a registered model and its admin config.

Source code in fastapi_admin_kit/registry/core.py

@dataclass
class RegisteredModel:
    """Central dataclass holding a registered model and its admin config."""

    model: type
    admin: ModelAdmin
    table_name: str
    verbose_name: str
    verbose_name_plural: str
    columns: list = field(default_factory=list)
    relationships: list = field(default_factory=list)
    pk_field: str | tuple[str, ...] | None = "id"
    _schemas: dict | None = field(default=None, repr=False)

    def __post_init__(self) -> None:
        # Find primary key
        for col in self.columns:
            if col.primary_key:
                self.pk_field = col.name
                break
        # Ensure admin has reference to model for form field deduplication
        if not hasattr(self.admin, "model") or self.admin.model is None:
            self.admin.model = self.model

    @property
    def form_fields(self) -> list[Any]:
        return self.admin.get_form_fields(
            columns=self.columns,
            relationships=self.relationships,
        )

    @property
    def list_fields(self) -> list[str]:
        if self.admin.list_display:
            # Registration-time validation (validate_admin_fields) already
            # verified every name exists on the model, so return as-is.
            return list(self.admin.list_display)
        return [c.name for c in self.columns if not c.primary_key]

    def get_widget(self, field_name: str, resolver: WidgetResolver | None = None) -> Widget:
        from fastapi_admin_kit.inspection import auto_label
        from fastapi_admin_kit.widgets.registry import widget_registry
        from fastapi_admin_kit.widgets.relation import (
            MultiRelationWidget,
            RelationPickerWidget,
        )
        from fastapi_admin_kit.widgets.resolver import WidgetResolver

        if resolver is None:
            resolver = WidgetResolver(widget_registry)

        overrides = getattr(self.admin, "formfield_overrides", {})
        if field_name in overrides:
            return overrides[field_name]

        col = next((c for c in self.columns if c.name == field_name), None)
        rel = next((r for r in self.relationships if r.name == field_name), None)
        if col is not None:
            widget = resolver.resolve(col)
            if (
                isinstance(widget, RelationPickerWidget)
                and not widget.related_table
                and col.foreign_keys
            ):
                fk = col.foreign_keys[0]
                widget.related_table = fk.column.table.name
            return widget
        if rel is not None:
            related_verbose = auto_label(rel.target_model.__tablename__)
            if rel.direction == "MANYTOONE" or not rel.uselist:
                return RelationPickerWidget(
                    related_table=rel.target_model.__tablename__,
                    related_verbose=related_verbose,
                )
            table_name = rel.target_model.__tablename__
            search_path = table_name
            if search_path.startswith("admin_"):
                search_path = search_path[len("admin_") :]
            return MultiRelationWidget(
                related_table=table_name,
                related_verbose=related_verbose,
                search_url=f"/admin/{search_path}/search",
            )
        return resolver.resolve(  # type: ignore[arg-type]
            type("_Col", (), {"type": type(None), "name": field_name})()
        )

ModelAdmin

fastapi_admin_kit.ModelAdmin

Base class for model admin configuration.

Subclass this to customise how a model is displayed, filtered, and edited. All fields are optional — unset fields fall back to auto-detected values.

Source code in fastapi_admin_kit/modeladmin.py

class ModelAdmin:
    """Base class for model admin configuration.

    Subclass this to customise how a model is displayed, filtered, and edited.
    All fields are optional — unset fields fall back to auto-detected values.
    """

    # List view config
    list_display: list[str] | None = None
    list_filter: list[str | Any] | None = None
    search_fields: list[str] | None = None
    ordering: list[str] | None = None
    per_page: int = 20
    pagination: Any = None  # OffsetPagination | CursorPagination | DynamicPagination
    list_filter_options: dict[str, dict[str, Any]] = {}
    list_filter_horizontal: bool = False

    @staticmethod
    def get_ordering(
        request_params: dict,
        admin_ordering: list[str] | None,
        model: Any = None,
    ) -> list[str]:
        """Resolve ordering configuration based on request params and admin settings.

        Priority (highest to lowest):
        1. Query parameter (e.g., from URL ?ordering=name)
        2. Admin class ordering (from ModelAdmin.ordering)
        3. Empty list (no default ordering applied)

        This prevents unwanted default sorting when ordering is not explicitly
        configured. When *model* is given, the result is restricted to real
        model columns/relationships via :func:`sanitize_ordering` — the query
        parameter is client-controlled and must not reach arbitrary
        ``getattr(model, ...)`` calls.
        """
        query_ordering = request_params.get("ordering", "")
        if query_ordering:
            order = [query_ordering]
        elif admin_ordering:
            order = admin_ordering
        else:
            return []
        if model is not None:
            order = sanitize_ordering(model, order)
        return order

    # Inline editing config
    inline_edit: bool = False
    inline_edit_fields: list[str] | None = None
    inline_exclude_fields: list[str] | None = None

    # Actions config
    actions_list: list[str] = []
    actions_row: list[str] = []
    actions_detail: list[str] = []
    actions_submit_line: list[str] = []
    actions_list_hide_default: bool = False

    # Export/Import config
    export_formats: dict[str, Any] | None = None
    import_formats: dict[str, Any] | None = None

    # Form config
    fields: list[str] | None = None
    exclude: list[str] | None = None
    readonly_fields: list[str] | None = None
    formfield_overrides: dict[str, Any] = {}
    extra_fields: list[ExtraField] = []
    fieldsets: list[Any] = []  # FieldsetSpec accepted but not strictly enforced here
    field_placeholders: dict[str, str] = {}  # {field_name: placeholder_text}

    # Inline admin config
    inlines: list[Any] = []  # list of InlineModelAdmin subclasses

    # Optional Redis rate limiting for this model's list endpoint.
    # ``None`` falls back to the global default
    # (``CacheConfig.rate_limit`` / ``rate_window``).
    rate_limit: int | None = None
    rate_window: int | None = None

    # Conditional fields
    conditional_fields: dict[str, dict[str, Any]] = {}

    # Form UX config
    warn_unsaved_form: bool = True
    compressed_fields: bool = True
    change_form_show_cancel_button: bool = True

    # Labels and display
    verbose_name: str | None = None
    verbose_name_plural: str | None = None
    icon: str | None = None
    tag: str | None = None
    tags: list[str] | None = None
    nav_order: int = 999
    nav_children: list[NavItemConfig] | None = None

    # Template customization (None = auto-discovery → built-in default)
    list_template: str | None = None
    create_template: str | None = None
    edit_template: str | None = None
    detail_template: str | None = None
    inline_edit_template: str | None = None

    # Route generation
    skip_auto_routes: bool = False
    # Endpoint export control: None (default) → both HTML + JSON API routers
    # are auto-built; "html" → only the admin HTML router is auto-built;
    # "api" → only the JSON API router is auto-built.
    export_endpoint: str | None = None

    # Custom display functions (dict-based fallback)
    display_functions: dict[str, Any] | None = None

    # Decorator for custom column display
    column = staticmethod(column)

    # Decorator for custom FastAPI endpoints
    endpoint = staticmethod(endpoint)

    # ── Standalone router export (no admin.register required) ───────

    def export_api_route(self, model: Any, prefix: str = "") -> Any:
        """Build a standalone JSON API router for ``model``.

        Unlike ``admin.register()`` this does not write to the singleton
        registry — the returned router can be mounted directly::

            app.include_router(ProductAdmin().export_api_route(Product))

        Args:
            model: A SQLAlchemy declarative model class.
            prefix: Optional extra path prefix prepended to the router.

        Returns:
            An ``APIRouter`` exposing the model's JSON CRUD endpoints.
        """
        from fastapi import APIRouter

        from fastapi_admin_kit.api.crud import build_api_router_for_model
        from fastapi_admin_kit.registry import build_registered_model

        registered = build_registered_model(model, self)
        router = build_api_router_for_model(registered)
        if prefix:
            wrapper = APIRouter(prefix=prefix)
            wrapper.include_router(router)
            return wrapper
        return router

    def export_admin_route(self, model: Any, prefix: str = "") -> Any:
        """Build a standalone HTML admin router for ``model``.

        Unlike ``admin.register()`` this does not write to the singleton
        registry — the returned router can be mounted directly::

            app.include_router(ProductAdmin().export_admin_route(Product), prefix="/admin")

        Args:
            model: A SQLAlchemy declarative model class.
            prefix: Optional extra path prefix prepended to the router.

        Returns:
            An ``APIRouter`` exposing the model's HTML admin views.
        """
        from fastapi import APIRouter

        from fastapi_admin_kit.registry import build_registered_model
        from fastapi_admin_kit.router import build_model_router

        registered = build_registered_model(model, self)
        # Explicit call — build regardless of the model's export_endpoint.
        router = build_model_router(registered, force=True)
        if prefix:
            wrapper = APIRouter(prefix=prefix)
            wrapper.include_router(router)
            return wrapper
        return router

    # Badge hook — return str e.g. "12" or None
    def get_nav_badge(self, request: Any = None) -> str | None:
        return None

    # Object display
    def __str__(self, obj: Any) -> str:
        """How to display an object in dropdowns/links."""
        return str(
            getattr(obj, "name", None)
            or getattr(obj, "title", None)
            or f"#{getattr(obj, 'id', '?')}"
        )

    def get_model(self) -> Any:
        return self.model

    # ── Query hooks ──────────────────────────────────────────────────

    def get_queryset(self, session: Session, request: Any = None) -> Any:
        """Override to filter records globally (e.g. soft-delete filter).

        Returns a SQLAlchemy ``select`` statement for the model.
        Override and call ``super().get_queryset(session, request)`` to
        chain additional ``.where()`` conditions.
        """
        from sqlalchemy import select

        return select(self.model)  # type: ignore[attr-defined]

    def get_object(self, session: Session, id: Any) -> Any:
        """Override for custom PK lookup."""
        return session.get(self.model, id)  # type: ignore[attr-defined]

    # ── Lifecycle hooks (stubs) ─────────────────────────────────────

    def prepare_create_data(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
        """Strip extra fields and return only model-column data for INSERT."""
        extra_names = {f.name for f in self.extra_fields}
        return {k: v for k, v in data.items() if k not in extra_names}

    def prepare_update_data(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
        """Strip extra fields and return only model-column data for UPDATE."""
        extra_names = {f.name for f in self.extra_fields}
        return {k: v for k, v in data.items() if k not in extra_names}

    def on_create(self, obj: Any, request: Any = None) -> None:
        """Called before INSERT. Mutate *obj* as needed."""

    def after_create(self, obj: Any, request: Any = None) -> None:
        """Called after INSERT commit."""

    def on_update(self, obj: Any, data: dict[str, Any], request: Any = None) -> None:
        """Called before UPDATE. *data* contains the incoming form values."""

    def after_update(self, obj: Any, request: Any = None) -> None:
        """Called after UPDATE commit."""

    def on_delete(self, obj: Any, request: Any = None) -> None:
        """Called before DELETE."""

    def after_delete(self, obj: Any, request: Any = None) -> None:
        """Called after DELETE commit."""

    # ── Validation hooks (stubs) ────────────────────────────────────

    def validate_create(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
        """Validate and/or transform form data before create.

        Return the (possibly modified) data dict.  Raise ``ValueError``
        with a user-facing message to abort the operation.
        """
        return data

    def validate_update(
        self, obj: Any, data: dict[str, Any], request: Any = None
    ) -> dict[str, Any]:
        """Validate and/or transform form data before update.

        Return the (possibly modified) data dict.  Raise ``ValueError``
        with a user-facing message to abort the operation.
        """
        return data

    # ── Form data processing hooks ──────────────────────────────────

    def process_form_data(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
        """Process form data before save. Override for custom processing.

        Called after form parsing and validation. Use this to transform
        form data, extract special fields, or prepare data for saving.
        """
        return data

    def save_model(
        self,
        obj: Any,
        data: dict[str, Any],
        request: Any = None,
        is_create: bool = False,
    ) -> None:
        """Custom save logic. Override for full control over save flow.

        When overridden, this method is called instead of the default
        save logic. The object is already added to the session but not
        committed yet. Call session.commit() yourself if needed.
        """
        pass

    # ── Form context hooks ──────────────────────────────────────────

    async def get_form_context(
        self, context: dict[str, Any], obj: Any = None, request: Any = None
    ) -> dict[str, Any]:
        """Customize form template context. Return modified context.

        Called when building the form context for create/edit views.
        Override to add custom template variables.
        """
        return context

    # ── Dynamic field hooks ─────────────────────────────────────────

    def get_readonly_fields(self, obj: Any = None, request: Any = None) -> list[str]:
        """Return list of readonly field names. Override for dynamic readonly.

        Called during form field generation. Fields returned here are
        in addition to the static readonly_fields list.
        """
        return self.readonly_fields or []

    def get_hidden_fields(self, obj: Any = None, request: Any = None) -> list[str]:
        """Return list of hidden field names. Override for dynamic hidden.

        Called during form field generation. Hidden fields are excluded
        from the form entirely.
        """
        return []

    # ── Form field helper ───────────────────────────────────────────

    def get_form_fields(
        self,
        obj: Any = None,
        request: Any = None,
        columns: list[Any] | None = None,
        relationships: list[Any] | None = None,
    ) -> list[FieldMeta]:
        """Return ordered list of FieldMeta objects for the create/edit form."""
        from fastapi_admin_kit.inspection import auto_label, is_required

        columns = columns or []
        relationships = relationships or []
        form_fields: list[FieldMeta] = []

        raw = []
        if self.fields is not None:
            names = set(self.fields)
            raw = [c for c in columns if c.name in names and not c.primary_key] + [
                r
                for r in relationships
                if r.name in names and r.direction in ("MANYTOONE", "MANYTOMANY")
            ]
        else:
            raw = [c for c in columns if not c.primary_key] + [
                r for r in relationships if r.direction in ("MANYTOONE", "MANYTOMANY")
            ]
            if self.exclude:
                raw = [x for x in raw if x.name not in self.exclude]

            # Exclude FK columns that have a corresponding relationship
            # to avoid duplicate fields (e.g., user_id + user both showing)
            from sqlalchemy import inspect as sa_inspect

            rel_fk_cols: set[str] = set()
            try:
                mapper = sa_inspect(self.model)
                for r in relationships:
                    if r.direction in ("MANYTOONE", "MANYTOMANY"):
                        rel_prop = mapper.relationships.get(r.name)
                        if rel_prop is not None:
                            for local_col in rel_prop.local_columns:
                                rel_fk_cols.add(local_col.key)
            except Exception:
                pass
            raw = [x for x in raw if not (hasattr(x, "foreign_keys") and x.name in rel_fk_cols)]

        dynamic_readonly = set(self.get_readonly_fields())
        dynamic_hidden = set(self.get_hidden_fields())

        for item in raw:
            name = item.name
            if name in dynamic_hidden:
                continue
            readonly = name in (self.readonly_fields or []) or name in dynamic_readonly
            required = is_required(item) if hasattr(item, "nullable") else False
            label = auto_label(name)
            placeholder = self.field_placeholders.get(name, f"Enter {label.lower()}...")
            form_fields.append(
                FieldMeta(
                    name=name,
                    label=label,
                    required=required,
                    readonly=readonly,
                    placeholder=placeholder,
                )
            )

        for extra in self.extra_fields:
            form_fields.append(
                FieldMeta(
                    name=extra.name,
                    label=extra.label or auto_label(extra.name),
                    required=extra.required,
                    readonly=False,
                    extra={
                        "extra_field": True,
                        "widget": extra.widget,
                        "required_on_create": extra.required_on_create,
                    },
                )
            )

        # Respect fieldsets ordering if defined
        if self.fieldsets:
            ordered: list[FieldMeta] = []
            seen: set[str] = set()
            for fs in self.fieldsets:
                for fname in fs.fields:
                    for fm in form_fields:
                        if fm.name == fname and fname not in seen:
                            ordered.append(fm)
                            seen.add(fname)
                            break
            for fm in form_fields:
                if fm.name not in seen:
                    ordered.append(fm)
            return ordered

        return form_fields

    # ── Action helpers ─────────────────────────────────────────────

    def get_actions_for_location(self, location: str) -> list[Any]:
        """Get resolved Action instances for a given location (list/row/detail/submit_line)."""
        from fastapi_admin_kit.actions.base import Action

        action_names = getattr(self, f"actions_{location}", [])
        resolved = []
        for name in action_names:
            action_fn = getattr(self, name, None)
            if not action_fn:
                continue
            if hasattr(action_fn, "_admin_action"):
                resolved.append(action_fn._admin_action())
            elif callable(action_fn):
                label = name.replace("_", " ").title()
                _fn = action_fn
                _admin = self

                class _FallbackAction(Action):
                    def __init__(self):
                        super().__init__(name=name, label=label)

                    async def execute(self, objects, request):
                        import inspect

                        if inspect.iscoroutinefunction(_fn):
                            await _fn(_admin, objects, request)
                        else:
                            _fn(_admin, objects, request)

                resolved.append(_FallbackAction())
        return resolved

    def get_list_actions(self) -> list[Any]:
        return self.get_actions_for_location("list")

    def get_row_actions(self) -> list[Any]:
        return self.get_actions_for_location("row")

    def get_detail_actions(self) -> list[Any]:
        return self.get_actions_for_location("detail")

    def get_submit_line_actions(self) -> list[Any]:
        return self.get_actions_for_location("submit_line")

    # ── Inline edit helpers ────────────────────────────────────────

    def get_inline_edit_fields(
        self,
        obj: Any = None,
        request: Any = None,
        columns: list[Any] | None = None,
        relationships: list[Any] | None = None,
    ) -> list[FieldMeta]:
        """Return FieldMeta list for the inline edit form."""
        all_fields = self.get_form_fields(
            obj=obj,
            request=request,
            columns=columns,
            relationships=relationships,
        )
        if self.inline_edit_fields is not None:
            allowed = set(self.inline_edit_fields)
            return [f for f in all_fields if f.name in allowed]
        if self.inline_exclude_fields is not None:
            excluded = set(self.inline_exclude_fields)
            return [f for f in all_fields if f.name not in excluded]
        return all_fields

    # ── Notification recipients hook ──────────────────────────────────

    def get_notification_recipients(
        self, event: str, request: Any = None, obj: Any = None
    ) -> list[dict[str, Any]] | None:
        """Get notification recipients for an admin change event.

        The default implementation returns ``None``, which signals the dispatcher
        to use the built-in default behaviour: superusers always receive notifications,
        regular admin users receive them only if they have enabled
        ``NotificationPreference`` rows.

        Subclasses may override this method to customise recipient selection,
        channels, or to bypass preference lookups entirely.

        Args:
            event: One of ``"create"``, ``"update"``, or ``"delete"``.
            request: Current request (optional, for accessing auth context).
            obj: The affected object instance (optional).

        Returns:
            ``list[{"id", "email", "phone", "channels"}]`` of recipient dicts,
            or ``[]`` to disable notifications for this model entirely,
            or ``None`` to use the default behaviour (superusers always,
            regular admins only with enabled preferences).
        """
        return None

    # ── Permission helpers ───────────────────────────────────────────

    def has_view_permission(self, request: Any = None) -> bool:
        return True

    def has_create_permission(self, request: Any = None) -> bool:
        return True

    def has_edit_permission(self, request: Any = None) -> bool:
        return True

    def has_delete_permission(self, request: Any = None) -> bool:
        return True

    def has_export_permission(self, request: Any = None) -> bool:
        return True

    def has_import_permission(self, request: Any = None) -> bool:
        return True

    # ── Export/Import helpers ───────────────────────────────────────

    def get_export_formats(self) -> dict[str, Any]:
        """Get available export formats for this model.

        Returns configured formats or defaults to CSV + Excel.
        """
        if self.export_formats:
            return self.export_formats
        from fastapi_admin_kit.export_import.registry import (
            get_available_export_formats,
        )

        formats = {}
        for name in get_available_export_formats():
            formats[name] = None
        return formats or {"csv": None}

    def get_import_formats(self) -> dict[str, Any]:
        """Get available import formats for this model.

        Returns configured formats or defaults to CSV + Excel.
        """
        if self.import_formats:
            return self.import_formats
        from fastapi_admin_kit.export_import.registry import (
            get_available_import_formats,
        )

        formats = {}
        for name in get_available_import_formats():
            formats[name] = None
        return formats or {"csv": None}

    def get_export_class(self, format_name: str) -> type | None:
        """Get the export class for a given format.

        Args:
            format_name: Format name (e.g., "csv", "excel")

        Returns:
            The export class or None if not available
        """
        formats = self.get_export_formats()
        if format_name not in formats:
            return None

        format_class = formats[format_name]
        if format_class is None:
            # Use default from registry
            from fastapi_admin_kit.export_import.registry import (
                get_export_class,
            )

            return get_export_class(format_name)
        return format_class

    def get_import_class(self, format_name: str) -> type | None:
        """Get the import class for a given format.

        Args:
            format_name: Format name (e.g., "csv", "excel")

        Returns:
            The import class or None if not available
        """
        formats = self.get_import_formats()
        if format_name not in formats:
            return None

        format_class = formats[format_name]
        if format_class is None:
            # Use default from registry
            from fastapi_admin_kit.export_import.registry import (
                get_import_class,
            )

            return get_import_class(format_name)
        return format_class

get_ordering(request_params, admin_ordering, model=None) staticmethod

Resolve ordering configuration based on request params and admin settings.

Priority (highest to lowest): 1. Query parameter (e.g., from URL ?ordering=name) 2. Admin class ordering (from ModelAdmin.ordering) 3. Empty list (no default ordering applied)

This prevents unwanted default sorting when ordering is not explicitly configured. When model is given, the result is restricted to real model columns/relationships via :func:sanitize_ordering — the query parameter is client-controlled and must not reach arbitrary getattr(model, ...) calls.

Source code in fastapi_admin_kit/modeladmin.py

@staticmethod
def get_ordering(
    request_params: dict,
    admin_ordering: list[str] | None,
    model: Any = None,
) -> list[str]:
    """Resolve ordering configuration based on request params and admin settings.

    Priority (highest to lowest):
    1. Query parameter (e.g., from URL ?ordering=name)
    2. Admin class ordering (from ModelAdmin.ordering)
    3. Empty list (no default ordering applied)

    This prevents unwanted default sorting when ordering is not explicitly
    configured. When *model* is given, the result is restricted to real
    model columns/relationships via :func:`sanitize_ordering` — the query
    parameter is client-controlled and must not reach arbitrary
    ``getattr(model, ...)`` calls.
    """
    query_ordering = request_params.get("ordering", "")
    if query_ordering:
        order = [query_ordering]
    elif admin_ordering:
        order = admin_ordering
    else:
        return []
    if model is not None:
        order = sanitize_ordering(model, order)
    return order

export_api_route(model, prefix='')

Build a standalone JSON API router for model.

Unlike admin.register() this does not write to the singleton registry — the returned router can be mounted directly::

app.include_router(ProductAdmin().export_api_route(Product))

Parameters:

Name	Type	Description	Default
model	Any	A SQLAlchemy declarative model class.	required
prefix	str	Optional extra path prefix prepended to the router.	''

Returns:

Type	Description
Any	An APIRouter exposing the model's JSON CRUD endpoints.

Source code in fastapi_admin_kit/modeladmin.py

def export_api_route(self, model: Any, prefix: str = "") -> Any:
    """Build a standalone JSON API router for ``model``.

    Unlike ``admin.register()`` this does not write to the singleton
    registry — the returned router can be mounted directly::

        app.include_router(ProductAdmin().export_api_route(Product))

    Args:
        model: A SQLAlchemy declarative model class.
        prefix: Optional extra path prefix prepended to the router.

    Returns:
        An ``APIRouter`` exposing the model's JSON CRUD endpoints.
    """
    from fastapi import APIRouter

    from fastapi_admin_kit.api.crud import build_api_router_for_model
    from fastapi_admin_kit.registry import build_registered_model

    registered = build_registered_model(model, self)
    router = build_api_router_for_model(registered)
    if prefix:
        wrapper = APIRouter(prefix=prefix)
        wrapper.include_router(router)
        return wrapper
    return router

export_admin_route(model, prefix='')

Build a standalone HTML admin router for model.

Unlike admin.register() this does not write to the singleton registry — the returned router can be mounted directly::

app.include_router(ProductAdmin().export_admin_route(Product), prefix="/admin")

Parameters:

Name	Type	Description	Default
model	Any	A SQLAlchemy declarative model class.	required
prefix	str	Optional extra path prefix prepended to the router.	''

Returns:

Type	Description
Any	An APIRouter exposing the model's HTML admin views.

Source code in fastapi_admin_kit/modeladmin.py

def export_admin_route(self, model: Any, prefix: str = "") -> Any:
    """Build a standalone HTML admin router for ``model``.

    Unlike ``admin.register()`` this does not write to the singleton
    registry — the returned router can be mounted directly::

        app.include_router(ProductAdmin().export_admin_route(Product), prefix="/admin")

    Args:
        model: A SQLAlchemy declarative model class.
        prefix: Optional extra path prefix prepended to the router.

    Returns:
        An ``APIRouter`` exposing the model's HTML admin views.
    """
    from fastapi import APIRouter

    from fastapi_admin_kit.registry import build_registered_model
    from fastapi_admin_kit.router import build_model_router

    registered = build_registered_model(model, self)
    # Explicit call — build regardless of the model's export_endpoint.
    router = build_model_router(registered, force=True)
    if prefix:
        wrapper = APIRouter(prefix=prefix)
        wrapper.include_router(router)
        return wrapper
    return router

__str__(obj)

How to display an object in dropdowns/links.

Source code in fastapi_admin_kit/modeladmin.py

def __str__(self, obj: Any) -> str:
    """How to display an object in dropdowns/links."""
    return str(
        getattr(obj, "name", None)
        or getattr(obj, "title", None)
        or f"#{getattr(obj, 'id', '?')}"
    )

get_queryset(session, request=None)

Override to filter records globally (e.g. soft-delete filter).

Returns a SQLAlchemy select statement for the model. Override and call super().get_queryset(session, request) to chain additional .where() conditions.

Source code in fastapi_admin_kit/modeladmin.py

def get_queryset(self, session: Session, request: Any = None) -> Any:
    """Override to filter records globally (e.g. soft-delete filter).

    Returns a SQLAlchemy ``select`` statement for the model.
    Override and call ``super().get_queryset(session, request)`` to
    chain additional ``.where()`` conditions.
    """
    from sqlalchemy import select

    return select(self.model)  # type: ignore[attr-defined]

get_object(session, id)

Override for custom PK lookup.

Source code in fastapi_admin_kit/modeladmin.py

def get_object(self, session: Session, id: Any) -> Any:
    """Override for custom PK lookup."""
    return session.get(self.model, id)  # type: ignore[attr-defined]

prepare_create_data(data, request=None)

Strip extra fields and return only model-column data for INSERT.

Source code in fastapi_admin_kit/modeladmin.py

def prepare_create_data(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
    """Strip extra fields and return only model-column data for INSERT."""
    extra_names = {f.name for f in self.extra_fields}
    return {k: v for k, v in data.items() if k not in extra_names}

prepare_update_data(data, request=None)

Strip extra fields and return only model-column data for UPDATE.

Source code in fastapi_admin_kit/modeladmin.py

def prepare_update_data(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
    """Strip extra fields and return only model-column data for UPDATE."""
    extra_names = {f.name for f in self.extra_fields}
    return {k: v for k, v in data.items() if k not in extra_names}

on_create(obj, request=None)

Called before INSERT. Mutate obj as needed.

Source code in fastapi_admin_kit/modeladmin.py

def on_create(self, obj: Any, request: Any = None) -> None:
    """Called before INSERT. Mutate *obj* as needed."""

after_create(obj, request=None)

Called after INSERT commit.

Source code in fastapi_admin_kit/modeladmin.py

def after_create(self, obj: Any, request: Any = None) -> None:
    """Called after INSERT commit."""

on_update(obj, data, request=None)

Called before UPDATE. data contains the incoming form values.

Source code in fastapi_admin_kit/modeladmin.py

def on_update(self, obj: Any, data: dict[str, Any], request: Any = None) -> None:
    """Called before UPDATE. *data* contains the incoming form values."""

after_update(obj, request=None)

Called after UPDATE commit.

Source code in fastapi_admin_kit/modeladmin.py

def after_update(self, obj: Any, request: Any = None) -> None:
    """Called after UPDATE commit."""

on_delete(obj, request=None)

Called before DELETE.

Source code in fastapi_admin_kit/modeladmin.py

def on_delete(self, obj: Any, request: Any = None) -> None:
    """Called before DELETE."""

after_delete(obj, request=None)

Called after DELETE commit.

Source code in fastapi_admin_kit/modeladmin.py

def after_delete(self, obj: Any, request: Any = None) -> None:
    """Called after DELETE commit."""

validate_create(data, request=None)

Validate and/or transform form data before create.

Return the (possibly modified) data dict. Raise ValueError with a user-facing message to abort the operation.

Source code in fastapi_admin_kit/modeladmin.py

def validate_create(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
    """Validate and/or transform form data before create.

    Return the (possibly modified) data dict.  Raise ``ValueError``
    with a user-facing message to abort the operation.
    """
    return data

validate_update(obj, data, request=None)

Validate and/or transform form data before update.

Return the (possibly modified) data dict. Raise ValueError with a user-facing message to abort the operation.

Source code in fastapi_admin_kit/modeladmin.py

def validate_update(
    self, obj: Any, data: dict[str, Any], request: Any = None
) -> dict[str, Any]:
    """Validate and/or transform form data before update.

    Return the (possibly modified) data dict.  Raise ``ValueError``
    with a user-facing message to abort the operation.
    """
    return data

process_form_data(data, request=None)

Process form data before save. Override for custom processing.

Called after form parsing and validation. Use this to transform form data, extract special fields, or prepare data for saving.

Source code in fastapi_admin_kit/modeladmin.py

def process_form_data(self, data: dict[str, Any], request: Any = None) -> dict[str, Any]:
    """Process form data before save. Override for custom processing.

    Called after form parsing and validation. Use this to transform
    form data, extract special fields, or prepare data for saving.
    """
    return data

save_model(obj, data, request=None, is_create=False)

Custom save logic. Override for full control over save flow.

When overridden, this method is called instead of the default save logic. The object is already added to the session but not committed yet. Call session.commit() yourself if needed.

Source code in fastapi_admin_kit/modeladmin.py

def save_model(
    self,
    obj: Any,
    data: dict[str, Any],
    request: Any = None,
    is_create: bool = False,
) -> None:
    """Custom save logic. Override for full control over save flow.

    When overridden, this method is called instead of the default
    save logic. The object is already added to the session but not
    committed yet. Call session.commit() yourself if needed.
    """
    pass

get_form_context(context, obj=None, request=None) async

Customize form template context. Return modified context.

Called when building the form context for create/edit views. Override to add custom template variables.

Source code in fastapi_admin_kit/modeladmin.py

async def get_form_context(
    self, context: dict[str, Any], obj: Any = None, request: Any = None
) -> dict[str, Any]:
    """Customize form template context. Return modified context.

    Called when building the form context for create/edit views.
    Override to add custom template variables.
    """
    return context

get_readonly_fields(obj=None, request=None)

Return list of readonly field names. Override for dynamic readonly.

Called during form field generation. Fields returned here are in addition to the static readonly_fields list.

Source code in fastapi_admin_kit/modeladmin.py

def get_readonly_fields(self, obj: Any = None, request: Any = None) -> list[str]:
    """Return list of readonly field names. Override for dynamic readonly.

    Called during form field generation. Fields returned here are
    in addition to the static readonly_fields list.
    """
    return self.readonly_fields or []

get_hidden_fields(obj=None, request=None)

Return list of hidden field names. Override for dynamic hidden.

Called during form field generation. Hidden fields are excluded from the form entirely.

Source code in fastapi_admin_kit/modeladmin.py

def get_hidden_fields(self, obj: Any = None, request: Any = None) -> list[str]:
    """Return list of hidden field names. Override for dynamic hidden.

    Called during form field generation. Hidden fields are excluded
    from the form entirely.
    """
    return []

get_form_fields(obj=None, request=None, columns=None, relationships=None)

Return ordered list of FieldMeta objects for the create/edit form.

Source code in fastapi_admin_kit/modeladmin.py

def get_form_fields(
    self,
    obj: Any = None,
    request: Any = None,
    columns: list[Any] | None = None,
    relationships: list[Any] | None = None,
) -> list[FieldMeta]:
    """Return ordered list of FieldMeta objects for the create/edit form."""
    from fastapi_admin_kit.inspection import auto_label, is_required

    columns = columns or []
    relationships = relationships or []
    form_fields: list[FieldMeta] = []

    raw = []
    if self.fields is not None:
        names = set(self.fields)
        raw = [c for c in columns if c.name in names and not c.primary_key] + [
            r
            for r in relationships
            if r.name in names and r.direction in ("MANYTOONE", "MANYTOMANY")
        ]
    else:
        raw = [c for c in columns if not c.primary_key] + [
            r for r in relationships if r.direction in ("MANYTOONE", "MANYTOMANY")
        ]
        if self.exclude:
            raw = [x for x in raw if x.name not in self.exclude]

        # Exclude FK columns that have a corresponding relationship
        # to avoid duplicate fields (e.g., user_id + user both showing)
        from sqlalchemy import inspect as sa_inspect

        rel_fk_cols: set[str] = set()
        try:
            mapper = sa_inspect(self.model)
            for r in relationships:
                if r.direction in ("MANYTOONE", "MANYTOMANY"):
                    rel_prop = mapper.relationships.get(r.name)
                    if rel_prop is not None:
                        for local_col in rel_prop.local_columns:
                            rel_fk_cols.add(local_col.key)
        except Exception:
            pass
        raw = [x for x in raw if not (hasattr(x, "foreign_keys") and x.name in rel_fk_cols)]

    dynamic_readonly = set(self.get_readonly_fields())
    dynamic_hidden = set(self.get_hidden_fields())

    for item in raw:
        name = item.name
        if name in dynamic_hidden:
            continue
        readonly = name in (self.readonly_fields or []) or name in dynamic_readonly
        required = is_required(item) if hasattr(item, "nullable") else False
        label = auto_label(name)
        placeholder = self.field_placeholders.get(name, f"Enter {label.lower()}...")
        form_fields.append(
            FieldMeta(
                name=name,
                label=label,
                required=required,
                readonly=readonly,
                placeholder=placeholder,
            )
        )

    for extra in self.extra_fields:
        form_fields.append(
            FieldMeta(
                name=extra.name,
                label=extra.label or auto_label(extra.name),
                required=extra.required,
                readonly=False,
                extra={
                    "extra_field": True,
                    "widget": extra.widget,
                    "required_on_create": extra.required_on_create,
                },
            )
        )

    # Respect fieldsets ordering if defined
    if self.fieldsets:
        ordered: list[FieldMeta] = []
        seen: set[str] = set()
        for fs in self.fieldsets:
            for fname in fs.fields:
                for fm in form_fields:
                    if fm.name == fname and fname not in seen:
                        ordered.append(fm)
                        seen.add(fname)
                        break
        for fm in form_fields:
            if fm.name not in seen:
                ordered.append(fm)
        return ordered

    return form_fields

get_actions_for_location(location)

Get resolved Action instances for a given location (list/row/detail/submit_line).

Source code in fastapi_admin_kit/modeladmin.py

def get_actions_for_location(self, location: str) -> list[Any]:
    """Get resolved Action instances for a given location (list/row/detail/submit_line)."""
    from fastapi_admin_kit.actions.base import Action

    action_names = getattr(self, f"actions_{location}", [])
    resolved = []
    for name in action_names:
        action_fn = getattr(self, name, None)
        if not action_fn:
            continue
        if hasattr(action_fn, "_admin_action"):
            resolved.append(action_fn._admin_action())
        elif callable(action_fn):
            label = name.replace("_", " ").title()
            _fn = action_fn
            _admin = self

            class _FallbackAction(Action):
                def __init__(self):
                    super().__init__(name=name, label=label)

                async def execute(self, objects, request):
                    import inspect

                    if inspect.iscoroutinefunction(_fn):
                        await _fn(_admin, objects, request)
                    else:
                        _fn(_admin, objects, request)

            resolved.append(_FallbackAction())
    return resolved

get_inline_edit_fields(obj=None, request=None, columns=None, relationships=None)

Return FieldMeta list for the inline edit form.

Source code in fastapi_admin_kit/modeladmin.py

def get_inline_edit_fields(
    self,
    obj: Any = None,
    request: Any = None,
    columns: list[Any] | None = None,
    relationships: list[Any] | None = None,
) -> list[FieldMeta]:
    """Return FieldMeta list for the inline edit form."""
    all_fields = self.get_form_fields(
        obj=obj,
        request=request,
        columns=columns,
        relationships=relationships,
    )
    if self.inline_edit_fields is not None:
        allowed = set(self.inline_edit_fields)
        return [f for f in all_fields if f.name in allowed]
    if self.inline_exclude_fields is not None:
        excluded = set(self.inline_exclude_fields)
        return [f for f in all_fields if f.name not in excluded]
    return all_fields

get_notification_recipients(event, request=None, obj=None)

Get notification recipients for an admin change event.

The default implementation returns None, which signals the dispatcher to use the built-in default behaviour: superusers always receive notifications, regular admin users receive them only if they have enabled NotificationPreference rows.

Subclasses may override this method to customise recipient selection, channels, or to bypass preference lookups entirely.

Parameters:

Name	Type	Description	Default
event	str	One of "create", "update", or "delete".	required
request	Any	Current request (optional, for accessing auth context).	None
obj	Any	The affected object instance (optional).	None

Returns:

Type	Description
list[dict[str, Any]] | None	list[{"id", "email", "phone", "channels"}] of recipient dicts,
list[dict[str, Any]] | None	or [] to disable notifications for this model entirely,
list[dict[str, Any]] | None	or None to use the default behaviour (superusers always,
list[dict[str, Any]] | None	regular admins only with enabled preferences).

Source code in fastapi_admin_kit/modeladmin.py

def get_notification_recipients(
    self, event: str, request: Any = None, obj: Any = None
) -> list[dict[str, Any]] | None:
    """Get notification recipients for an admin change event.

    The default implementation returns ``None``, which signals the dispatcher
    to use the built-in default behaviour: superusers always receive notifications,
    regular admin users receive them only if they have enabled
    ``NotificationPreference`` rows.

    Subclasses may override this method to customise recipient selection,
    channels, or to bypass preference lookups entirely.

    Args:
        event: One of ``"create"``, ``"update"``, or ``"delete"``.
        request: Current request (optional, for accessing auth context).
        obj: The affected object instance (optional).

    Returns:
        ``list[{"id", "email", "phone", "channels"}]`` of recipient dicts,
        or ``[]`` to disable notifications for this model entirely,
        or ``None`` to use the default behaviour (superusers always,
        regular admins only with enabled preferences).
    """
    return None

get_export_formats()

Get available export formats for this model.

Returns configured formats or defaults to CSV + Excel.

Source code in fastapi_admin_kit/modeladmin.py

def get_export_formats(self) -> dict[str, Any]:
    """Get available export formats for this model.

    Returns configured formats or defaults to CSV + Excel.
    """
    if self.export_formats:
        return self.export_formats
    from fastapi_admin_kit.export_import.registry import (
        get_available_export_formats,
    )

    formats = {}
    for name in get_available_export_formats():
        formats[name] = None
    return formats or {"csv": None}

get_import_formats()

Get available import formats for this model.

Returns configured formats or defaults to CSV + Excel.

Source code in fastapi_admin_kit/modeladmin.py

def get_import_formats(self) -> dict[str, Any]:
    """Get available import formats for this model.

    Returns configured formats or defaults to CSV + Excel.
    """
    if self.import_formats:
        return self.import_formats
    from fastapi_admin_kit.export_import.registry import (
        get_available_import_formats,
    )

    formats = {}
    for name in get_available_import_formats():
        formats[name] = None
    return formats or {"csv": None}

get_export_class(format_name)

Get the export class for a given format.

Parameters:

Name	Type	Description	Default
format_name	str	Format name (e.g., "csv", "excel")	required

Returns:

Type	Description
type | None	The export class or None if not available

Source code in fastapi_admin_kit/modeladmin.py

def get_export_class(self, format_name: str) -> type | None:
    """Get the export class for a given format.

    Args:
        format_name: Format name (e.g., "csv", "excel")

    Returns:
        The export class or None if not available
    """
    formats = self.get_export_formats()
    if format_name not in formats:
        return None

    format_class = formats[format_name]
    if format_class is None:
        # Use default from registry
        from fastapi_admin_kit.export_import.registry import (
            get_export_class,
        )

        return get_export_class(format_name)
    return format_class

get_import_class(format_name)

Get the import class for a given format.

Parameters:

Name	Type	Description	Default
format_name	str	Format name (e.g., "csv", "excel")	required

Returns:

Type	Description
type | None	The import class or None if not available

Source code in fastapi_admin_kit/modeladmin.py

def get_import_class(self, format_name: str) -> type | None:
    """Get the import class for a given format.

    Args:
        format_name: Format name (e.g., "csv", "excel")

    Returns:
        The import class or None if not available
    """
    formats = self.get_import_formats()
    if format_name not in formats:
        return None

    format_class = formats[format_name]
    if format_class is None:
        # Use default from registry
        from fastapi_admin_kit.export_import.registry import (
            get_import_class,
        )

        return get_import_class(format_name)
    return format_class

DatabaseConfig

fastapi_admin_kit.config.DatabaseConfig

Database connection configuration.

Two usage modes (url takes precedence):

1. Full URL — pass url="sqlite+aiosqlite:///./db.sqlite3". The URL is automatically inspected and its driver is upgraded to an async driver if necessary (e.g. sqlite:///… → sqlite+aiosqlite:///…).

2. Structured fields — set db_type, host, port, database, username, password individually.

Always creates a SQLAlchemy async engine via :meth:create_engine.

Source code in fastapi_admin_kit/config/database.py

class DatabaseConfig:
    """Database connection configuration.

    Two usage modes (``url`` takes precedence):

    1. **Full URL** — pass ``url="sqlite+aiosqlite:///./db.sqlite3"``.
       The URL is automatically inspected and its driver is upgraded to an
       async driver if necessary (e.g. ``sqlite:///…`` → ``sqlite+aiosqlite:///…``).

    2. **Structured fields** — set ``db_type``, ``host``, ``port``,
       ``database``, ``username``, ``password`` individually.

    Always creates a **SQLAlchemy async engine** via :meth:`create_engine`.
    """

    def __init__(
        self,
        db_type: DatabaseType = DatabaseType.SQLITE,
        url: str | None = None,
        host: str = "",
        port: int | None = None,
        database: str = "",
        username: str = "",
        password: str = "",
        echo: bool = False,
        pool_size: int = 5,
        max_overflow: int = 10,
        pool_pre_ping: bool = True,
        connect_args: dict[str, Any] | None = None,
    ) -> None:
        self.db_type = db_type
        self.url = url
        self.host = host
        self.port = port
        self.database = database
        self.username = username
        self.password = password
        self.echo = echo
        self.pool_size = pool_size
        self.max_overflow = max_overflow
        self.pool_pre_ping = pool_pre_ping
        self.connect_args = connect_args or {}

    def build_url(self) -> str:
        """Build or normalise the async connection URL from this config."""
        if self.url is not None:
            return _ensure_async_url(self.url, self.db_type)

        driver = _ASYNC_DRIVERS[self.db_type]

        if self.db_type == DatabaseType.SQLITE:
            return f"sqlite+{driver}:///{self.database}"

        port_part = f":{self.port}" if self.port else ""
        return (
            f"{self.db_type.value}+{driver}://"
            f"{self.username}:{self.password}@"
            f"{self.host}{port_part}/{self.database}"
        )

    def create_engine(self) -> Any:
        """Create a SQLAlchemy **async** engine from this config."""
        from sqlalchemy.ext.asyncio import create_async_engine

        url = self.build_url()
        kwargs: dict[str, Any] = {
            "echo": self.echo,
            "pool_pre_ping": self.pool_pre_ping,
        }

        if self.db_type != DatabaseType.SQLITE:
            kwargs["pool_size"] = self.pool_size
            kwargs["max_overflow"] = self.max_overflow

        # For SQLite, concurrent writers are serialized by file-level locks.
        # Without a busy timeout the driver fails immediately with
        # "database is locked" instead of waiting for the lock to clear.
        # Default a 30s busy timeout (mirrors the CLI engine setup) so brief
        # contention — e.g. streaming persistence vs an in-flight request —
        # resolves instead of erroring. User-supplied connect_args win.
        connect_args: dict[str, Any] = dict(self.connect_args)
        if self.db_type == DatabaseType.SQLITE:
            connect_args.setdefault("timeout", 30)
        if connect_args:
            kwargs["connect_args"] = connect_args

        return create_async_engine(url, **kwargs)

build_url()

Build or normalise the async connection URL from this config.

Source code in fastapi_admin_kit/config/database.py

def build_url(self) -> str:
    """Build or normalise the async connection URL from this config."""
    if self.url is not None:
        return _ensure_async_url(self.url, self.db_type)

    driver = _ASYNC_DRIVERS[self.db_type]

    if self.db_type == DatabaseType.SQLITE:
        return f"sqlite+{driver}:///{self.database}"

    port_part = f":{self.port}" if self.port else ""
    return (
        f"{self.db_type.value}+{driver}://"
        f"{self.username}:{self.password}@"
        f"{self.host}{port_part}/{self.database}"
    )

create_engine()

Create a SQLAlchemy async engine from this config.

Source code in fastapi_admin_kit/config/database.py

def create_engine(self) -> Any:
    """Create a SQLAlchemy **async** engine from this config."""
    from sqlalchemy.ext.asyncio import create_async_engine

    url = self.build_url()
    kwargs: dict[str, Any] = {
        "echo": self.echo,
        "pool_pre_ping": self.pool_pre_ping,
    }

    if self.db_type != DatabaseType.SQLITE:
        kwargs["pool_size"] = self.pool_size
        kwargs["max_overflow"] = self.max_overflow

    # For SQLite, concurrent writers are serialized by file-level locks.
    # Without a busy timeout the driver fails immediately with
    # "database is locked" instead of waiting for the lock to clear.
    # Default a 30s busy timeout (mirrors the CLI engine setup) so brief
    # contention — e.g. streaming persistence vs an in-flight request —
    # resolves instead of erroring. User-supplied connect_args win.
    connect_args: dict[str, Any] = dict(self.connect_args)
    if self.db_type == DatabaseType.SQLITE:
        connect_args.setdefault("timeout", 30)
    if connect_args:
        kwargs["connect_args"] = connect_args

    return create_async_engine(url, **kwargs)

DatabaseType

fastapi_admin_kit.config.DatabaseType

Bases: StrEnum

Supported database types for async connections.

Source code in fastapi_admin_kit/config/database.py

class DatabaseType(StrEnum):
    """Supported database types for async connections."""

    SQLITE = "sqlite"
    POSTGRESQL = "postgresql"
    MYSQL = "mysql"

Configuration Options

BehaviorConfig

fastapi_admin_kit.config.BehaviorConfig

Behavior configuration.

Source code in fastapi_admin_kit/config/behavior.py

class BehaviorConfig:
    """Behavior configuration."""

    def __init__(
        self,
        auto_discover: bool = True,
        dashboard_stats: list[str] | None = None,
        dashboard_charts: bool = True,
        dashboard_callback: str | None = None,
        dashboard_components: list[Any] | None = None,
        skip_models: list[str] | None = None,
        use_alembic: bool = False,
    ):
        self.auto_discover = auto_discover
        self.dashboard_stats = dashboard_stats or []
        self.dashboard_charts = dashboard_charts
        self.dashboard_callback = dashboard_callback
        self.dashboard_components = dashboard_components or []
        self.skip_models = set(skip_models) if skip_models else set()
        self.use_alembic = use_alembic

    def validate_behavior_config(self) -> None:
        """Validate behavior configuration."""
        pass

validate_behavior_config()

Validate behavior configuration.

Source code in fastapi_admin_kit/config/behavior.py

def validate_behavior_config(self) -> None:
    """Validate behavior configuration."""
    pass

AuthConfig

fastapi_admin_kit.config.AuthConfig

Authentication configuration.

Source code in fastapi_admin_kit/config/auth.py

class AuthConfig:
    """Authentication configuration."""

    def __init__(
        self,
        auth_model: type | None = None,
        auth_backend: Any | None = None,
        password_hasher: Any | None = None,
        session_ttl: int = 28800,
        session_cookie_name: str = "admin_session",
        session_secure: bool = True,
        superuser_emails: list[str] | None = None,
        password_min_length: int = 12,
        password_require_uppercase: bool = True,
        password_require_lowercase: bool = True,
        password_require_digit: bool = True,
        password_require_special: bool = True,
        session_samesite: str = "strict",
        access_token_ttl: int = 600,
        api_token_middleware: bool = True,
        api_token_strict: bool = False,
        trusted_proxies: list[str] | None = None,
    ):
        self.auth_model = auth_model
        self.auth_backend = auth_backend
        self.password_hasher = password_hasher
        self.session_ttl = session_ttl
        self.session_cookie_name = session_cookie_name
        self.session_secure = session_secure
        self.superuser_emails = superuser_emails or []
        self.password_min_length = password_min_length
        self.password_require_uppercase = password_require_uppercase
        self.password_require_lowercase = password_require_lowercase
        self.password_require_digit = password_require_digit
        self.password_require_special = password_require_special
        self.session_samesite = session_samesite
        # Short-lived JWT API access tokens (see api/auth.py). Refresh
        # tokens keep sessions alive; a stolen bearer token expires within
        # this window.
        self.access_token_ttl = access_token_ttl
        # Pre-validate bearer tokens on /api/* routes in middleware and
        # cache the decoded payload for the request.
        self.api_token_middleware = api_token_middleware
        # When True, /api/* routes (minus exempt paths) require a bearer
        # token even if the route itself has no auth dependency.
        self.api_token_strict = api_token_strict
        # Reverse proxies whose X-Forwarded-For header may be trusted when
        # they are the DIRECT peer of the app (single IPs or CIDR networks).
        # Empty (default) = never trust X-Forwarded-For; rate limiting and
        # audit logs use the socket peer address.
        self.trusted_proxies = [str(p) for p in trusted_proxies] if trusted_proxies else []

    def get_hasher(self) -> Any:
        """Return the configured password hasher, or default BcryptHasher."""
        if self.password_hasher is not None:
            return self.password_hasher
        from fastapi_admin_kit.auth.hasher import BcryptHasher

        return BcryptHasher

    def validate_auth_model(self) -> None:
        """Validate that auth_model satisfies AdminUserProtocol."""
        model = self.auth_model
        if model is None:
            return

        # Required: id, email
        required_attrs = ["id", "email"]
        missing = [attr for attr in required_attrs if not hasattr(model, attr)]
        if missing:
            raise ConfigError(
                f"auth_model {model.__name__!r} is missing required attributes: "
                f"{', '.join(missing)}. Every auth model must have id and email."
            )

        # Required: is_active, is_superuser (can be provided by AuthModelMixin)
        missing_flags = []
        if not hasattr(model, "is_active"):
            missing_flags.append("is_active")
        if not hasattr(model, "is_superuser"):
            missing_flags.append("is_superuser")
        if missing_flags:
            raise ConfigError(
                f"auth_model {model.__name__!r} is missing: {', '.join(missing_flags)}. "
                f"Use AuthModelMixin or add these columns to your model."
            )

        # Required: roles or role_ids (for RBAC)
        if not hasattr(model, "roles") and not hasattr(model, "role_ids"):
            raise ConfigError(
                f"auth_model {model.__name__!r} has no 'roles' relationship or "
                f"'role_ids' property. RBAC requires role lookups. "
                f"Use AuthModelMixin or define a roles relationship on your model."
            )

        # Check password-related attributes for authentication
        missing_auth = []
        if not hasattr(model, "password"):
            missing_auth.append("password")
        if not callable(getattr(model, "verify_password", None)):
            missing_auth.append("verify_password()")
        if missing_auth:
            raise ConfigError(
                f"auth_model {model.__name__!r} is missing password-related "
                f"attributes: {', '.join(missing_auth)}. "
                f"Use AuthModelMixin or implement password (str) and "
                f"verify_password(password) -> bool."
            )

get_hasher()

Return the configured password hasher, or default BcryptHasher.

Source code in fastapi_admin_kit/config/auth.py

def get_hasher(self) -> Any:
    """Return the configured password hasher, or default BcryptHasher."""
    if self.password_hasher is not None:
        return self.password_hasher
    from fastapi_admin_kit.auth.hasher import BcryptHasher

    return BcryptHasher

validate_auth_model()

Validate that auth_model satisfies AdminUserProtocol.

Source code in fastapi_admin_kit/config/auth.py

def validate_auth_model(self) -> None:
    """Validate that auth_model satisfies AdminUserProtocol."""
    model = self.auth_model
    if model is None:
        return

    # Required: id, email
    required_attrs = ["id", "email"]
    missing = [attr for attr in required_attrs if not hasattr(model, attr)]
    if missing:
        raise ConfigError(
            f"auth_model {model.__name__!r} is missing required attributes: "
            f"{', '.join(missing)}. Every auth model must have id and email."
        )

    # Required: is_active, is_superuser (can be provided by AuthModelMixin)
    missing_flags = []
    if not hasattr(model, "is_active"):
        missing_flags.append("is_active")
    if not hasattr(model, "is_superuser"):
        missing_flags.append("is_superuser")
    if missing_flags:
        raise ConfigError(
            f"auth_model {model.__name__!r} is missing: {', '.join(missing_flags)}. "
            f"Use AuthModelMixin or add these columns to your model."
        )

    # Required: roles or role_ids (for RBAC)
    if not hasattr(model, "roles") and not hasattr(model, "role_ids"):
        raise ConfigError(
            f"auth_model {model.__name__!r} has no 'roles' relationship or "
            f"'role_ids' property. RBAC requires role lookups. "
            f"Use AuthModelMixin or define a roles relationship on your model."
        )

    # Check password-related attributes for authentication
    missing_auth = []
    if not hasattr(model, "password"):
        missing_auth.append("password")
    if not callable(getattr(model, "verify_password", None)):
        missing_auth.append("verify_password()")
    if missing_auth:
        raise ConfigError(
            f"auth_model {model.__name__!r} is missing password-related "
            f"attributes: {', '.join(missing_auth)}. "
            f"Use AuthModelMixin or implement password (str) and "
            f"verify_password(password) -> bool."
        )

AuditConfig

fastapi_admin_kit.config.AuditConfig

Audit configuration.

Source code in fastapi_admin_kit/config/audit.py

class AuditConfig:
    """Audit configuration."""

    def __init__(
        self,
        audit_retention_days: int = 365,
    ):
        self.audit_retention_days = audit_retention_days

    def validate_audit_config(self) -> None:
        """Validate audit configuration."""
        if self.audit_retention_days < 0:
            raise ConfigError("audit_retention_days must be non-negative")

validate_audit_config()

Validate audit configuration.

Source code in fastapi_admin_kit/config/audit.py

def validate_audit_config(self) -> None:
    """Validate audit configuration."""
    if self.audit_retention_days < 0:
        raise ConfigError("audit_retention_days must be non-negative")

UIConfig

fastapi_admin_kit.config.UIConfig

UI configuration — wraps ThemeConfig + component-level options.

Source code in fastapi_admin_kit/config/ui.py

class UIConfig:
    """UI configuration — wraps ThemeConfig + component-level options."""

    def __init__(
        self,
        title: str = "FastAPI Admin Kit",
        logo_url: str | None = None,
        favicon_url: str | None = None,
        primary_color: str = "#0ea5e9",
        primary_color_dark: str = "#0284c7",
        dark_mode_default: bool = False,
        per_page_default: int = 25,
        # Theme
        theme: ThemeConfig | None = None,
        # Component config
        sidebar_style: str = "default",
        sidebar_show_icons: bool = True,
        sidebar_show_badges: bool = True,
        sidebar_group_style: str = "label",
        sidebar_position: str = "left",
        table_style: str = "default",
        table_hover_effect: bool = True,
        table_row_height: str = "normal",
        form_layout: str = "two-column",
        form_label_position: str = "top",
        form_spacing: str = "normal",
        form_card_style: bool = True,
        dashboard_grid: str = "auto",
        dashboard_card_style: str = "default",
        dashboard_stat_size: str = "normal",
        content_width: str = "default",
        topbar_style: str = "default",
        sticky_header: bool = True,
        # Custom injection
        custom_css: str = "",
        custom_css_url: str = "",
        custom_js: str = "",
        custom_js_url: str = "",
        # Feature toggles
        show_history: bool = True,
        show_view_on_site: bool = True,
        show_back_button: bool = False,
        environment_label: str | None = None,
        environment_color: str = "info",
        site_url: str = "/",
        site_symbol: str | None = None,
        login_background_url: str | None = None,
        # Mobile
        mobile_sidebar: str = "overlay",
        mobile_topbar_height: str = "48px",
        mobile_content_padding: str = "16px",
    ):
        self.title = title
        self.logo_url = logo_url
        self.favicon_url = favicon_url
        self.primary_color = primary_color
        self.primary_color_dark = primary_color_dark
        self.dark_mode_default = dark_mode_default
        self.per_page_default = per_page_default
        self.theme = theme
        self.sidebar_style = sidebar_style
        self.sidebar_show_icons = sidebar_show_icons
        self.sidebar_show_badges = sidebar_show_badges
        self.sidebar_group_style = sidebar_group_style
        self.sidebar_position = sidebar_position
        self.table_style = table_style
        self.table_hover_effect = table_hover_effect
        self.table_row_height = table_row_height
        self.form_layout = form_layout
        self.form_label_position = form_label_position
        self.form_spacing = form_spacing
        self.form_card_style = form_card_style
        self.dashboard_grid = dashboard_grid
        self.dashboard_card_style = dashboard_card_style
        self.dashboard_stat_size = dashboard_stat_size
        self.content_width = content_width
        self.topbar_style = topbar_style
        self.sticky_header = sticky_header
        self.custom_css = custom_css
        self.custom_css_url = custom_css_url
        self.custom_js = custom_js
        self.custom_js_url = custom_js_url
        self.show_history = show_history
        self.show_view_on_site = show_view_on_site
        self.show_back_button = show_back_button
        self.environment_label = environment_label
        self.environment_color = environment_color
        self.site_url = site_url
        self.site_symbol = site_symbol
        self.login_background_url = login_background_url
        self.mobile_sidebar = mobile_sidebar
        self.mobile_topbar_height = mobile_topbar_height
        self.mobile_content_padding = mobile_content_padding

    def apply_to_template_context(self) -> dict:
        """Apply UI configuration to template context."""
        ctx = {
            "title": self.title,
            "logo_url": self.logo_url,
            "favicon_url": self.favicon_url,
            "primary_color": self.primary_color,
            "primary_color_dark": self.primary_color_dark,
            "dark_mode_default": self.dark_mode_default,
            "per_page_default": self.per_page_default,
            "sidebar_style": self.sidebar_style,
            "sidebar_show_icons": self.sidebar_show_icons,
            "sidebar_show_badges": self.sidebar_show_badges,
            "sidebar_group_style": self.sidebar_group_style,
            "sidebar_position": self.sidebar_position,
            "table_style": self.table_style,
            "table_hover_effect": self.table_hover_effect,
            "table_row_height": self.table_row_height,
            "form_layout": self.form_layout,
            "form_label_position": self.form_label_position,
            "form_spacing": self.form_spacing,
            "form_card_style": self.form_card_style,
            "dashboard_grid": self.dashboard_grid,
            "dashboard_card_style": self.dashboard_card_style,
            "dashboard_stat_size": self.dashboard_stat_size,
            "content_width": self.content_width,
            "topbar_style": self.topbar_style,
            "sticky_header": self.sticky_header,
            "show_history": self.show_history,
            "show_view_on_site": self.show_view_on_site,
            "show_back_button": self.show_back_button,
            "environment_label": self.environment_label,
            "environment_color": self.environment_color,
            "site_url": self.site_url,
            "site_symbol": self.site_symbol,
            "login_background_url": self.login_background_url,
            "mobile_sidebar": self.mobile_sidebar,
            "mobile_topbar_height": self.mobile_topbar_height,
            "mobile_content_padding": self.mobile_content_padding,
        }
        if self.theme:
            ctx.update(self.theme.to_context())
        return ctx

apply_to_template_context()

Apply UI configuration to template context.

Source code in fastapi_admin_kit/config/ui.py

def apply_to_template_context(self) -> dict:
    """Apply UI configuration to template context."""
    ctx = {
        "title": self.title,
        "logo_url": self.logo_url,
        "favicon_url": self.favicon_url,
        "primary_color": self.primary_color,
        "primary_color_dark": self.primary_color_dark,
        "dark_mode_default": self.dark_mode_default,
        "per_page_default": self.per_page_default,
        "sidebar_style": self.sidebar_style,
        "sidebar_show_icons": self.sidebar_show_icons,
        "sidebar_show_badges": self.sidebar_show_badges,
        "sidebar_group_style": self.sidebar_group_style,
        "sidebar_position": self.sidebar_position,
        "table_style": self.table_style,
        "table_hover_effect": self.table_hover_effect,
        "table_row_height": self.table_row_height,
        "form_layout": self.form_layout,
        "form_label_position": self.form_label_position,
        "form_spacing": self.form_spacing,
        "form_card_style": self.form_card_style,
        "dashboard_grid": self.dashboard_grid,
        "dashboard_card_style": self.dashboard_card_style,
        "dashboard_stat_size": self.dashboard_stat_size,
        "content_width": self.content_width,
        "topbar_style": self.topbar_style,
        "sticky_header": self.sticky_header,
        "show_history": self.show_history,
        "show_view_on_site": self.show_view_on_site,
        "show_back_button": self.show_back_button,
        "environment_label": self.environment_label,
        "environment_color": self.environment_color,
        "site_url": self.site_url,
        "site_symbol": self.site_symbol,
        "login_background_url": self.login_background_url,
        "mobile_sidebar": self.mobile_sidebar,
        "mobile_topbar_height": self.mobile_topbar_height,
        "mobile_content_padding": self.mobile_content_padding,
    }
    if self.theme:
        ctx.update(self.theme.to_context())
    return ctx

StorageConfig

fastapi_admin_kit.config.StorageConfig

Storage configuration.

Source code in fastapi_admin_kit/config/storage.py

class StorageConfig:
    """Storage configuration."""

    def __init__(
        self,
        storage: Any | None = None,
        uploads_url: str = "/uploads",
    ):
        self.storage = storage
        self.uploads_url = uploads_url

    def validate_storage_config(self) -> None:
        """Validate storage configuration."""
        if self.uploads_url and not self.uploads_url.startswith("/"):
            raise ConfigError("uploads_url must start with '/' or be empty")

validate_storage_config()

Validate storage configuration.

Source code in fastapi_admin_kit/config/storage.py

def validate_storage_config(self) -> None:
    """Validate storage configuration."""
    if self.uploads_url and not self.uploads_url.startswith("/"):
        raise ConfigError("uploads_url must start with '/' or be empty")

NavConfig

fastapi_admin_kit.config.NavConfig

Navigation configuration.

Source code in fastapi_admin_kit/config/nav.py

class NavConfig:
    """Navigation configuration."""

    def __init__(
        self,
        nav_groups: list[Any] | None = None,
        sidebar_builder: Any | None = None,
        require_tags: bool = False,
        site_dropdown: list[dict[str, str]] | None = None,
        dashboard_permission: str | None = None,
        settings_permission: str | None = None,
        sidebar_bottom_links: list[dict[str, str]] | None = None,
    ):
        self.nav_groups = nav_groups or []
        self.sidebar_builder = sidebar_builder
        self.require_tags = require_tags
        self.site_dropdown = site_dropdown or []
        self.dashboard_permission = dashboard_permission
        self.settings_permission = settings_permission
        self.sidebar_bottom_links = sidebar_bottom_links or []

    def validate_nav_config(self) -> None:
        """Validate navigation configuration."""
        if self.require_tags and not self.nav_groups:
            raise ConfigError("require_tags=True requires nav_groups to be configured")

validate_nav_config()

Validate navigation configuration.

Source code in fastapi_admin_kit/config/nav.py

def validate_nav_config(self) -> None:
    """Validate navigation configuration."""
    if self.require_tags and not self.nav_groups:
        raise ConfigError("require_tags=True requires nav_groups to be configured")

Actions

Action

fastapi_admin_kit.actions.base.Action

Bases: ABC

Abstract base class for admin actions.

Supports list-level, row-level, detail-level, and submit-line actions with Unfold-style icon, variant, and permission configuration.

Source code in fastapi_admin_kit/actions/base.py

class Action(ABC):
    """Abstract base class for admin actions.

    Supports list-level, row-level, detail-level, and submit-line actions
    with Unfold-style icon, variant, and permission configuration.
    """

    def __init__(
        self,
        name: str,
        label: str = "",
        confirmation_message: str = "",
        icon: str | None = None,
        variant: str = "default",
        permissions: list[str] | None = None,
        location: str = "list",
        description: str = "",
    ) -> None:
        self.name = name
        self.label = label or name.replace("_", " ").title()
        self.confirmation_message = confirmation_message
        self.icon = icon
        self.variant = variant  # default, primary, danger, success, warning
        self.permissions = permissions or []
        self.location = location  # list, row, detail, submit_line
        self.description = description

    @abstractmethod
    async def execute(self, objects: list[Any], request: Request | None) -> None:
        """Run the action against the selected objects."""
        ...

    def get_confirmation_message(self) -> str:
        return self.confirmation_message or f"Run {self.label}?"

    def has_permission(self, user: Any) -> bool:
        """Check if the user has permission to run this action."""
        if not self.permissions:
            return True
        if getattr(user, "is_superuser", False):
            return True
        user_perms = getattr(user, "permissions", []) or []
        return any(p in user_perms for p in self.permissions)

execute(objects, request) abstractmethod async

Run the action against the selected objects.

Source code in fastapi_admin_kit/actions/base.py

@abstractmethod
async def execute(self, objects: list[Any], request: Request | None) -> None:
    """Run the action against the selected objects."""
    ...

has_permission(user)

Check if the user has permission to run this action.

Source code in fastapi_admin_kit/actions/base.py

def has_permission(self, user: Any) -> bool:
    """Check if the user has permission to run this action."""
    if not self.permissions:
        return True
    if getattr(user, "is_superuser", False):
        return True
    user_perms = getattr(user, "permissions", []) or []
    return any(p in user_perms for p in self.permissions)

ModelAction

fastapi_admin_kit.actions.base.ModelAction

Bases: Action

Action that operates on a single model instance (row/detail actions).

Source code in fastapi_admin_kit/actions/base.py

class ModelAction(Action):
    """Action that operates on a single model instance (row/detail actions)."""

    async def execute_single(self, obj: Any, request: Request | None) -> None:
        """Run the action on a single object.

        Override this instead of execute for single-object actions.
        """
        await self.execute([obj], request)

execute_single(obj, request) async

Run the action on a single object.

Override this instead of execute for single-object actions.

Source code in fastapi_admin_kit/actions/base.py

async def execute_single(self, obj: Any, request: Request | None) -> None:
    """Run the action on a single object.

    Override this instead of execute for single-object actions.
    """
    await self.execute([obj], request)

Pagination

BasePagination

fastapi_admin_kit.pagination.base.BasePagination

Bases: ABC

Abstract base for all pagination strategies.

Source code in fastapi_admin_kit/pagination/base.py

class BasePagination(ABC):
    """Abstract base for all pagination strategies."""

    @abstractmethod
    async def paginate(
        self,
        stmt: Any,
        session: Any,
        per_page: int,
        page: int = 1,
        after: str | None = None,
        before: str | None = None,
        pk_col: Any = None,
        model: Any = None,
        query_adapter: Any = None,
    ) -> PaginationResult:
        """Execute paginated query and return results.

        Args:
            stmt: The base query/statement to paginate.
            session: Session or SessionBackend for query execution.
            per_page: Number of items per page.
            page: Page number (for offset pagination).
            after: Cursor for forward pagination.
            before: Cursor for backward pagination.
            pk_col: Primary key column for cursor pagination.
            model: SQLAlchemy model class.
            query_adapter: QueryBackend for ORM-agnostic query construction.
        """
        ...

paginate(stmt, session, per_page, page=1, after=None, before=None, pk_col=None, model=None, query_adapter=None) abstractmethod async

Execute paginated query and return results.

Parameters:

Name	Type	Description	Default
stmt	Any	The base query/statement to paginate.	required
session	Any	Session or SessionBackend for query execution.	required
per_page	int	Number of items per page.	required
page	int	Page number (for offset pagination).	1
after	str | None	Cursor for forward pagination.	None
before	str | None	Cursor for backward pagination.	None
pk_col	Any	Primary key column for cursor pagination.	None
model	Any	SQLAlchemy model class.	None
query_adapter	Any	QueryBackend for ORM-agnostic query construction.	None

Source code in fastapi_admin_kit/pagination/base.py

@abstractmethod
async def paginate(
    self,
    stmt: Any,
    session: Any,
    per_page: int,
    page: int = 1,
    after: str | None = None,
    before: str | None = None,
    pk_col: Any = None,
    model: Any = None,
    query_adapter: Any = None,
) -> PaginationResult:
    """Execute paginated query and return results.

    Args:
        stmt: The base query/statement to paginate.
        session: Session or SessionBackend for query execution.
        per_page: Number of items per page.
        page: Page number (for offset pagination).
        after: Cursor for forward pagination.
        before: Cursor for backward pagination.
        pk_col: Primary key column for cursor pagination.
        model: SQLAlchemy model class.
        query_adapter: QueryBackend for ORM-agnostic query construction.
    """
    ...

PaginationResult

fastapi_admin_kit.pagination.base.PaginationResult dataclass

Unified result from any pagination strategy.

Source code in fastapi_admin_kit/pagination/base.py

@dataclass
class PaginationResult:
    """Unified result from any pagination strategy."""

    items: list[Any]
    total: int
    per_page: int
    page: int | None = None
    total_pages: int | None = None
    next_cursor: str | None = None
    has_next: bool = False
    mode: str = "offset"

OffsetPagination

fastapi_admin_kit.pagination.offset.OffsetPagination

Bases: BasePagination

Traditional page-number pagination using OFFSET/LIMIT.

Source code in fastapi_admin_kit/pagination/offset.py

class OffsetPagination(BasePagination):
    """Traditional page-number pagination using OFFSET/LIMIT."""

    async def paginate(
        self,
        stmt: Any,
        session: Any,
        per_page: int,
        page: int = 1,
        query_adapter: Any = None,
        **kw: Any,
    ) -> PaginationResult:
        session = as_session_backend(session)
        if query_adapter is not None:
            count_q = query_adapter.count(stmt)
            total = await session.count(count_q)
        else:
            from sqlalchemy import func, select

            count_q = select(func.count()).select_from(stmt.subquery())
            total = await session.count(count_q)

        total_pages = max(1, math.ceil(total / per_page))
        page = max(1, min(page, total_pages))
        offset = (page - 1) * per_page

        if query_adapter is not None:
            stmt = query_adapter.offset(stmt, offset)
            stmt = query_adapter.limit(stmt, per_page)
        else:
            stmt = stmt.offset(offset).limit(per_page)

        items = list(await session.all(stmt, unique=True))

        return PaginationResult(
            items=items,
            total=total,
            per_page=per_page,
            page=page,
            total_pages=total_pages,
            mode="offset",
        )

CursorPagination

fastapi_admin_kit.pagination.cursor.CursorPagination

Bases: BasePagination

Keyset pagination using opaque base64-encoded cursors.

Uses a configurable column (default: primary key) for cursor values. Supports forward (after) and backward (before) navigation.

Source code in fastapi_admin_kit/pagination/cursor.py

class CursorPagination(BasePagination):
    """Keyset pagination using opaque base64-encoded cursors.

    Uses a configurable column (default: primary key) for cursor values.
    Supports forward (after) and backward (before) navigation.
    """

    def __init__(self, cursor_column: str | None = None):
        self.cursor_column = cursor_column

    def _decode_cursor(self, cursor: str) -> Any:
        """Decode base64 cursor to Python value, or parse as plain value."""
        try:
            return json.loads(base64.b64decode(cursor))
        except Exception:
            # Fallback: try to parse as plain value (integer, float, etc.)
            try:
                return int(cursor)
            except ValueError:
                try:
                    return float(cursor)
                except ValueError:
                    return cursor

    def _encode_cursor(self, value: Any) -> str:
        """Encode Python value to base64 cursor string."""
        return base64.b64encode(json.dumps(value).encode()).decode()

    async def paginate(
        self,
        stmt: Any,
        session: Any,
        per_page: int,
        page: int = 1,
        after: str | None = None,
        before: str | None = None,
        pk_col: Any = None,
        model: Any = None,
        query_adapter: Any = None,
    ) -> PaginationResult:
        session = as_session_backend(session)
        # Determine cursor column
        if self.cursor_column and model is not None:
            col = getattr(model, self.cursor_column)
        elif pk_col is not None:
            col = pk_col
        else:
            raise ValueError(
                "CursorPagination requires either cursor_column on a model "
                "or pk_col to be provided."
            )

        # Apply cursor filter
        if after:
            cursor_val = self._decode_cursor(after)
            if query_adapter is not None:
                stmt = query_adapter.where(stmt, col > cursor_val)
            else:
                stmt = stmt.where(col > cursor_val)
        elif before:
            cursor_val = self._decode_cursor(before)
            if query_adapter is not None:
                stmt = query_adapter.where(stmt, col < cursor_val)
                stmt = query_adapter.order_by(stmt, f"-{col.key}")
            else:
                stmt = stmt.where(col < cursor_val)
                from sqlalchemy import desc as sa_desc

                stmt = stmt.order_by(sa_desc(col))

        # Count filtered total
        if query_adapter is not None:
            count_q = query_adapter.count(stmt)
            total = await session.count(count_q)
        else:
            from sqlalchemy import func, select

            count_q = select(func.count()).select_from(stmt.subquery())
            total = await session.count(count_q)

        # Fetch per_page + 1 to detect has_next
        if query_adapter is not None:
            stmt = query_adapter.limit(stmt, per_page + 1)
        else:
            stmt = stmt.limit(per_page + 1)

        items = list(await session.all(stmt, unique=True))

        # For backward pagination, reverse back to natural order
        if before:
            items = list(reversed(items))

        has_next = len(items) > per_page
        if has_next:
            items = items[:per_page]

        # Build next cursor from last item
        next_cursor = None
        if has_next and items:
            last_val = getattr(items[-1], self.cursor_column or "id")
            next_cursor = self._encode_cursor(last_val)

        return PaginationResult(
            items=items,
            total=total,
            per_page=per_page,
            next_cursor=next_cursor,
            has_next=has_next,
            mode="cursor",
        )

DynamicPagination

fastapi_admin_kit.pagination.dynamic.DynamicPagination

Bases: BasePagination

Automatically switches between offset and cursor pagination.

Uses offset for small datasets (fast page jumping), cursor for large datasets (consistent performance at any depth).

Source code in fastapi_admin_kit/pagination/dynamic.py

class DynamicPagination(BasePagination):
    """Automatically switches between offset and cursor pagination.

    Uses offset for small datasets (fast page jumping),
    cursor for large datasets (consistent performance at any depth).
    """

    def __init__(
        self,
        cursor_column: str | None = None,
        threshold: int = 1000,
    ):
        self.cursor_column = cursor_column
        self.threshold = threshold
        self._offset = OffsetPagination()
        self._cursor = CursorPagination(cursor_column=cursor_column)

    async def paginate(
        self,
        stmt: Any,
        session: Any,
        per_page: int,
        query_adapter: Any = None,
        **kw: Any,
    ) -> PaginationResult:
        session = as_session_backend(session)
        # Count total to decide strategy
        if query_adapter is not None:
            count_q = query_adapter.count(stmt)
            total = await session.count(count_q)
        else:
            from sqlalchemy import func, select

            count_q = select(func.count()).select_from(stmt.subquery())
            total = await session.count(count_q)

        if total <= self.threshold:
            result = await self._offset.paginate(
                stmt, session, per_page, query_adapter=query_adapter, **kw
            )
        else:
            result = await self._cursor.paginate(
                stmt, session, per_page, query_adapter=query_adapter, **kw
            )

        result.mode = "dynamic_offset" if total <= self.threshold else "dynamic_cursor"
        return result

Permissions

PermissionChecker

fastapi_admin_kit.auth.permissions.PermissionChecker

Per-request permission checker.

Instantiated once per request via Depends(get_permission_checker). Caches permission results in-memory for the lifetime of the request.

Merges permissions from: 1. All assigned roles (M2M via admin_user_roles) 2. Direct per-user overrides (UserPermission)

Role permissions are OR'd together, then direct overrides are OR'd on top.

Source code in fastapi_admin_kit/auth/permissions.py

class PermissionChecker:
    """Per-request permission checker.

    Instantiated once per request via ``Depends(get_permission_checker)``.
    Caches permission results in-memory for the lifetime of the request.

    Merges permissions from:
    1. All assigned roles (M2M via admin_user_roles)
    2. Direct per-user overrides (UserPermission)

    Role permissions are OR'd together, then direct overrides are OR'd on top.
    """

    def __init__(
        self,
        session: AsyncSession,
        user: AdminUserProtocol,
        *,
        user_snapshot: dict[str, object] | None = None,
    ) -> None:
        self.session = as_session_backend(session)
        self.user = user
        snap = user_snapshot or {}
        self._is_superuser: bool = (
            bool(snap["is_superuser"]) if "is_superuser" in snap else bool(user.is_superuser)
        )
        self._role_ids: list[int] = (
            snap["role_ids"] if "role_ids" in snap else getattr(user, "role_ids", [])
        )
        self._user_id: int | str | None = snap.get("id") if snap else getattr(user, "id", None)
        self._role_cache: dict[str, PermissionSet | None] | None = None
        self._direct_cache: dict[str, PermissionSet | None] | None = None
        self._cache: dict[tuple[str, str], bool] = {}
        self._field_cache: dict[tuple[str, str], set[str] | None] = {}

    async def _load_role_permissions(self) -> dict[str, PermissionSet | None]:
        """Load and cache all role-based permissions, merged with OR logic."""
        if self._role_cache is not None:
            return self._role_cache

        self._role_cache = {}
        if not self._role_ids:
            return self._role_cache

        from sqlalchemy import select

        from fastapi_admin_kit.auth.models import admin_role_permissions

        for perm in await self.session.all(
            select(Permission)
            .join(
                admin_role_permissions,
                Permission.id == admin_role_permissions.c.permission_id,
            )
            .where(admin_role_permissions.c.role_id.in_(self._role_ids))
        ):
            table = perm.table_name
            if table not in self._role_cache:
                self._role_cache[table] = PermissionSet()
            ps = self._role_cache[table]
            if perm.can_view:
                ps.can_view = True
            if perm.can_create:
                ps.can_create = True
            if perm.can_edit:
                ps.can_edit = True
            if perm.can_delete:
                ps.can_delete = True
            if perm.can_export:
                ps.can_export = True
            if perm.can_import:
                ps.can_import = True

        return self._role_cache

    async def _load_direct_permissions(self) -> dict[str, PermissionSet | None]:
        """Load and cache direct per-user permission overrides."""
        if self._direct_cache is not None:
            return self._direct_cache

        self._direct_cache = {}
        if self._user_id is None:
            return self._direct_cache

        from sqlalchemy import select

        from fastapi_admin_kit.auth.models import Permission

        result = await self.session.execute(
            select(UserPermission, Permission)
            .join(Permission, UserPermission.permission_id == Permission.id)
            .where(UserPermission.user_id == self._user_id)
        )
        for up, perm in result:
            table = perm.table_name
            if table not in self._direct_cache:
                self._direct_cache[table] = PermissionSet()
            ps = self._direct_cache[table]
            if perm.can_view:
                ps.can_view = True
            if perm.can_create:
                ps.can_create = True
            if perm.can_edit:
                ps.can_edit = True
            if perm.can_delete:
                ps.can_delete = True
            if perm.can_export:
                ps.can_export = True
            if perm.can_import:
                ps.can_import = True

        return self._direct_cache

    async def _get_merged_permission(self, table_name: str, action: str) -> bool:
        """Get merged permission for a table+action across roles and direct overrides."""
        role_perms = await self._load_role_permissions()
        direct_perms = await self._load_direct_permissions()

        attr = f"can_{action}"

        role_ps = role_perms.get(table_name)
        direct_ps = direct_perms.get(table_name)

        role_val = getattr(role_ps, attr, False) if role_ps else False
        direct_val = getattr(direct_ps, attr, False) if direct_ps else False

        return role_val or direct_val

    async def has_permission(self, table_name: str, action: str) -> bool:
        """Return True if the current user may perform *action* on *table_name*.

        Actions: ``"view"`` | ``"create"`` | ``"edit"`` | ``"delete"`` | ``"export"`` | ``"import"``

        Superusers always return True. Results are cached per-request.
        """
        if self._is_superuser:
            return True

        cache_key = (table_name, action)
        if cache_key in self._cache:
            return self._cache[cache_key]

        result_bool = await self._get_merged_permission(table_name, action)
        self._cache[cache_key] = result_bool
        return result_bool

    async def get_allowed_fields(self, table_name: str, mode: str) -> set[str] | None:
        """Return the set of field names the user may access, or ``None``.

        mode: ``"view"`` | ``"edit"``

        Semantics:
        - ``None`` → no field-level restrictions exist → all fields allowed.
        - Empty ``set()`` → restriction rows exist but none grant access → no fields.
        - Non-empty ``set()`` → only those field names are permitted.
        """
        if self._is_superuser:
            return None

        cache_key = (table_name, mode)
        if cache_key in self._field_cache:
            return self._field_cache[cache_key]

        # No role and no direct permissions → all fields restricted
        if not self._role_ids and self._user_id is None:
            self._field_cache[cache_key] = set()
            return set()

        # No field-level restrictions in this system anymore
        self._field_cache[cache_key] = None
        return None

    def permission_set(self, table_name: str) -> PermissionSet:
        """Return a :class:`PermissionSet` for convenient template / UI use.

        Warning: This reads from the async-populated cache. If ``load_permissions()``
        was not called first, all values will be ``False`` (except for superusers).
        Always call ``await load_permissions(table_name)`` before using this method.
        """
        if self._is_superuser:
            return PermissionSet(
                can_view=True,
                can_create=True,
                can_edit=True,
                can_delete=True,
                can_export=True,
                can_import=True,
            )
        # Check if cache has been populated for this table
        cache_populated = any(key[0] == table_name for key in self._cache)
        if not cache_populated:
            import logging

            logging.getLogger(__name__).warning(
                "permission_set('%s') called before load_permissions() — "
                "returning default (all-False). Always call await load_permissions() first.",
                table_name,
            )
        return PermissionSet(
            can_view=self._cache.get((table_name, "view"), False),
            can_create=self._cache.get((table_name, "create"), False),
            can_edit=self._cache.get((table_name, "edit"), False),
            can_delete=self._cache.get((table_name, "delete"), False),
            can_export=self._cache.get((table_name, "export"), False),
            can_import=self._cache.get((table_name, "import"), False),
        )

    async def load_permissions(self, table_name: str) -> PermissionSet:
        """Async method to load and cache all permissions for a table.

        Call this before using ``permission_set()`` to ensure the cache is populated.
        """
        await self.has_permission(table_name, "view")
        await self.has_permission(table_name, "create")
        await self.has_permission(table_name, "edit")
        await self.has_permission(table_name, "delete")
        await self.has_permission(table_name, "export")
        await self.has_permission(table_name, "import")
        return self.permission_set(table_name)

has_permission(table_name, action) async

Return True if the current user may perform action on table_name.

Actions: "view" | "create" | "edit" | "delete" | "export" | "import"

Superusers always return True. Results are cached per-request.

Source code in fastapi_admin_kit/auth/permissions.py

async def has_permission(self, table_name: str, action: str) -> bool:
    """Return True if the current user may perform *action* on *table_name*.

    Actions: ``"view"`` | ``"create"`` | ``"edit"`` | ``"delete"`` | ``"export"`` | ``"import"``

    Superusers always return True. Results are cached per-request.
    """
    if self._is_superuser:
        return True

    cache_key = (table_name, action)
    if cache_key in self._cache:
        return self._cache[cache_key]

    result_bool = await self._get_merged_permission(table_name, action)
    self._cache[cache_key] = result_bool
    return result_bool

get_allowed_fields(table_name, mode) async

Return the set of field names the user may access, or None.

mode: "view" | "edit"

Semantics: - None → no field-level restrictions exist → all fields allowed. - Empty set() → restriction rows exist but none grant access → no fields. - Non-empty set() → only those field names are permitted.

Source code in fastapi_admin_kit/auth/permissions.py

async def get_allowed_fields(self, table_name: str, mode: str) -> set[str] | None:
    """Return the set of field names the user may access, or ``None``.

    mode: ``"view"`` | ``"edit"``

    Semantics:
    - ``None`` → no field-level restrictions exist → all fields allowed.
    - Empty ``set()`` → restriction rows exist but none grant access → no fields.
    - Non-empty ``set()`` → only those field names are permitted.
    """
    if self._is_superuser:
        return None

    cache_key = (table_name, mode)
    if cache_key in self._field_cache:
        return self._field_cache[cache_key]

    # No role and no direct permissions → all fields restricted
    if not self._role_ids and self._user_id is None:
        self._field_cache[cache_key] = set()
        return set()

    # No field-level restrictions in this system anymore
    self._field_cache[cache_key] = None
    return None

permission_set(table_name)

Return a :class:PermissionSet for convenient template / UI use.

Warning: This reads from the async-populated cache. If load_permissions() was not called first, all values will be False (except for superusers). Always call await load_permissions(table_name) before using this method.

Source code in fastapi_admin_kit/auth/permissions.py

def permission_set(self, table_name: str) -> PermissionSet:
    """Return a :class:`PermissionSet` for convenient template / UI use.

    Warning: This reads from the async-populated cache. If ``load_permissions()``
    was not called first, all values will be ``False`` (except for superusers).
    Always call ``await load_permissions(table_name)`` before using this method.
    """
    if self._is_superuser:
        return PermissionSet(
            can_view=True,
            can_create=True,
            can_edit=True,
            can_delete=True,
            can_export=True,
            can_import=True,
        )
    # Check if cache has been populated for this table
    cache_populated = any(key[0] == table_name for key in self._cache)
    if not cache_populated:
        import logging

        logging.getLogger(__name__).warning(
            "permission_set('%s') called before load_permissions() — "
            "returning default (all-False). Always call await load_permissions() first.",
            table_name,
        )
    return PermissionSet(
        can_view=self._cache.get((table_name, "view"), False),
        can_create=self._cache.get((table_name, "create"), False),
        can_edit=self._cache.get((table_name, "edit"), False),
        can_delete=self._cache.get((table_name, "delete"), False),
        can_export=self._cache.get((table_name, "export"), False),
        can_import=self._cache.get((table_name, "import"), False),
    )

load_permissions(table_name) async

Async method to load and cache all permissions for a table.

Call this before using permission_set() to ensure the cache is populated.

Source code in fastapi_admin_kit/auth/permissions.py

async def load_permissions(self, table_name: str) -> PermissionSet:
    """Async method to load and cache all permissions for a table.

    Call this before using ``permission_set()`` to ensure the cache is populated.
    """
    await self.has_permission(table_name, "view")
    await self.has_permission(table_name, "create")
    await self.has_permission(table_name, "edit")
    await self.has_permission(table_name, "delete")
    await self.has_permission(table_name, "export")
    await self.has_permission(table_name, "import")
    return self.permission_set(table_name)

Views

ListView

fastapi_admin_kit.views.class_views.ListView

Bases: BaseView

Orchestrates list view: query -> render HTML or API.

Source code in fastapi_admin_kit/views/class_views.py

class ListView(BaseView):
    """Orchestrates list view: query -> render HTML or API."""

    html_renderer_class = ListHTMLRenderer
    api_renderer_class = ListAPIRenderer

    def _build_display_columns(self) -> list[DisplayColumn]:
        """Build display column metadata."""
        from sqlalchemy import inspect as sa_inspect

        model = self.registered.model
        mapper = sa_inspect(model)
        rel_names = {r.key for r in mapper.relationships}

        list_display = self.admin.list_display or [
            c.name for c in self.registered.columns if c.name != "id"
        ]

        # Collect @column decorated methods (check _column_options attribute)
        decorated_columns: dict[str, Any] = {}
        for attr_name in list_display:
            method = getattr(self.admin, attr_name, None)
            if method and hasattr(method, "_column_options"):
                decorated_columns[attr_name] = method._column_options

        display_columns = []
        for col_name in list_display:
            label = col_name.replace("_", " ").title()
            display_fn = None
            options = None

            # Check @column decorator first
            if col_name in decorated_columns:
                options = decorated_columns[col_name]
                display_fn = getattr(self.admin, col_name)
                label = options.header or label
            # Check display_functions dict fallback
            elif self.admin.display_functions and col_name in self.admin.display_functions:
                display_fn = self.admin.display_functions[col_name]

            display_columns.append(
                DisplayColumn(col_name, label, col_name in rel_names, display_fn, options)
            )
        return display_columns

    async def _build_filter_fields(self, request: Request) -> dict[str, dict[str, Any]]:
        """Build filter field metadata."""
        if not self.admin.list_filter:
            return {}
        session = get_db_session(request)
        model = self.registered.model
        filter_fields: dict[str, dict[str, Any]] = {}
        for filter_field in self.admin.list_filter:
            filter_fields[filter_field] = await self.query_provider._get_filter_choices(
                request, model, filter_field, session
            )
        return filter_fields

    async def get_context(
        self, request: Request, q: str, page: int, checker: Any
    ) -> dict[str, Any]:
        """Build template context — delegates to ListContextBuilder."""
        ctx = await self.list_context_builder.build_list_context(
            self.registered, request, q, page, checker
        )
        ctx["view"] = self
        return ctx

    async def html_response(self, request: Request, q: str = "", page: int = 1) -> Response:
        checker = await _resolve_permission_checker(request)
        if checker:
            await checker.load_permissions(self.registered.table_name)
        ctx = await self.get_context(request, q, page, checker)
        return await self.html_renderer.render(request, ctx)

    async def api_response(
        self,
        request: Request,
        page: int = 1,
        per_page: int = 25,
        q: str = "",
        order: str = "",
        after: str | None = None,
        before: str | None = None,
    ) -> Any:
        (
            items,
            total,
            page,
            per_page,
            next_cursor,
            has_next,
            pagination_mode,
        ) = await self.query_provider.get_list(request, q, page, order=order)
        item_list = [self._serialize(item) for item in items]
        return await self.api_renderer.render(
            request,
            {
                "items": item_list,
                "total": total,
                "page": page,
                "per_page": per_page,
                "total_pages": math.ceil(total / per_page) if per_page else 1,
                "next_cursor": next_cursor,
                "has_next": has_next,
            },
        )

get_context(request, q, page, checker) async

Build template context — delegates to ListContextBuilder.

Source code in fastapi_admin_kit/views/class_views.py

async def get_context(
    self, request: Request, q: str, page: int, checker: Any
) -> dict[str, Any]:
    """Build template context — delegates to ListContextBuilder."""
    ctx = await self.list_context_builder.build_list_context(
        self.registered, request, q, page, checker
    )
    ctx["view"] = self
    return ctx

CreateView

fastapi_admin_kit.views.class_views.CreateView

Bases: BaseView

Orchestrates create: parse -> validate -> save -> respond.

Source code in fastapi_admin_kit/views/class_views.py

class CreateView(BaseView):
    """Orchestrates create: parse -> validate -> save -> respond."""

    html_renderer_class = FormHTMLRenderer
    form_parser_class = HTMLFormParser
    api_renderer_class = ItemAPIRenderer

    async def _build_form_context(
        self,
        request: Request,
        obj: Any | None = None,
        values: dict[str, Any] | None = None,
        errors: dict[str, list[str]] | None = None,
        is_create: bool = True,
        checker: Any = None,
        inline_values: dict[str, dict[str, list[str]]] | None = None,
        inline_errors: dict[str, dict[int, dict[str, list[str]]]] | None = None,
    ) -> dict[str, Any]:
        """Build form template context."""
        from fastapi_admin_kit.auth.types import PermissionSet
        from fastapi_admin_kit.form.pipeline import (
            build_form_context as _build_form_ctx,
        )
        from fastapi_admin_kit.form.pipeline import (
            build_inline_formsets,
        )
        from fastapi_admin_kit.views.sidebar import inject_sidebar_context

        # Build inline formsets
        inlines = getattr(self.admin, "inlines", [])
        inline_formsets = await build_inline_formsets(
            self.registered,
            obj=obj,
            inlines=inlines,
            request=request,
            inline_values=inline_values,
            inline_errors=inline_errors,
        )

        ctx = _build_form_ctx(
            self.registered,
            obj=obj,
            values=values,
            errors=errors,
            request=request,
            is_create=is_create,
            inline_formsets=inline_formsets,
        )
        template_context = {
            "form_context": ctx,
            "registered": self.registered,
            "obj": ctx.obj,
            "form_fields": ctx.fieldsets[0].fields if ctx.fieldsets else [],
            "fieldsets": ctx.fieldsets,
            "errors": ctx.errors,
            "is_create": is_create,
            "has_file_field": ctx.has_file_field,
            "permissions": checker.permission_set(self.registered.table_name)
            if checker
            else PermissionSet(can_view=True, can_create=True, can_edit=True, can_delete=True),
            "detail_actions": self.admin.get_detail_actions(),
            "submit_line_actions": self.admin.get_submit_line_actions(),
            "conditional_fields": getattr(self.admin, "conditional_fields", {}),
            "warn_unsaved_form": getattr(self.admin, "warn_unsaved_form", True),
            "compressed_fields": getattr(self.admin, "compressed_fields", True),
            "change_form_show_cancel_button": getattr(
                self.admin, "change_form_show_cancel_button", True
            ),
            "inline_formsets": ctx.inline_formsets,
            "view": self,
        }
        template_context.update(self._get_extra_context(request))
        await inject_sidebar_context(request, template_context)

        template_context = await self.admin.get_form_context(template_context, obj, request)

        return template_context

    async def _create_object(self, request: Request, parsed: dict[str, Any]) -> RedirectResponse:
        """Create object in database."""
        try:
            session = get_db_session(request)
            m2m_data = self.model_saver.extract_m2m(self.registered.model, parsed, request)
            resolved = self.model_saver.resolve_rel_keys(parsed, request)
            resolved = self.admin.prepare_create_data(resolved, request)
            obj = self.registered.model(**resolved)
            self.admin.on_create(obj, request)
            session.add(obj)
            await session.flush()
            await self.model_saver.apply_m2m(obj, m2m_data, request)

            # Save inline objects
            await self.model_saver.save_inline_objects(request, obj)

            self.admin.after_create(obj, request)
            await dispatch_model_change(
                request,
                registered=self.registered,
                event="create",
                obj=obj,
            )
            await flush_pending_perm_ops(request)
            await add_flash(request, "success", f"{self.registered.verbose_name} created.")
        except Exception:
            session = get_db_session(request)
            await session.rollback()
            raise

        # Determine redirect target based on which save button was clicked
        form_data = getattr(request, "_cached_form_data", None)
        save_action = form_data.get("_save_action", "save") if form_data else "save"
        admin_path = request.app.state.admin_config["admin_path"]
        if save_action == "add_another":
            url = f"{admin_path}/{self.registered.table_name}/create"
        else:
            url = f"{admin_path}/{self.registered.table_name}/"
        return RedirectResponse(url=url, status_code=303)

    async def _save_inline_objects(self, request: Request, parent_obj: Any) -> None:
        """Parse and save inline formset objects."""
        from fastapi_admin_kit.inspection import cast_pk_value

        inlines = getattr(self.admin, "inlines", [])
        if not inlines:
            return

        # Use cached form data from form_parser.parse()
        form_data = getattr(request, "_cached_form_data", None)
        if form_data is None:
            form_data = await request.form()
        session = get_db_session(request)

        for inline_cls in inlines:
            inline_instance = inline_cls() if isinstance(inline_cls, type) else inline_cls
            related_model = inline_instance.model
            if related_model is None:
                continue

            prefix = f"{related_model.__tablename__}_set"

            # Get FK field
            fk_field = inline_instance.fk_name
            if not fk_field:
                from sqlalchemy import inspect as sa_inspect

                mapper = sa_inspect(related_model)
                parent_table = self.registered.table_name
                for rel_key, rel_prop in mapper.relationships.items():
                    if rel_prop.direction.name == "MANYTOONE":
                        target_table = rel_prop.mapper.class_.__tablename__
                        if target_table == parent_table:
                            local_cols = [c.key for c in rel_prop.local_columns]
                            if local_cols:
                                fk_field = local_cols[0]
                                break

            if not fk_field:
                continue

            # Parse formset data
            total_forms = int(form_data.get(f"{prefix}-TOTAL_FORMS", "0"))
            initial_forms = int(form_data.get(f"{prefix}-INITIAL_FORMS", "0"))
            deleted_ids: list[str] = []

            # Collect deleted IDs
            for i in range(initial_forms):
                delete_key = f"{prefix}-{i}-DELETE"
                if form_data.get(delete_key) in ("on", "1"):
                    obj_id = form_data.get(f"{prefix}-{i}-id", "")
                    if obj_id:
                        deleted_ids.append(obj_id)

            # Delete marked objects
            for obj_id in deleted_ids:
                try:
                    pk_val = cast_pk_value(related_model, obj_id)
                    existing = await session.get(related_model, pk_val)
                    if existing:
                        await session.delete(existing)
                except Exception:
                    pass

            # Get fields to save — use registered columns if inline has no explicit fields
            from fastapi_admin_kit.registry import AdminRegistry

            related_registry = AdminRegistry()
            related_registered = related_registry.get(related_model.__tablename__)
            related_columns = related_registered.columns if related_registered else None
            fields = inline_instance.get_form_fields(columns=related_columns)
            if not fields:
                continue

            # Create/update objects
            for i in range(total_forms):
                obj_id = form_data.get(f"{prefix}-{i}-id", "")
                delete_key = f"{prefix}-{i}-DELETE"
                if form_data.get(delete_key) in ("on", "1"):
                    continue

                # Build data dict
                data: dict[str, Any] = {}
                for field_name in fields:
                    val = form_data.get(f"{prefix}-{i}-{field_name}")
                    if val is not None:
                        from sqlalchemy import inspect as sa_inspect

                        try:
                            mapper = sa_inspect(related_model)
                            rel = mapper.relationships.get(field_name)
                        except Exception:
                            rel = None

                        if rel is not None:
                            local_cols = [c.key for c in rel.local_columns]
                            fk_col = local_cols[0] if local_cols else None
                            if val and fk_col:
                                casted_val = val
                                try:
                                    casted_val = cast_pk_value(related_model, val)
                                except Exception:
                                    pass
                                data[fk_col] = casted_val
                            elif not val and fk_col:
                                data[fk_col] = None
                        else:
                            related_col = None
                            if related_registered:
                                related_col = next(
                                    (c for c in related_registered.columns if c.name == field_name),
                                    None,
                                )
                            if related_col is not None:
                                from fastapi_admin_kit.inspection import (
                                    cast_value,
                                )

                                val = cast_value(related_col, val)
                            data[field_name] = val

                if obj_id:
                    try:
                        pk_val = cast_pk_value(related_model, obj_id)
                        existing = await session.get(related_model, pk_val)
                        if existing:
                            for k, v in data.items():
                                setattr(existing, k, v)
                    except Exception:
                        pass
                else:
                    parent_pk_field = self.registered.pk_field or "id"
                    data[fk_field] = getattr(parent_obj, parent_pk_field, None)
                    new_obj = related_model(**data)
                    session.add(new_obj)

            await session.flush()

    async def html_response(self, request: Request) -> Response:
        checker = await _resolve_permission_checker(request)
        if checker:
            await checker.load_permissions(self.registered.table_name)

        if request.method == "GET":
            ctx = await self._build_form_context(request, is_create=True, checker=checker)
            return await self.html_renderer.render(request, ctx, is_create=True)

        # POST
        parsed, errors = await self.form_parser.parse(request)

        # Extract perm_data for User model direct permissions
        if self.registered.table_name == "admin_users":
            import json

            form = await request.form()
            perm_data_raw = form.get("perm_data")
            if perm_data_raw:
                try:
                    perm_data = (
                        json.loads(perm_data_raw)
                        if isinstance(perm_data_raw, str)
                        else perm_data_raw
                    )
                    request.state._admin_perm_data = perm_data
                except (json.JSONDecodeError, TypeError):
                    pass

        if errors:
            session = get_db_session(request)
            await session.rollback()
            user = getattr(request.state, "admin_user", None)
            if user is not None:
                await session.refresh(user)
            ctx = await self._build_form_context(
                request,
                values=parsed,
                errors=errors,
                is_create=True,
                checker=checker,
            )
            return await self.html_renderer.render(request, ctx, is_create=True)

        try:
            result = self.admin.validate_create(parsed, request)
        except FieldError as e:
            session = get_db_session(request)
            await session.rollback()
            ctx = await self._build_form_context(
                request,
                values=parsed,
                errors=e.field_errors,
                is_create=True,
                checker=checker,
            )
            return await self.html_renderer.render(request, ctx, is_create=True)
        except ValueError as e:
            session = get_db_session(request)
            await session.rollback()
            ctx = await self._build_form_context(
                request,
                values=parsed,
                errors={"__all__": [str(e)]},
                is_create=True,
                checker=checker,
            )
            return await self.html_renderer.render(request, ctx, is_create=True)

        parsed = result

        parsed = self.admin.process_form_data(parsed, request)

        return await self._create_object(request, parsed)

    async def api_response(self, request: Request) -> Any:
        parser = JSONBodyParser(self.registered)
        parsed, errors = await parser.parse(request)
        if errors:
            raise HTTPException(status_code=422, detail=errors)
        session = get_db_session(request)
        # Validate related model IDs exist
        from fastapi_admin_kit.inspection import validate_related_id

        for rel in self.registered.relationships:
            raw_val = parsed.get(rel.name)
            if raw_val is None:
                continue
            # Skip many-to-many relationships — they are handled separately
            # by extract_m2m/apply_m2m, not by per-ID validation.
            if rel.direction == "MANYTOMANY":
                continue
            # Validate the related ID exists
            pk_value, error = await validate_related_id(
                model=self.registered.model,
                related_model=rel.target_model,
                id_value=raw_val,
                field_name=rel.name,
                request=request,
            )
            if error:
                raise HTTPException(
                    status_code=422,
                    detail=[
                        {
                            "loc": ["body", rel.name],
                            "msg": error["msg"],
                            "type": "value_error",
                            "input": error["input"],
                            "ctx": error["ctx"],
                        }
                    ],
                )
        m2m_data = self.model_saver.extract_m2m(self.registered.model, parsed, request)
        resolved = self.model_saver.resolve_rel_keys(parsed, request)
        resolved = self.admin.prepare_create_data(resolved, request)
        obj = self.registered.model(**resolved)
        self.admin.on_create(obj, request)
        session.add(obj)
        await session.flush()
        await self.model_saver.apply_m2m(obj, m2m_data, request)
        self.admin.after_create(obj, request)
        await dispatch_model_change(
            request,
            registered=self.registered,
            event="create",
            obj=obj,
        )
        await flush_pending_perm_ops(request)
        return await self.api_renderer.render(request, await self._serialize_for_api(obj, request))

EditView

fastapi_admin_kit.views.class_views.EditView

Bases: BaseView

Orchestrates edit: fetch -> parse -> validate -> update -> respond.

Source code in fastapi_admin_kit/views/class_views.py

class EditView(BaseView):
    """Orchestrates edit: fetch -> parse -> validate -> update -> respond."""

    html_renderer_class = FormHTMLRenderer
    form_parser_class = HTMLFormParser
    api_renderer_class = ItemAPIRenderer

    async def _resolve_rel_labels(self, obj: Any, request: Request) -> dict[str, str]:
        """Resolve display labels for relationship fields from FK values."""
        from sqlalchemy import inspect as sa_inspect

        from fastapi_admin_kit.inspection import model_display_name

        labels: dict[str, str] = {}
        if obj is None:
            return labels
        try:
            mapper = sa_inspect(type(obj))
        except Exception:
            return labels
        session = get_db_session(request)
        for rel_key, rel_prop in mapper.relationships.items():
            local_cols = [c.key for c in rel_prop.local_columns]
            if not local_cols:
                continue
            fk_val = getattr(obj, local_cols[0], None)
            if fk_val is None:
                continue
            target_cls = rel_prop.mapper.class_
            try:
                target = await session.get(target_cls, fk_val)
                if target is not None:
                    labels[rel_key] = model_display_name(target)
            except Exception:
                labels[rel_key] = str(fk_val)
        return labels

    async def _build_form_context(
        self,
        request: Request,
        obj: Any | None = None,
        values: dict[str, Any] | None = None,
        errors: dict[str, list[str]] | None = None,
        is_create: bool = False,
        checker: Any = None,
        rel_labels: dict[str, str] | None = None,
        inline_values: dict[str, dict[str, list[str]]] | None = None,
        inline_errors: dict[str, dict[int, dict[str, list[str]]]] | None = None,
    ) -> dict[str, Any]:
        """Build form template context."""
        from fastapi_admin_kit.auth.types import PermissionSet
        from fastapi_admin_kit.form.pipeline import (
            build_form_context as _build_form_ctx,
        )
        from fastapi_admin_kit.form.pipeline import (
            build_inline_formsets,
        )
        from fastapi_admin_kit.views.sidebar import inject_sidebar_context

        # Build inline formsets
        inlines = getattr(self.admin, "inlines", [])
        inline_formsets = await build_inline_formsets(
            self.registered,
            obj=obj,
            inlines=inlines,
            request=request,
            inline_values=inline_values,
            inline_errors=inline_errors,
        )

        ctx = _build_form_ctx(
            self.registered,
            obj=obj,
            values=values,
            errors=errors,
            request=request,
            is_create=is_create,
            rel_labels=rel_labels,
            inline_formsets=inline_formsets,
        )
        template_context = {
            "form_context": ctx,
            "registered": self.registered,
            "obj": ctx.obj,
            "form_fields": ctx.fieldsets[0].fields if ctx.fieldsets else [],
            "fieldsets": ctx.fieldsets,
            "errors": ctx.errors,
            "is_create": is_create,
            "has_file_field": ctx.has_file_field,
            "permissions": checker.permission_set(self.registered.table_name)
            if checker
            else PermissionSet(can_view=True, can_create=True, can_edit=True, can_delete=True),
            "detail_actions": self.admin.get_detail_actions(),
            "submit_line_actions": self.admin.get_submit_line_actions(),
            "conditional_fields": getattr(self.admin, "conditional_fields", {}),
            "warn_unsaved_form": getattr(self.admin, "warn_unsaved_form", True),
            "compressed_fields": getattr(self.admin, "compressed_fields", True),
            "change_form_show_cancel_button": getattr(
                self.admin, "change_form_show_cancel_button", True
            ),
            "inline_formsets": ctx.inline_formsets,
            "view": self,
        }
        template_context.update(self._get_extra_context(request))
        await inject_sidebar_context(request, template_context)

        template_context = await self.admin.get_form_context(template_context, obj, request)

        return template_context

    def _apply_parsed(self, obj: Any, parsed: dict[str, Any]) -> None:
        """Apply parsed form/JSON data to an ORM object.

        Relationship fields (e.g. ``"user"``) are resolved to their
        local foreign-key column (e.g. ``"user_id"``) so that the
        correct column is persisted by SQLAlchemy.
        """
        from sqlalchemy import inspect as sa_inspect

        col_names = {c.name for c in self.registered.columns}

        # Build mapping: relationship key -> local FK column key
        rel_fk_map: dict[str, str] = {}
        try:
            mapper = sa_inspect(type(obj))
        except Exception:
            mapper = None
        if mapper is not None:
            for rel_key, rel_prop in mapper.relationships.items():
                if rel_prop.direction.name == "MANYTOMANY":
                    continue
                local_cols = [c.key for c in rel_prop.local_columns]
                if local_cols:
                    rel_fk_map[rel_key] = local_cols[0]

        for key, value in parsed.items():
            if key in col_names:
                setattr(obj, key, value)
            elif key in rel_fk_map:
                setattr(obj, rel_fk_map[key], value)

    async def _update_object(
        self, request: Request, obj: Any, parsed: dict[str, Any]
    ) -> RedirectResponse:
        """Update object in database."""
        try:
            parsed = self.admin.prepare_update_data(parsed, request)
            m2m_data = self.model_saver.extract_m2m(obj, parsed, request)
            self.model_saver.apply_parsed(obj, parsed, request)
            session = get_db_session(request)
            await self.model_saver.apply_m2m(obj, m2m_data, request)
            self.admin.on_update(obj, parsed, request)
            await session.flush()

            # Save inline objects
            await self.model_saver.save_inline_objects(request, obj)

            self.admin.after_update(obj, request)
            await dispatch_model_change(
                request,
                registered=self.registered,
                event="update",
                obj=obj,
            )
            await flush_pending_perm_ops(request)
            await add_flash(request, "success", f"{self.registered.verbose_name} updated.")
        except Exception:
            session = get_db_session(request)
            await session.rollback()
            raise

        # Determine redirect target based on which save button was clicked
        form_data = getattr(request, "_cached_form_data", None)
        save_action = form_data.get("_save_action", "save") if form_data else "save"
        admin_path = request.app.state.admin_config["admin_path"]
        if save_action == "add_another":
            url = f"{admin_path}/{self.registered.table_name}/create"
        else:
            url = f"{admin_path}/{self.registered.table_name}/"
        return RedirectResponse(url=url, status_code=303)

    async def _save_inline_objects(self, request: Request, parent_obj: Any) -> None:
        """Parse and save inline formset objects."""
        from fastapi_admin_kit.inspection import cast_pk_value

        inlines = getattr(self.admin, "inlines", [])
        if not inlines:
            return

        # Use cached form data from form_parser.parse()
        form_data = getattr(request, "_cached_form_data", None)
        if form_data is None:
            form_data = await request.form()
        session = get_db_session(request)

        for inline_cls in inlines:
            inline_instance = inline_cls() if isinstance(inline_cls, type) else inline_cls
            related_model = inline_instance.model
            if related_model is None:
                continue

            prefix = f"{related_model.__tablename__}_set"

            # Get FK field
            fk_field = inline_instance.fk_name
            if not fk_field:
                from sqlalchemy import inspect as sa_inspect

                mapper = sa_inspect(related_model)
                parent_table = self.registered.table_name
                for rel_key, rel_prop in mapper.relationships.items():
                    if rel_prop.direction.name == "MANYTOONE":
                        target_table = rel_prop.mapper.class_.__tablename__
                        if target_table == parent_table:
                            local_cols = [c.key for c in rel_prop.local_columns]
                            if local_cols:
                                fk_field = local_cols[0]
                                break

            if not fk_field:
                continue

            # Parse formset data
            total_forms = int(form_data.get(f"{prefix}-TOTAL_FORMS", "0"))
            initial_forms = int(form_data.get(f"{prefix}-INITIAL_FORMS", "0"))
            deleted_ids: list[str] = []

            # Collect deleted IDs
            for i in range(initial_forms):
                delete_key = f"{prefix}-{i}-DELETE"
                if form_data.get(delete_key) in ("on", "1"):
                    obj_id = form_data.get(f"{prefix}-{i}-id", "")
                    if obj_id:
                        deleted_ids.append(obj_id)

            # Delete marked objects
            for obj_id in deleted_ids:
                try:
                    pk_val = cast_pk_value(related_model, obj_id)
                    existing = await session.get(related_model, pk_val)
                    if existing:
                        await session.delete(existing)
                except Exception:
                    pass

            # Get fields to save — use registered columns if inline has no explicit fields
            from fastapi_admin_kit.registry import AdminRegistry

            related_registry = AdminRegistry()
            related_registered = related_registry.get(related_model.__tablename__)
            related_columns = related_registered.columns if related_registered else None
            fields = inline_instance.get_form_fields(columns=related_columns)
            if not fields:
                continue

            # Create/update objects
            for i in range(total_forms):
                obj_id = form_data.get(f"{prefix}-{i}-id", "")
                delete_key = f"{prefix}-{i}-DELETE"
                if form_data.get(delete_key) in ("on", "1"):
                    continue

                # Build data dict
                data: dict[str, Any] = {}
                for field_name in fields:
                    val = form_data.get(f"{prefix}-{i}-{field_name}")
                    if val is not None:
                        from sqlalchemy import inspect as sa_inspect

                        try:
                            mapper = sa_inspect(related_model)
                            rel = mapper.relationships.get(field_name)
                        except Exception:
                            rel = None

                        if rel is not None:
                            local_cols = [c.key for c in rel.local_columns]
                            fk_col = local_cols[0] if local_cols else None
                            if val and fk_col:
                                casted_val = val
                                try:
                                    casted_val = cast_pk_value(related_model, val)
                                except Exception:
                                    pass
                                data[fk_col] = casted_val
                            elif not val and fk_col:
                                data[fk_col] = None
                        else:
                            related_col = None
                            if related_registered:
                                related_col = next(
                                    (c for c in related_registered.columns if c.name == field_name),
                                    None,
                                )
                            if related_col is not None:
                                from fastapi_admin_kit.inspection import (
                                    cast_value,
                                )

                                val = cast_value(related_col, val)
                            data[field_name] = val

                if obj_id:
                    try:
                        pk_val = cast_pk_value(related_model, obj_id)
                        existing = await session.get(related_model, pk_val)
                        if existing:
                            for k, v in data.items():
                                setattr(existing, k, v)
                    except Exception:
                        pass
                else:
                    parent_pk_field = self.registered.pk_field or "id"
                    data[fk_field] = getattr(parent_obj, parent_pk_field, None)
                    new_obj = related_model(**data)
                    session.add(new_obj)

            await session.flush()

    async def _build_detail_context(
        self,
        request: Request,
        obj: Any,
        checker: Any = None,
    ) -> dict[str, Any]:
        """Build read-only detail view context with all fields."""
        from fastapi_admin_kit.auth.types import PermissionSet
        from fastapi_admin_kit.form.pipeline import (
            build_form_context as _build_form_ctx,
        )
        from fastapi_admin_kit.views.sidebar import inject_sidebar_context

        rel_labels = await self._resolve_rel_labels(obj, request)
        ctx = _build_form_ctx(
            self.registered,
            obj=obj,
            request=request,
            is_create=False,
            rel_labels=rel_labels,
        )
        template_context = {
            "form_context": ctx,
            "registered": self.registered,
            "obj": obj,
            "form_fields": ctx.fieldsets[0].fields if ctx.fieldsets else [],
            "fieldsets": ctx.fieldsets,
            "is_create": False,
            "has_file_field": ctx.has_file_field,
            "permissions": checker.permission_set(self.registered.table_name)
            if checker
            else PermissionSet(can_view=True, can_create=True, can_edit=True, can_delete=True),
            "view": self,
        }
        template_context.update(self._get_extra_context(request))
        await inject_sidebar_context(request, template_context)
        return template_context

    async def _preload_m2m_relationships(self, obj: Any, request: Request) -> None:
        """Preload many-to-many relationship collections on obj to avoid
        MissingGreenlet in async context.
        """
        from sqlalchemy import inspect as sa_inspect

        if obj is None:
            return
        try:
            mapper = sa_inspect(type(obj))
            session = get_db_session(request)
            for rel_key, rel_prop in mapper.relationships.items():
                if rel_prop.direction.name == "MANYTOMANY":
                    await session.refresh(obj, [rel_key])
        except Exception:
            pass

    async def html_response(self, request: Request, id: Any = None) -> Response:
        obj = await self.query_provider.get_object(request, id)
        if not obj:
            raise HTTPException(status_code=404, detail="Not found")

        checker = await _resolve_permission_checker(request)
        if checker:
            await checker.load_permissions(self.registered.table_name)

        perms = checker.permission_set(self.registered.table_name) if checker else None

        if obj:
            await self._preload_m2m_relationships(obj, request)

        if request.method == "GET":
            if perms and not perms.can_edit and perms.can_view:
                ctx = await self._build_detail_context(request, obj, checker)
                # Custom detail template: explicit → auto-discovery → global → default
                from fastapi_admin_kit.views.renderers import resolve_template

                candidates = []
                if self.detail_template:
                    candidates.append(self.detail_template)
                candidates.append(f"admin/{self.registered.table_name}/detail.html")
                candidates += ["admin/detail.html", "pages/detail.html"]
                detail_template = resolve_template(request, candidates)
                return request.app.state.admin_jinja_env.TemplateResponse(
                    request, detail_template, ctx
                )
            rel_labels = await self._resolve_rel_labels(obj, request)
            ctx = await self._build_form_context(
                request,
                obj=obj,
                is_create=False,
                checker=checker,
                rel_labels=rel_labels,
            )
            return await self.html_renderer.render(request, ctx, is_create=False)

        # POST
        parsed, errors = await self.form_parser.parse(request, obj=obj)

        # Extract perm_data for User model direct permissions
        if self.registered.table_name == "admin_users":
            import json

            form = await request.form()
            perm_data_raw = form.get("perm_data")
            if perm_data_raw:
                try:
                    perm_data = (
                        json.loads(perm_data_raw)
                        if isinstance(perm_data_raw, str)
                        else perm_data_raw
                    )
                    request.state._admin_perm_data = perm_data
                except (json.JSONDecodeError, TypeError):
                    pass

        if errors:
            session = get_db_session(request)
            await session.rollback()
            await session.refresh(obj)
            user = getattr(request.state, "admin_user", None)
            if user is not None:
                await session.refresh(user)
            rel_labels = await self._resolve_rel_labels(obj, request)
            ctx = await self._build_form_context(
                request,
                obj=obj,
                values=parsed,
                errors=errors,
                checker=checker,
                rel_labels=rel_labels,
            )
            return await self.html_renderer.render(request, ctx, is_create=False)

        try:
            result = self.admin.validate_update(obj, parsed, request)
        except FieldError as e:
            session = get_db_session(request)
            await session.rollback()
            await session.refresh(obj)
            rel_labels = await self._resolve_rel_labels(obj, request)
            ctx = await self._build_form_context(
                request,
                obj=obj,
                values=parsed,
                errors=e.field_errors,
                checker=checker,
                rel_labels=rel_labels,
            )
            return await self.html_renderer.render(request, ctx, is_create=False)
        except ValueError as e:
            session = get_db_session(request)
            await session.rollback()
            await session.refresh(obj)
            rel_labels = await self._resolve_rel_labels(obj, request)
            ctx = await self._build_form_context(
                request,
                obj=obj,
                values=parsed,
                errors={"__all__": [str(e)]},
                checker=checker,
                rel_labels=rel_labels,
            )
            return await self.html_renderer.render(request, ctx)

        parsed = result

        parsed = self.admin.process_form_data(parsed, request)

        return await self._update_object(request, obj, parsed)

    async def api_response(
        self,
        request: Request,
        id: Any = None,
        item_id: Any = None,
    ) -> Any:
        pk = id or item_id
        obj = await self.query_provider.get_object(request, pk)
        if not obj:
            raise HTTPException(status_code=404, detail="Not found")

        if request.method == "GET":
            return self._serialize(obj)

        # PUT
        parser = JSONBodyParser(self.registered)
        parsed, errors = await parser.parse(request, obj)
        if errors:
            raise HTTPException(status_code=422, detail=errors)
        # Validate related model IDs exist
        from fastapi_admin_kit.inspection import validate_related_id

        for rel in self.registered.relationships:
            raw_val = parsed.get(rel.name)
            if raw_val is None:
                continue
            # Skip many-to-many relationships — they are handled separately
            # by extract_m2m/apply_m2m, not by per-ID validation.
            if rel.direction == "MANYTOMANY":
                continue
            # Validate the related ID exists
            pk_value, error = await validate_related_id(
                model=self.registered.model,
                related_model=rel.target_model,
                id_value=raw_val,
                field_name=rel.name,
                request=request,
            )
            if error:
                raise HTTPException(
                    status_code=422,
                    detail=[
                        {
                            "loc": ["body", rel.name],
                            "msg": error["msg"],
                            "type": "value_error",
                            "input": error["input"],
                            "ctx": error["ctx"],
                        }
                    ],
                )
        try:
            parsed = self.admin.prepare_update_data(parsed, request)
            m2m_data = self.model_saver.extract_m2m(obj, parsed, request)
            self.model_saver.apply_parsed(obj, parsed, request)
            session = get_db_session(request)
            await self.model_saver.apply_m2m(obj, m2m_data, request)
            self.admin.on_update(obj, parsed, request)
            await session.flush()
            self.admin.after_update(obj, request)
            await dispatch_model_change(
                request,
                registered=self.registered,
                event="update",
                obj=obj,
            )
            await flush_pending_perm_ops(request)
        except Exception:
            session = get_db_session(request)
            await session.rollback()
            raise
        return await self._serialize_for_api(obj, request)

DeleteView

fastapi_admin_kit.views.class_views.DeleteView

Bases: BaseView

Orchestrates delete: fetch -> delete -> respond.

Source code in fastapi_admin_kit/views/class_views.py

class DeleteView(BaseView):
    """Orchestrates delete: fetch -> delete -> respond."""

    async def html_response(self, request: Request, id: Any = None) -> Response:
        obj = await self.query_provider.get_object(request, id)
        if not obj:
            raise HTTPException(status_code=404, detail="Not found")
        try:
            self.admin.on_delete(obj, request)
            session = get_db_session(request)
            await session.delete(obj)
            await session.flush()
            self.admin.after_delete(obj, request)
            await dispatch_model_change(
                request,
                registered=self.registered,
                event="delete",
                obj=obj,
            )
            await add_flash(request, "success", f"{self.registered.verbose_name} deleted.")
        except Exception:
            session = get_db_session(request)
            await session.rollback()
            raise
        url = f"{request.app.state.admin_config['admin_path']}/{self.registered.table_name}/"
        return RedirectResponse(url=url, status_code=303)

    async def api_response(
        self,
        request: Request,
        id: Any = None,
        item_id: Any = None,
    ) -> Response:
        pk = id or item_id
        obj = await self.query_provider.get_object(request, pk)
        if not obj:
            raise HTTPException(status_code=404, detail="Not found")
        self.admin.on_delete(obj, request)
        session = get_db_session(request)
        await session.delete(obj)
        await session.flush()
        self.admin.after_delete(obj, request)
        await dispatch_model_change(
            request,
            registered=self.registered,
            event="delete",
            obj=obj,
        )
        return Response(status_code=204)

BulkView

fastapi_admin_kit.views.class_views.BulkView

Bases: BaseView

Orchestrates bulk actions on multiple objects.

Source code in fastapi_admin_kit/views/class_views.py

class BulkView(BaseView):
    """Orchestrates bulk actions on multiple objects."""

    html_renderer_class = ListHTMLRenderer

    async def html_response(self, request: Request) -> Response:
        session = get_db_session(request)
        form = await request.form()
        action = form.get("action", "")
        ids = form.getlist("ids[]")

        is_htmx = request.headers.get("HX-Request") == "true"

        if not ids:
            if is_htmx:
                list_view = ListView(self.registered)
                checker = await _resolve_permission_checker(request)
                ctx = await list_view.get_context(request, "", 1, checker)
                return await self.html_renderer.render(request, ctx)
            url = f"{request.app.state.admin_config['admin_path']}/{self.registered.table_name}/"
            return RedirectResponse(url=url, status_code=303)

        if action == "delete_selected":
            for pid in ids:
                obj = await session.get(self.registered.model, pid)
                if obj:
                    self.admin.on_delete(obj, request)
                    await dispatch_model_change(
                        request,
                        registered=self.registered,
                        event="delete",
                        obj=obj,
                    )
                    await session.delete(obj)
            await session.flush()
        else:
            action_obj = None
            for a in self.admin.get_list_actions():
                if a.name == action:
                    action_obj = a
                    break

            if action_obj:
                objects = []
                for pid in ids:
                    obj = await session.get(self.registered.model, pid)
                    if obj:
                        objects.append(obj)
                if objects:
                    await action_obj.execute(objects, request)
                await session.flush()
            else:
                action_fn = getattr(self.admin, f"action_{action}", None)
                if not action_fn:
                    raise HTTPException(status_code=400, detail=f"Unknown action: {action}")
                for pid in ids:
                    obj = await session.get(self.registered.model, pid)
                    if obj:
                        action_fn(obj)
                await session.flush()

        if is_htmx:
            list_view = ListView(self.registered)
            checker = await _resolve_permission_checker(request)
            ctx = await list_view.get_context(request, "", 1, checker)
            return await self.html_renderer.render(request, ctx)

        url = f"{request.app.state.admin_config['admin_path']}/{self.registered.table_name}/"
        return RedirectResponse(url=url, status_code=303)

    async def api_response(self, request: Request) -> Any:
        from fastapi.responses import JSONResponse

        session = get_db_session(request)
        content_type = request.headers.get("content-type", "")
        is_json = content_type.startswith("application/json")
        body = await request.json() if is_json else {}
        action = body.get("action", "")
        ids = body.get("ids", [])

        if action == "delete_selected":
            deleted = 0
            for pid in ids:
                obj = await session.get(self.registered.model, pid)
                if obj:
                    self.admin.on_delete(obj, request)
                    await dispatch_model_change(
                        request,
                        registered=self.registered,
                        event="delete",
                        obj=obj,
                    )
                    await session.delete(obj)
                    deleted += 1
            await session.flush()
            return JSONResponse({"deleted": deleted})

        action_obj = None
        for a in self.admin.get_list_actions():
            if a.name == action:
                action_obj = a
                break

        if action_obj:
            objects = []
            for pid in ids:
                obj = await session.get(self.registered.model, pid)
                if obj:
                    objects.append(obj)
            if objects:
                await action_obj.execute(objects, request)
            await session.flush()
            return JSONResponse({"executed": len(objects)})

        action_fn = getattr(self.admin, f"action_{action}", None)
        if not action_fn:
            raise HTTPException(status_code=400, detail=f"Unknown action: {action}")

        executed = 0
        for pid in ids:
            obj = await session.get(self.registered.model, pid)
            if obj:
                action_fn(obj)
                executed += 1
        await session.flush()
        return JSONResponse({"executed": executed})

Schema Generation

fastapi_admin_kit.api.schema_generator

Dynamic Pydantic schema generation from SQLAlchemy models.

build_create_schema(registered)

Build a Pydantic model for create requests.

Excludes PK, readonly fields, and server-default-only fields.

Source code in fastapi_admin_kit/api/schema_generator.py

def build_create_schema(registered: Any) -> type[BaseModel]:
    """Build a Pydantic model for create requests.

    Excludes PK, readonly fields, and server-default-only fields.
    """
    admin = registered.admin
    readonly = set(admin.readonly_fields or [])
    columns = _collect_fields(registered, exclude_pk=True)

    fields: dict[str, Any] = {}
    for col in columns:
        if col.name in readonly:
            continue
        if col.server_default is not None and col.default is None:
            continue

        python_type = _get_column_python_type(col)
        if col.nullable:
            field_info = (python_type | None, Field(default=None))
        else:
            field_info = (python_type, Field(...))

        fields[col.name] = field_info

    for rel in _collect_relationships(registered):
        if rel.name in readonly or rel.name in fields:
            continue
        python_type = _relationship_python_type(rel)
        fields[rel.name] = (python_type | None, Field(default=None))

    _add_extra_fields(admin, fields, required_on_create=True)

    model_name = f"{registered.verbose_name.replace(' ', '')}Create"
    return create_model(model_name, __config__=None, **fields)

build_update_schema(registered)

Build a Pydantic model for update requests.

All fields optional. Excludes PK and readonly fields.

Source code in fastapi_admin_kit/api/schema_generator.py

def build_update_schema(registered: Any) -> type[BaseModel]:
    """Build a Pydantic model for update requests.

    All fields optional. Excludes PK and readonly fields.
    """
    admin = registered.admin
    readonly = set(admin.readonly_fields or [])
    columns = _collect_fields(registered, exclude_pk=True)

    fields: dict[str, Any] = {}
    for col in columns:
        if col.name in readonly:
            continue

        python_type = _get_column_python_type(col)
        field_info = (python_type | None, Field(default=None))
        fields[col.name] = field_info

    for rel in _collect_relationships(registered):
        if rel.name in readonly or rel.name in fields:
            continue
        python_type = _relationship_python_type(rel)
        fields[rel.name] = (python_type | None, Field(default=None))

    _add_extra_fields(admin, fields, required_on_create=False)

    model_name = f"{registered.verbose_name.replace(' ', '')}Update"
    return create_model(model_name, __config__=None, **fields)

build_response_schema(registered)

Build a Pydantic model for response output.

Source code in fastapi_admin_kit/api/schema_generator.py

def build_response_schema(registered: Any) -> type[BaseModel]:
    """Build a Pydantic model for response output."""
    from fastapi_admin_kit.inspection.types import SENSITIVE_FIELDS

    columns = [c for c in registered.columns if c.name not in SENSITIVE_FIELDS]

    fields: dict[str, Any] = {}
    for col in columns:
        python_type = _get_column_python_type(col)
        if col.nullable:
            field_info = (python_type | None, Field(default=None))
        else:
            field_info = (python_type, Field(...))
        fields[col.name] = field_info

    model_name = f"{registered.verbose_name.replace(' ', '')}Response"
    return create_model(model_name, __config__=None, **fields)

build_list_response_schema(registered)

Build a paginated list response schema wrapping the response schema.

Source code in fastapi_admin_kit/api/schema_generator.py

def build_list_response_schema(registered: Any) -> type[BaseModel]:
    """Build a paginated list response schema wrapping the response schema."""
    item_schema = build_response_schema(registered)

    model_name = f"{registered.verbose_name.replace(' ', '')}ListResponse"
    return create_model(
        model_name,
        items=(list[item_schema], Field(...)),
        total=(int, Field(...)),
        page=(int | None, Field(default=None)),
        per_page=(int, Field(...)),
        total_pages=(int | None, Field(default=None)),
        next_cursor=(str | None, Field(default=None)),
        has_next=(bool, Field(default=False)),
    )

get_or_build_schemas(registered)

Get or generate and cache schemas for a registered model.

Source code in fastapi_admin_kit/api/schema_generator.py

def get_or_build_schemas(registered: Any) -> dict[str, type[BaseModel]]:
    """Get or generate and cache schemas for a registered model."""
    if hasattr(registered, "_schemas") and registered._schemas is not None:
        return registered._schemas

    schemas = {
        "create": build_create_schema(registered),
        "update": build_update_schema(registered),
        "response": build_response_schema(registered),
        "list_response": build_list_response_schema(registered),
    }
    registered._schemas = schemas
    return schemas