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: Engine | 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,
        # 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 = None,
        session_cookie_name: str = "admin_session",
        session_secure: bool = False,
        session_samesite: str = "strict",
        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,
    ):
        self.registry = AdminRegistry()
        self._app: FastAPI | None = app

        # 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
        else:
            self._csrf_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,
                ),
                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,
                ),
            )

        if database is None:
            database = AdminDatabase(engine=engine, base=base, database_config=database_config)

        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,
            )

        self.config = config
        self.database = database
        self.router = router
        self.template = template

        # 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] = []

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

        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) -> Engine | 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

        # 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 per-request session middleware
        if 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:
                pass

        # Add audit context middleware
        if 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:
                pass

        # 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
        skip_create_tables = os.environ.get("SKIP_CREATE_TABLES", "false").lower() == "true"
        if not skip_create_tables:
            await self.database._create_tables()

        # 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)

        # 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()

        # 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
        default_skip = {"RefreshToken", "UserPermission", "UserTOTP"}
        all_skip = default_skip | skip_models
        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)
        from fastapi_admin_kit.audit.listener import attach_audit_listener

        engine = self.database.engine
        if engine is not None:
            from sqlalchemy.ext.asyncio import AsyncEngine

            if isinstance(engine, AsyncEngine):
                from fastapi_admin_kit.db import create_session_factory

                session_factory = create_session_factory(engine)
                attach_audit_listener(session_factory, self.registry)

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

        # 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)

    # ------------------------------------------------------------------
    # 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 _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,
            "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
        engine = self.database.engine
        if engine is not None:
            from sqlalchemy.ext.asyncio import AsyncEngine

            if isinstance(engine, AsyncEngine):
                from fastapi_admin_kit.db import create_session_factory

                session_factory = create_session_factory(engine)
                # Legacy fallback — a single session for backward compat
                db_session = session_factory()
            else:
                from sqlalchemy.orm import sessionmaker as sync_sessionmaker

                session_factory = sync_sessionmaker(bind=engine, expire_on_commit=False)
                db_session = session_factory()

        # Inject auth_model into the backend if provided
        if self.config.auth.auth_backend is not None and self.config.auth.auth_model is not None:
            self.config.auth.auth_backend._auth_model = self.config.auth.auth_model

        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,
        )

        # 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

        # 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."""
        from starlette.templating import Jinja2Templates

        templates_dir = Path(__file__).parent.parent / "templates"
        self._jinja_env = Jinja2Templates(directory=str(templates_dir))

        # Disable autoescape — templates are server-controlled, no user XSS risk
        self._jinja_env.env.autoescape = False

        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
        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["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-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",
        }

        def _icon(name: str, size: str = "", **kwargs) -> str:
            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 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,
        }
        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/tokens.css",
            "css/presets.css",
            "css/admin.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 _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
            model_router = build_model_router(registered)
            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"],
        )

        # 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,
            PermissionAdmin,
            RoleAdmin,
            UserAdmin,
        )
        from fastapi_admin_kit.audit.models import AuditLog
        from fastapi_admin_kit.auth.models import (
            LoginAttempt,
            Permission,
            Role,
            User,
        )

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

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

    # ------------------------------------------------------------------
    # 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,
        )

    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 self.template.build_sidebar_context(
            request, user=user, permissions_map=permissions_map
        )

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

    def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
        """Inject nav_groups + permissions_map into a template context dict."""
        return 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

    # 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 per-request session middleware
    if 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:
            pass

    # Add audit context middleware
    if 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:
            pass

    # 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
    skip_create_tables = os.environ.get("SKIP_CREATE_TABLES", "false").lower() == "true"
    if not skip_create_tables:
        await self.database._create_tables()

    # 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)

    # 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()

    # 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
    default_skip = {"RefreshToken", "UserPermission", "UserTOTP"}
    all_skip = default_skip | skip_models
    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)
    from fastapi_admin_kit.audit.listener import attach_audit_listener

    engine = self.database.engine
    if engine is not None:
        from sqlalchemy.ext.asyncio import AsyncEngine

        if isinstance(engine, AsyncEngine):
            from fastapi_admin_kit.db import create_session_factory

            session_factory = create_session_factory(engine)
            attach_audit_listener(session_factory, self.registry)

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

    # 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)

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)

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

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

Source code in fastapi_admin_kit/admin/core.py

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 self.template.build_sidebar_context(
        request, user=user, permissions_map=permissions_map
    )

sidebar_template_kwargs(request)

Thin wrapper — returns sidebar kwargs for TemplateResponse contexts.

Source code in fastapi_admin_kit/admin/core.py

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

apply_sidebar_context(request, user, context)

Inject nav_groups + permissions_map into a template context dict.

Source code in fastapi_admin_kit/admin/core.py

def apply_sidebar_context(self, request: Any, user: Any, context: dict) -> dict:
    """Inject nav_groups + permissions_map into a template context dict."""
    return 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,
    ):
        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()

    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()

    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()

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)

        # Inspect using the injected inspector
        columns, relationships = self._inspector.inspect_model(model)

        admin = admin_class() if admin_class else ModelAdmin()
        table_name = model.__tablename__
        if admin.verbose_name:
            verbose_name = admin.verbose_name
        else:
            class_name = getattr(model, "__name__", None)
            if class_name and not class_name.startswith("_"):
                verbose_name = re.sub(r"([A-Z])", r" \1", class_name).strip().title()
            else:
                verbose_name = table_name.replace("_", " ").title()
        if admin.verbose_name_plural:
            verbose_name_plural = admin.verbose_name_plural
        elif (
            verbose_name.endswith("y")
            and len(verbose_name) > 1
            and verbose_name[-2].lower() not in "aeiou"
        ):
            verbose_name_plural = f"{verbose_name[:-1]}ies"
        else:
            verbose_name_plural = f"{verbose_name}s"

        registered = RegisteredModel(
            model=model,
            admin=admin,
            table_name=table_name,
            verbose_name=verbose_name,
            verbose_name_plural=verbose_name_plural,
            columns=columns,
            relationships=relationships,
        )

        self._models[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) -> list[RegisteredModel]:
        """Scan all subclasses of DeclarativeBase and register unregistered ones.

        Also discovers SQLModel subclasses if SQLModel is installed.

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

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

        # Discover SQLAlchemy DeclarativeBase subclasses
        for subclass in _all_declarative_subclasses(DeclarativeBase):
            if hasattr(subclass, "registry"):
                for mapper in subclass.registry.mappers:
                    cls = mapper.class_
                    if cls not in seen:
                        seen.add(cls)
                        if hasattr(cls, "__tablename__") and cls.__tablename__ not in self._models:
                            discovered.append(self.register(cls))

        # 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:
                        cls = mapper.class_
                        if cls not in seen:
                            seen.add(cls)
                            if (
                                hasattr(cls, "__tablename__")
                                and cls.__tablename__ not in self._models
                            ):
                                discovered.append(self.register(cls))
        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)

    # Inspect using the injected inspector
    columns, relationships = self._inspector.inspect_model(model)

    admin = admin_class() if admin_class else ModelAdmin()
    table_name = model.__tablename__
    if admin.verbose_name:
        verbose_name = admin.verbose_name
    else:
        class_name = getattr(model, "__name__", None)
        if class_name and not class_name.startswith("_"):
            verbose_name = re.sub(r"([A-Z])", r" \1", class_name).strip().title()
        else:
            verbose_name = table_name.replace("_", " ").title()
    if admin.verbose_name_plural:
        verbose_name_plural = admin.verbose_name_plural
    elif (
        verbose_name.endswith("y")
        and len(verbose_name) > 1
        and verbose_name[-2].lower() not in "aeiou"
    ):
        verbose_name_plural = f"{verbose_name[:-1]}ies"
    else:
        verbose_name_plural = f"{verbose_name}s"

    registered = RegisteredModel(
        model=model,
        admin=admin,
        table_name=table_name,
        verbose_name=verbose_name,
        verbose_name_plural=verbose_name_plural,
        columns=columns,
        relationships=relationships,
    )

    self._models[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()

Scan all subclasses of DeclarativeBase and register unregistered ones.

Also discovers SQLModel subclasses if SQLModel is installed.

Returns:

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

Source code in fastapi_admin_kit/registry/core.py

def auto_discover(self) -> list[RegisteredModel]:
    """Scan all subclasses of DeclarativeBase and register unregistered ones.

    Also discovers SQLModel subclasses if SQLModel is installed.

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

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

    # Discover SQLAlchemy DeclarativeBase subclasses
    for subclass in _all_declarative_subclasses(DeclarativeBase):
        if hasattr(subclass, "registry"):
            for mapper in subclass.registry.mappers:
                cls = mapper.class_
                if cls not in seen:
                    seen.add(cls)
                    if hasattr(cls, "__tablename__") and cls.__tablename__ not in self._models:
                        discovered.append(self.register(cls))

    # 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:
                    cls = mapper.class_
                    if cls not in seen:
                        seen.add(cls)
                        if (
                            hasattr(cls, "__tablename__")
                            and cls.__tablename__ not in self._models
                        ):
                            discovered.append(self.register(cls))
    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:
            valid = {c.name for c in self.columns}
            return [f for f in self.admin.list_display if f in valid]
        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] | 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

    # 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

    # 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}

    # 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

    # Route generation
    skip_auto_routes: bool = False

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

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

    # 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)."""
        return session.query(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 ──────────────────────────────────────────

    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

    # ── 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

__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).

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)."""
    return session.query(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)

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

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

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

        if self.connect_args:
            kwargs["connect_args"] = self.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

    if self.connect_args:
        kwargs["connect_args"] = self.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

ThemeConfig

fastapi_admin_kit.config.ThemeConfig

Complete theme configuration — maps to CSS custom properties.

When preset is set, its defaults are used. Any explicit attribute override takes precedence over the preset defaults.

Source code in fastapi_admin_kit/config/theme.py

class ThemeConfig:
    """Complete theme configuration — maps to CSS custom properties.

    When preset is set, its defaults are used. Any explicit attribute
    override takes precedence over the preset defaults.
    """

    def __init__(
        self,
        preset: str = "editorial",
        *,
        primary_color: str | None = None,
        surface_base: str | None = None,
        surface_raised: str | None = None,
        text_primary: str | None = None,
        text_secondary: str | None = None,
        border_color: str | None = None,
        font_display: str | None = None,
        font_body: str | None = None,
        font_mono: str | None = None,
        font_import_url: str | None = None,
        radius_sm: str | None = None,
        radius_md: str | None = None,
        radius_lg: str | None = None,
        shadow_sm: str | None = None,
        shadow_md: str | None = None,
        shadow_lg: str | None = None,
        topbar_height: str = "56px",
        sidebar_width: str = "248px",
        sidebar_collapsed_width: str = "60px",
        content_max_width: str = "1360px",
        content_padding: str = "32px",
        duration_fast: str = "100ms",
        duration_base: str = "180ms",
        duration_slow: str = "280ms",
        easing: str = "cubic-bezier(0.16, 1, 0.3, 1)",
        show_grain_texture: bool = True,
        show_accent_line: bool = True,
        compact_mode: bool = False,
    ):
        defaults = PRESET_DEFAULTS.get(preset, PRESET_DEFAULTS["editorial"])
        self.preset = preset
        self.primary_color = primary_color or defaults["primary_color"]
        self.surface_base = surface_base or defaults["surface_base"]
        self.surface_raised = surface_raised or defaults["surface_raised"]
        self.text_primary = text_primary or defaults["text_primary"]
        self.text_secondary = text_secondary or defaults["text_secondary"]
        self.border_color = border_color or defaults["border_color"]
        self.font_display = font_display or defaults["font_display"]
        self.font_body = font_body or defaults["font_body"]
        self.font_mono = font_mono or defaults["font_mono"]
        self.font_import_url = font_import_url or defaults["font_import_url"]
        self.radius_sm = radius_sm or defaults["radius_sm"]
        self.radius_md = radius_md or defaults["radius_md"]
        self.radius_lg = radius_lg or defaults["radius_lg"]
        self.shadow_sm = shadow_sm
        self.shadow_md = shadow_md
        self.shadow_lg = shadow_lg
        self.topbar_height = topbar_height
        self.sidebar_width = sidebar_width
        self.sidebar_collapsed_width = sidebar_collapsed_width
        self.content_max_width = content_max_width
        self.content_padding = content_padding
        self.duration_fast = duration_fast
        self.duration_base = duration_base
        self.duration_slow = duration_slow
        self.easing = easing
        self.show_grain_texture = show_grain_texture
        self.show_accent_line = show_accent_line
        self.compact_mode = compact_mode

    def to_css_variables(self) -> str:
        """Generate CSS :root{} block from config."""
        lines = [
            f"  --primary-500: {self.primary_color};",
            f"  --surface-base: {self.surface_base};",
            f"  --surface-raised: {self.surface_raised};",
            f"  --text-primary: {self.text_primary};",
            f"  --text-secondary: {self.text_secondary};",
            f"  --surface-border: {self.border_color};",
            f"  --font-display: {self.font_display};",
            f"  --font-body: {self.font_body};",
            f"  --font-mono: {self.font_mono};",
            f"  --topbar-height: {self.topbar_height};",
            f"  --sidebar-width: {self.sidebar_width};",
            f"  --sidebar-collapsed: {self.sidebar_collapsed_width};",
            f"  --content-max-width: {self.content_max_width};",
            f"  --content-padding: {self.content_padding};",
            f"  --radius-sm: {self.radius_sm};",
            f"  --radius-md: {self.radius_md};",
            f"  --radius-lg: {self.radius_lg};",
            f"  --duration-fast: {self.duration_fast};",
            f"  --duration-base: {self.duration_base};",
            f"  --duration-slow: {self.duration_slow};",
            f"  --easing-out: {self.easing};",
            f"  --admin-grain-opacity: {'0.025' if self.show_grain_texture else '0'};",
            f"  --admin-accent-line-opacity: {'0.4' if self.show_accent_line else '0'};",
        ]
        if self.shadow_sm:
            lines.append(f"  --shadow-sm: {self.shadow_sm};")
        if self.shadow_md:
            lines.append(f"  --shadow-md: {self.shadow_md};")
        if self.shadow_lg:
            lines.append(f"  --shadow-lg: {self.shadow_lg};")
        body = "\n".join(lines)
        return f":root {{\n{body}\n}}"

    def to_context(self) -> dict:
        """Return dict suitable for template context."""
        return {
            "theme": self,
            "theme_preset": self.preset,
            "theme_css": self.to_css_variables(),
            "theme_font_import_url": self.font_import_url,
            "theme_show_grain": self.show_grain_texture,
            "theme_show_accent_line": self.show_accent_line,
        }

to_css_variables()

Generate CSS :root{} block from config.

Source code in fastapi_admin_kit/config/theme.py

def to_css_variables(self) -> str:
    """Generate CSS :root{} block from config."""
    lines = [
        f"  --primary-500: {self.primary_color};",
        f"  --surface-base: {self.surface_base};",
        f"  --surface-raised: {self.surface_raised};",
        f"  --text-primary: {self.text_primary};",
        f"  --text-secondary: {self.text_secondary};",
        f"  --surface-border: {self.border_color};",
        f"  --font-display: {self.font_display};",
        f"  --font-body: {self.font_body};",
        f"  --font-mono: {self.font_mono};",
        f"  --topbar-height: {self.topbar_height};",
        f"  --sidebar-width: {self.sidebar_width};",
        f"  --sidebar-collapsed: {self.sidebar_collapsed_width};",
        f"  --content-max-width: {self.content_max_width};",
        f"  --content-padding: {self.content_padding};",
        f"  --radius-sm: {self.radius_sm};",
        f"  --radius-md: {self.radius_md};",
        f"  --radius-lg: {self.radius_lg};",
        f"  --duration-fast: {self.duration_fast};",
        f"  --duration-base: {self.duration_base};",
        f"  --duration-slow: {self.duration_slow};",
        f"  --easing-out: {self.easing};",
        f"  --admin-grain-opacity: {'0.025' if self.show_grain_texture else '0'};",
        f"  --admin-accent-line-opacity: {'0.4' if self.show_accent_line else '0'};",
    ]
    if self.shadow_sm:
        lines.append(f"  --shadow-sm: {self.shadow_sm};")
    if self.shadow_md:
        lines.append(f"  --shadow-md: {self.shadow_md};")
    if self.shadow_lg:
        lines.append(f"  --shadow-lg: {self.shadow_lg};")
    body = "\n".join(lines)
    return f":root {{\n{body}\n}}"

to_context()

Return dict suitable for template context.

Source code in fastapi_admin_kit/config/theme.py

def to_context(self) -> dict:
    """Return dict suitable for template context."""
    return {
        "theme": self,
        "theme_preset": self.preset,
        "theme_css": self.to_css_variables(),
        "theme_font_import_url": self.font_import_url,
        "theme_show_grain": self.show_grain_texture,
        "theme_show_accent_line": self.show_accent_line,
    }

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,
    ):
        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()

    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 = False,
        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",
    ):
        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

    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 AutoModelMixin)
        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 AutoModelMixin 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 AutoModelMixin or define a roles relationship on your model."
            )

        # Check password-related attributes for authentication
        missing_auth = []
        if not hasattr(model, "hashed_password"):
            missing_auth.append("hashed_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 AutoModelMixin or implement hashed_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 AutoModelMixin)
    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 AutoModelMixin 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 AutoModelMixin or define a roles relationship on your model."
        )

    # Check password-related attributes for authentication
    missing_auth = []
    if not hasattr(model, "hashed_password"):
        missing_auth.append("hashed_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 AutoModelMixin or implement hashed_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,
    ):
        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

    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,
    ) -> PaginationResult:
        """Execute paginated query and return results."""
        ...

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

Execute paginated query and return results.

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,
) -> PaginationResult:
    """Execute paginated query and return results."""
    ...

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,
        **kw: Any,
    ) -> PaginationResult:
        count_q = select(func.count()).select_from(stmt.subquery())
        total = (await session.execute(count_q)).scalar() or 0

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

        stmt = stmt.offset(offset).limit(per_page)
        result = await session.execute(stmt)
        items = list(result.unique().scalars().all())

        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."""
        return json.loads(base64.b64decode(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,
    ) -> PaginationResult:
        # 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)
            stmt = stmt.where(col > cursor_val)
        elif before:
            cursor_val = self._decode_cursor(before)
            stmt = stmt.where(col < cursor_val)
            # For backward pagination, we need to reverse order then flip results
            from sqlalchemy import desc as sa_desc

            # Check current ordering and reverse it
            stmt = stmt.order_by(sa_desc(col))

        # Count filtered total
        count_q = select(func.count()).select_from(stmt.subquery())
        total = (await session.execute(count_q)).scalar() or 0

        # Fetch per_page + 1 to detect has_next
        stmt = stmt.limit(per_page + 1)
        result = await session.execute(stmt)
        items = list(result.unique().scalars().all())

        # 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,
        **kw: Any,
    ) -> PaginationResult:
        # Count total to decide strategy
        count_q = select(func.count()).select_from(stmt.subquery())
        total = (await session.execute(count_q)).scalar() or 0

        if total <= self.threshold:
            result = await self._offset.paginate(stmt, session, per_page, **kw)
        else:
            result = await self._cursor.paginate(stmt, session, per_page, **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 = 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

        result = await self.session.execute(
            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))
        )
        for perm in result.scalars():
            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

        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

        result = await self.session.execute(
            select(UserPermission).where(UserPermission.user_id == self._user_id)
        )
        for perm in result.scalars():
            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

        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"``

        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.

        Note: This is a sync convenience wrapper. For async contexts,
        use the individual async methods directly.
        """
        if self._is_superuser:
            return PermissionSet(
                can_view=True,
                can_create=True,
                can_edit=True,
                can_delete=True,
            )
        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),
        )

    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")
        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"

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"``

    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.

Note: This is a sync convenience wrapper. For async contexts, use the individual async methods directly.

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.

    Note: This is a sync convenience wrapper. For async contexts,
    use the individual async methods directly.
    """
    if self._is_superuser:
        return PermissionSet(
            can_view=True,
            can_create=True,
            can_edit=True,
            can_delete=True,
        )
    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),
    )

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")
    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(
                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 — override to add custom context."""
        from fastapi_admin_kit.types import PermissionSet
        from fastapi_admin_kit.views.sidebar import inject_sidebar_context

        (
            items,
            total,
            page,
            per_page,
            next_cursor,
            has_next,
            pagination_mode,
        ) = await self.query_provider.get_list(request, q, page)

        active_filters: dict[str, str] = {}
        if self.admin.list_filter:
            for filter_field in self.admin.list_filter:
                val = request.query_params.get(f"filter_{filter_field}", "")
                if val:
                    active_filters[filter_field] = val
                for suffix in ("__gte", "__lte", "__from", "__to"):
                    val = request.query_params.get(f"filter_{filter_field}{suffix}", "")
                    if val:
                        active_filters[f"{filter_field}{suffix}"] = val

        display_columns = self._build_display_columns()
        filter_fields = await self._build_filter_fields(request)
        ordering = request.query_params.get("ordering", "")
        if not ordering and self.admin.ordering:
            ordering = self.admin.ordering[0]

        template_context = {
            "model": self.registered,
            "registered": self.registered,
            "display_columns": display_columns,
            "items": items,
            "search_query": q,
            "page": page,
            "total_pages": max(1, math.ceil(total / per_page)) if per_page else 1,
            "total": total,
            "per_page": per_page,
            "next_cursor": next_cursor,
            "has_next": has_next,
            "pagination_mode": pagination_mode,
            "filter_fields": filter_fields,
            "active_filters": active_filters,
            "ordering": ordering,
            "permissions": checker.permission_set(self.registered.table_name)
            if checker
            else PermissionSet(can_view=True, can_create=True, can_edit=True, can_delete=True),
            "list_actions": self.admin.get_list_actions(),
            "row_actions": self.admin.get_row_actions(),
            "list_tabs": getattr(self.admin, "list_tabs", []),
            "list_sections": getattr(self.admin, "list_sections", []),
            "ordering_field": getattr(self.admin, "ordering_field", None),
            "hide_ordering_field": getattr(self.admin, "hide_ordering_field", False),
            "list_filter_options": getattr(self.admin, "list_filter_options", {}),
            "list_filter_horizontal": getattr(self.admin, "list_filter_horizontal", False),
        }
        template_context.update(self._get_extra_context(request))
        await inject_sidebar_context(request, template_context)
        return template_context

    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)
        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 — override to add custom context.

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 — override to add custom context."""
    from fastapi_admin_kit.types import PermissionSet
    from fastapi_admin_kit.views.sidebar import inject_sidebar_context

    (
        items,
        total,
        page,
        per_page,
        next_cursor,
        has_next,
        pagination_mode,
    ) = await self.query_provider.get_list(request, q, page)

    active_filters: dict[str, str] = {}
    if self.admin.list_filter:
        for filter_field in self.admin.list_filter:
            val = request.query_params.get(f"filter_{filter_field}", "")
            if val:
                active_filters[filter_field] = val
            for suffix in ("__gte", "__lte", "__from", "__to"):
                val = request.query_params.get(f"filter_{filter_field}{suffix}", "")
                if val:
                    active_filters[f"{filter_field}{suffix}"] = val

    display_columns = self._build_display_columns()
    filter_fields = await self._build_filter_fields(request)
    ordering = request.query_params.get("ordering", "")
    if not ordering and self.admin.ordering:
        ordering = self.admin.ordering[0]

    template_context = {
        "model": self.registered,
        "registered": self.registered,
        "display_columns": display_columns,
        "items": items,
        "search_query": q,
        "page": page,
        "total_pages": max(1, math.ceil(total / per_page)) if per_page else 1,
        "total": total,
        "per_page": per_page,
        "next_cursor": next_cursor,
        "has_next": has_next,
        "pagination_mode": pagination_mode,
        "filter_fields": filter_fields,
        "active_filters": active_filters,
        "ordering": ordering,
        "permissions": checker.permission_set(self.registered.table_name)
        if checker
        else PermissionSet(can_view=True, can_create=True, can_edit=True, can_delete=True),
        "list_actions": self.admin.get_list_actions(),
        "row_actions": self.admin.get_row_actions(),
        "list_tabs": getattr(self.admin, "list_tabs", []),
        "list_sections": getattr(self.admin, "list_sections", []),
        "ordering_field": getattr(self.admin, "ordering_field", None),
        "hide_ordering_field": getattr(self.admin, "hide_ordering_field", False),
        "list_filter_options": getattr(self.admin, "list_filter_options", {}),
        "list_filter_horizontal": getattr(self.admin, "list_filter_horizontal", False),
    }
    template_context.update(self._get_extra_context(request))
    await inject_sidebar_context(request, template_context)
    return template_context

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,
    ) -> dict[str, Any]:
        """Build form template context."""
        from fastapi_admin_kit.form.pipeline import (
            build_form_context as _build_form_ctx,
        )
        from fastapi_admin_kit.types import PermissionSet
        from fastapi_admin_kit.views.sidebar import inject_sidebar_context

        ctx = _build_form_ctx(
            self.registered,
            obj=obj,
            values=values,
            errors=errors,
            request=request,
            is_create=is_create,
        )
        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,
            "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
            ),
        }
        template_context.update(self._get_extra_context(request))
        await inject_sidebar_context(request, template_context)

        template_context = 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._pop_manytomany_keys(self.registered.model, parsed)
            resolved = self._resolve_rel_keys(parsed)
            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._apply_m2m_from_data(obj, m2m_data, session)
            self.admin.after_create(obj, request)
            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
        url = f"{request.app.state.admin_config['admin_path']}/{self.registered.table_name}/"
        return RedirectResponse(url=url, status_code=303)

    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)

        # POST
        parsed, errors = await self.form_parser.parse(request)
        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)

        try:
            parsed = self.admin.validate_create(parsed, request)
        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)

        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)
        m2m_data = self._pop_manytomany_keys(self.registered.model, parsed)
        resolved = self._resolve_rel_keys(parsed)
        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._apply_m2m_from_data(obj, m2m_data, session)
        self.admin.after_create(obj, request)
        await flush_pending_perm_ops(request)
        return await self.api_renderer.render(request, self._serialize(obj))

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,
    ) -> dict[str, Any]:
        """Build form template context."""
        from fastapi_admin_kit.form.pipeline import (
            build_form_context as _build_form_ctx,
        )
        from fastapi_admin_kit.types import PermissionSet
        from fastapi_admin_kit.views.sidebar import inject_sidebar_context

        ctx = _build_form_ctx(
            self.registered,
            obj=obj,
            values=values,
            errors=errors,
            request=request,
            is_create=is_create,
            rel_labels=rel_labels,
        )
        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,
            "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
            ),
        }
        template_context.update(self._get_extra_context(request))
        await inject_sidebar_context(request, template_context)

        template_context = 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._pop_manytomany_keys(obj, parsed)
            self._apply_parsed(obj, parsed)
            session = get_db_session(request)
            await self._apply_m2m_from_data(obj, m2m_data, session)
            self.admin.on_update(obj, parsed, request)
            await session.flush()
            self.admin.after_update(obj, request)
            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
        url = f"{request.app.state.admin_config['admin_path']}/{self.registered.table_name}/"
        return RedirectResponse(url=url, status_code=303)

    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.form.pipeline import (
            build_form_context as _build_form_ctx,
        )
        from fastapi_admin_kit.types import PermissionSet
        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,
            "permissions": checker.permission_set(self.registered.table_name)
            if checker
            else PermissionSet(can_view=True, can_create=True, can_edit=True, can_delete=True),
        }
        template_context.update(self._get_extra_context(request))
        await inject_sidebar_context(request, template_context)
        return template_context

    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 request.method == "GET":
            if perms and not perms.can_edit and perms.can_view:
                ctx = await self._build_detail_context(request, obj, checker)
                return request.app.state.admin_jinja_env.TemplateResponse(
                    request, "pages/detail.html", 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)

        # POST
        parsed, errors = await self.form_parser.parse(request, obj=obj)
        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)

        try:
            parsed = self.admin.validate_update(obj, parsed, request)
        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 = 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, _ = await parser.parse(request, obj)
        try:
            m2m_data = self._pop_manytomany_keys(obj, parsed)
            self._apply_parsed(obj, parsed)
            session = get_db_session(request)
            await self._apply_m2m_from_data(obj, m2m_data, session)
            self.admin.on_update(obj, parsed, request)
            await session.flush()
            self.admin.after_update(obj, request)
            await flush_pending_perm_ops(request)
        except Exception:
            session = get_db_session(request)
            await session.rollback()
            raise
        return self._serialize(obj)

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 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)
        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 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 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

    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

    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."""
    columns = list(registered.columns)

    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