diff --git a/docs/guide/json-api.md b/docs/guide/json-api.md index b7faeda..7ac44d0 100644 --- a/docs/guide/json-api.md +++ b/docs/guide/json-api.md @@ -27,6 +27,7 @@ The API is mounted at `/admin/api/` by default. | `POST` | `/admin/api/{model}/` | Create record | | `GET` | `/admin/api/{model}/{id}` | Get single record | | `PUT` | `/admin/api/{model}/{id}` | Update record | +| `PATCH` | `/admin/api/{model}/{id}` | Partially update record (only provided fields) | | `DELETE` | `/admin/api/{model}/{id}` | Delete record | ### Roles @@ -59,6 +60,49 @@ class SecretAdmin(ModelAdmin): skip_auto_routes = True ``` +## Per-model endpoint control (`export_endpoint`) + +`ModelAdmin.export_endpoint` controls **which routers are auto-built** for a +model. It only affects the routers generated by `admin.setup()` — admin HTML +routes are never shown in the `/openapi.json` Swagger doc, only JSON API routes +are. + +| Value | Admin (HTML) router | JSON API router | +|---------|---------------------|-----------------| +| `None` | built | built | +| `"html"`| built | skipped | +| `"api"` | skipped | built | + +```python +@admin.register(Product) +class ProductAdmin(ModelAdmin): + export_endpoint = "api" # JSON API only — no /admin/products pages +``` + +With `export_endpoint = "api"` the model is also hidden from the sidebar and +topbar search suggestions (it has no HTML pages). + +### Standalone router export + +You can build a model's routers **without** calling `admin.register()` at all +using `export_api_route()` / `export_admin_route()`: + +```python +from fastapi_admin_kit import ModelAdmin + +class ProductAdmin(ModelAdmin): + export_endpoint = "api" + +app.include_router(ProductAdmin().export_api_route(Product)) +app.include_router(ProductAdmin().export_admin_route(Product), prefix="/admin") +``` + +- `export_api_route(model, prefix="")` — JSON CRUD router (appears in Swagger). +- `export_admin_route(model, prefix="")` — HTML admin router (hidden from Swagger). + +These helpers build the routers directly and do **not** write to the admin +registry, so no `admin.register()` (and no sidebar entry) is created. + ## Authentication ### Token Obtain diff --git a/docs/guide/model-registration.md b/docs/guide/model-registration.md index 2e62c90..d4e4d62 100644 --- a/docs/guide/model-registration.md +++ b/docs/guide/model-registration.md @@ -157,6 +157,36 @@ When `inline_edit = True`, a 3-dot menu appears per row with an "Edit" option th | `nav_order` | `int` | `999` | Sidebar ordering (lower = higher) | | `nav_children` | `list[NavItemConfig]` | `None` | Nested nav items | | `skip_auto_routes` | `bool` | `False` | Skip automatic route generation | +| `export_endpoint` | `str \| None` | `None` | Control which routers are auto-built (`None`, `"html"`, or `"api"`) | + +### Endpoint Export Control + +`export_endpoint` controls which routers are auto-built for a model: + +| Value | Admin (HTML) router | JSON API router | +|---------|---------------------|-----------------| +| `None` | built | built | +| `"html"`| built | skipped | +| `"api"` | skipped | built | + +```python +@admin.register(Product) +class ProductAdmin(ModelAdmin): + export_endpoint = "api" # JSON API only +``` + +Admin HTML routes are never shown in `/openapi.json`; only JSON API routes are. + +You can also build routers for a model **without** `admin.register()` using the +standalone `export_api_route()` / `export_admin_route()` helpers: + +```python +class ProductAdmin(ModelAdmin): + export_endpoint = "api" + +app.include_router(ProductAdmin().export_api_route(Product)) +app.include_router(ProductAdmin().export_admin_route(Product), prefix="/admin") +``` ### Pagination Strategies diff --git a/example/example.py b/example/example.py index 43910ad..3a480a4 100644 --- a/example/example.py +++ b/example/example.py @@ -29,7 +29,8 @@ AuditLog, # noqa: F401 — ensure table is created ) from fastapi_admin_kit.auth.backend import BuiltinAuthBackend -from fastapi_admin_kit.auth.models import User +from fastapi_admin_kit.auth.mixins import AuthModelMixin +from fastapi_admin_kit.auth.password import password_manager from fastapi_admin_kit.backends import SqlAlchemyBackend from fastapi_admin_kit.config import ThemeConfig from fastapi_admin_kit.dashboard import ( @@ -40,6 +41,7 @@ ) from fastapi_admin_kit.inline import StackedInline, TabularInline from fastapi_admin_kit.models import Base as AdminBase +from fastapi_admin_kit.pagination.cursor import CursorPagination from fastapi_admin_kit.types import TabConfig, TableSection from fastapi_admin_kit.widgets.inputs import ArrayWidget, WysiwygWidget @@ -95,7 +97,7 @@ def __str__(self) -> str: return self.name -class User(Base): +class User(AuthModelMixin, Base): """User model.""" __tablename__ = "users" @@ -313,6 +315,7 @@ class ProductAdmin(ModelAdmin): "status", "created_at", ] + pagination = CursorPagination(cursor_column="id") list_filter = ["is_active", "category"] search_fields = ["name", "description"] ordering = ["-created_at"] @@ -711,8 +714,14 @@ async def seed_demo_data(session: AsyncSession) -> None: session.add_all(products) await session.flush() - user1 = User(email="alice@example.com", full_name="Alice Johnson", is_active=True) - user2 = User(email="bob@example.com", full_name="Bob Smith", is_active=True) + user1 = User( + email="alice@example.com", full_name="Alice Johnson", is_active=True, + hashed_password=password_manager.hash("alice"), + ) + user2 = User( + email="bob@example.com", full_name="Bob Smith", is_active=True, + hashed_password=password_manager.hash("bob"), + ) session.add_all([user1, user2]) await session.flush() @@ -739,7 +748,7 @@ async def seed_demo_data(session: AsyncSession) -> None: async def seed_admin_user(session: AsyncSession) -> None: """Create a default superadmin if none exists.""" - result = await session.execute(select(User).limit(1)) + result = await session.execute(select(User).where(User.email == "admin@example.com")) if result.scalars().first() is not None: return diff --git a/fastapi_admin_kit/admin/admin_router.py b/fastapi_admin_kit/admin/admin_router.py index 91e5e56..5994def 100644 --- a/fastapi_admin_kit/admin/admin_router.py +++ b/fastapi_admin_kit/admin/admin_router.py @@ -31,7 +31,12 @@ def _build_router(self, app: Any) -> None: for registered in registry.all(): if getattr(registered.admin, "skip_auto_routes", False): continue + # API-only models (export_endpoint="api") get no admin HTML router. + if getattr(registered.admin, "export_endpoint", None) == "api": + continue model_router = build_model_router(registered) + if model_router is None: + continue app.include_router(model_router, prefix=self.admin_path) # Auth routes (login/logout) diff --git a/fastapi_admin_kit/admin/builtin_models.py b/fastapi_admin_kit/admin/builtin_models.py index ee51e73..cf05319 100644 --- a/fastapi_admin_kit/admin/builtin_models.py +++ b/fastapi_admin_kit/admin/builtin_models.py @@ -70,7 +70,7 @@ async def flush_pending_perm_ops(request): from fastapi_admin_kit.db import get_db_session perm_ids = getattr(request.state, "_admin_perm_perm_ids", None) - if not perm_ids or not isinstance(perm_ids, list): + if perm_ids is None or not isinstance(perm_ids, list): return # Get the user object from request state @@ -169,7 +169,7 @@ def after_create(self, obj, request=None): if request is None: return perm_data = getattr(request.state, "_admin_perm_data", None) - if perm_data: + if perm_data is not None: request.state._admin_perm_perm_ids = perm_data request.state._admin_perm_user_obj = obj @@ -177,7 +177,7 @@ def after_update(self, obj, request=None): if request is None: return perm_data = getattr(request.state, "_admin_perm_data", None) - if perm_data: + if perm_data is not None: request.state._admin_perm_perm_ids = perm_data request.state._admin_perm_user_obj = obj diff --git a/fastapi_admin_kit/admin/core.py b/fastapi_admin_kit/admin/core.py index 0adb85c..ab15367 100644 --- a/fastapi_admin_kit/admin/core.py +++ b/fastapi_admin_kit/admin/core.py @@ -468,7 +468,7 @@ def __init__( # Store notification paths on config for template access default_notifications_path = f"{self.router.admin_path}/notifications" - default_notifications_list = f"{default_notifications_path}/" + default_notifications_list = f"{self.router.admin_path}/admin_notifications/" self.config.notifications_api_path = notifications_api_path or default_notifications_path self.config.notifications_list_path = notifications_list_path or default_notifications_list @@ -1032,7 +1032,7 @@ def _attr(obj: Any, name: str) -> Any: self.config, "notifications_api_path", f"{self.router.admin_path}/notifications" ) self._jinja_env.env.globals["notifications_list_path"] = getattr( - self.config, "notifications_list_path", f"{self.router.admin_path}/notifications/" + self.config, "notifications_list_path", f"{self.router.admin_path}/admin_notifications/" ) self._jinja_env.env.globals["notifications_enabled"] = self._enable_notification self._jinja_env.env.globals["nav_groups"] = self._nav_groups_built @@ -1207,7 +1207,12 @@ def _build_router(self, app: FastAPI) -> None: for registered in self.registry.all(): if getattr(registered.admin, "skip_auto_routes", False): continue + # API-only models (export_endpoint="api") get no admin HTML router. + if getattr(registered.admin, "export_endpoint", None) == "api": + continue model_router = build_model_router(registered) + if model_router is None: + continue app.include_router(model_router, prefix=self.router.admin_path) # Auth routes (login/logout) @@ -1235,6 +1240,7 @@ def _build_router(self, app: FastAPI) -> None: dashboard_view, methods=["GET"], tags=["admin"], + include_in_schema=False, ) # JSON API for external frontend apps diff --git a/fastapi_admin_kit/api/__init__.py b/fastapi_admin_kit/api/__init__.py index 27d144b..da78cbc 100644 --- a/fastapi_admin_kit/api/__init__.py +++ b/fastapi_admin_kit/api/__init__.py @@ -4,11 +4,12 @@ from typing import Any -from fastapi import APIRouter +from fastapi import APIRouter, Depends from fastapi_admin_kit.api.auth import router as auth_router from fastapi_admin_kit.api.crud import build_api_router from fastapi_admin_kit.api.roles import router as roles_router +from fastapi_admin_kit.api.security import bearer_scheme class AdminAPIRouter: @@ -33,12 +34,12 @@ def build_router(self) -> APIRouter: # Auth routes (token obtain, refresh, logout, me) router.include_router(auth_router) - # Role management routes (superuser only) - router.include_router(roles_router) + # Role management routes (superuser only) — bearer protected + router.include_router(roles_router, dependencies=[Depends(bearer_scheme)]) - # CRUD routes for all registered models + # CRUD routes for all registered models — bearer protected if self.registry is not None: crud_router = build_api_router(self.registry) - router.include_router(crud_router) + router.include_router(crud_router, dependencies=[Depends(bearer_scheme)]) return router diff --git a/fastapi_admin_kit/api/auth.py b/fastapi_admin_kit/api/auth.py index f6ec8b1..56bc52f 100644 --- a/fastapi_admin_kit/api/auth.py +++ b/fastapi_admin_kit/api/auth.py @@ -8,7 +8,8 @@ from typing import Any import jwt -from fastapi import APIRouter, HTTPException, Request +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.security import HTTPBasicCredentials from fastapi_admin_kit.api.schemas import ( RefreshRequest, @@ -16,6 +17,7 @@ TokenRequest, TokenResponse, ) +from fastapi_admin_kit.api.security import basic_scheme, bearer_scheme from fastapi_admin_kit.auth.ratelimit import RateLimiter, check_rate_limit from fastapi_admin_kit.db import get_db_session @@ -161,10 +163,22 @@ def _hash_token(token: str) -> str: @router.post("/token", response_model=TokenResponse) async def obtain_token( request: Request, - body: TokenRequest, + body: TokenRequest | None = None, + credentials: HTTPBasicCredentials | None = Depends(basic_scheme), ) -> TokenResponse: - """POST /api/auth/token — obtain JWT access + refresh tokens.""" - check_rate_limit(_api_rate_limiter, body.email) + """POST /api/auth/token — obtain JWT access + refresh tokens. + + Credentials may be provided either as a JSON body (``email``/``password``) + or via HTTP Basic auth. Basic auth takes precedence when both are given. + """ + if credentials is not None: + email, password = credentials.username, credentials.password + elif body is not None: + email, password = body.email, body.password + else: + raise HTTPException(status_code=422, detail="Credentials required.") + + check_rate_limit(_api_rate_limiter, email) auth_backend = getattr(request.app.state, "admin_auth_backend", None) if auth_backend is None: @@ -174,12 +188,12 @@ async def obtain_token( if db_session is None: raise HTTPException(status_code=500, detail="Database session not available.") - user = await auth_backend.authenticate(body.email, body.password, db_session) + user = await auth_backend.authenticate(email, password, db_session) if user is None: - _api_rate_limiter.record_attempt(body.email) + _api_rate_limiter.record_attempt(email) raise HTTPException(status_code=401, detail="Invalid credentials.") - _api_rate_limiter.reset(body.email) + _api_rate_limiter.reset(email) secret_key = _get_secret_key(request) ttl = _get_token_ttl(request) @@ -221,6 +235,7 @@ async def refresh_token( raise HTTPException(status_code=500, detail="Database session not available.") from sqlalchemy import select + from sqlalchemy.orm import selectinload from fastapi_admin_kit.auth.models import RefreshToken, User @@ -235,12 +250,17 @@ async def refresh_token( if refresh_record is None: raise HTTPException(status_code=401, detail="Invalid refresh token.") - if refresh_record.expires_at < datetime.now(UTC): + expires_at = refresh_record.expires_at + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=UTC) + if expires_at < datetime.now(UTC): raise HTTPException(status_code=401, detail="Refresh token expired.") - # Load user + # Load user (eagerly load roles to avoid lazy-load in async session) user = await db_session.scalar_one_or_none( - select(User).where( + select(User) + .options(selectinload(User.roles)) + .where( User.id == refresh_record.user_id, User.is_active, ) @@ -305,6 +325,7 @@ async def api_logout( @router.get("/me") async def get_current_user_info( request: Request, + _: Any = Depends(bearer_scheme), ) -> dict[str, Any]: """GET /api/auth/me — return current user info from JWT (no DB hit).""" auth_header = request.headers.get("Authorization", "") diff --git a/fastapi_admin_kit/api/crud.py b/fastapi_admin_kit/api/crud.py index 5c72c79..9722778 100644 --- a/fastapi_admin_kit/api/crud.py +++ b/fastapi_admin_kit/api/crud.py @@ -5,11 +5,13 @@ from __future__ import annotations -from typing import Any +from typing import Annotated, Any -from fastapi import APIRouter, HTTPException, Query, Request -from fastapi.responses import JSONResponse +from fastapi import APIRouter, Body, Depends, HTTPException, Request +from fastapi.responses import Response +from pydantic import BaseModel +from fastapi_admin_kit.api.deps import require_api_permission from fastapi_admin_kit.api.schema_generator import get_or_build_schemas from fastapi_admin_kit.views.class_views import ( CreateView, @@ -43,6 +45,11 @@ async def _check_permission( ) +def _export_endpoint(registered: Any) -> str | None: + """Return the model's ``export_endpoint`` setting (None / "html" / "api").""" + return getattr(registered.admin, "export_endpoint", None) + + def build_api_router(registry: Any) -> APIRouter: """Build the CRUD API router for all registered models.""" router = APIRouter(tags=["api-crud"]) @@ -53,15 +60,98 @@ def build_api_router(registry: Any) -> APIRouter: # admin_refresh_tokens / admin_user_totp are never exposed over JSON API. if getattr(registered.admin, "skip_auto_routes", False): continue - _register_model_routes(router, registered) + # Models with export_endpoint="html" only expose their admin HTML + # router — the JSON API router is skipped. + if _export_endpoint(registered) == "html": + continue + router.include_router(build_api_router_for_model(registered)) + + return router + +def build_api_router_for_model(registered: Any) -> APIRouter: + """Build a standalone CRUD router for a single model. + + Used by ``AdminRegistry``-driven route building and by the standalone + ``ModelAdmin.export_api_route`` helper, so a model's JSON API can be + exposed without registering it with the admin. + """ + router = APIRouter( + prefix=f"/{registered.table_name}", + tags=["api-crud", registered.verbose_name], + ) + _register_model_routes(router, registered) return router +def _wrap_body_handler( + handler: Any, payload_schema: type[BaseModel], *, include_item_id: bool = False +) -> Any: + """Wrap a view handler so FastAPI can document the request body. + + The view handlers parse the JSON body themselves, so we accept the + generated Pydantic schema as a body parameter and stash the parsed dict + on ``request.state`` for ``JSONBodyParser`` to pick up. + + Set ``include_item_id`` only for routes whose path contains ``{item_id}`` + so it is documented as a path parameter (and never leaks a query param + on collection routes like POST). + """ + payload_type = Annotated[payload_schema, Body(...)] + + if include_item_id: + + async def wrapped(request: Request, payload: payload_type, item_id: Any) -> Any: + request.state._api_payload = payload.model_dump(exclude_unset=True) + return await handler(request, item_id=item_id) + + wrapped.__annotations__ = { + "request": Request, + "payload": payload_type, + "item_id": Any, + "return": Any, + } + else: + + async def wrapped(request: Request, payload: payload_type) -> Any: + request.state._api_payload = payload.model_dump(exclude_unset=True) + return await handler(request) + + wrapped.__annotations__ = { + "request": Request, + "payload": payload_type, + "return": Any, + } + + wrapped.__name__ = getattr(handler, "__name__", "api_response") + wrapped.__doc__ = getattr(handler, "__doc__", None) + return wrapped + + +def _wrap_item_handler(handler: Any, *, returns_response: bool = False) -> Any: + """Wrap a single-item view handler so only the ``item_id`` path param is exposed. + + The underlying handlers accept both ``id`` and ``item_id`` (the admin + HTML routes pass ``id``); for the JSON API we only want ``item_id`` so a + redundant ``id`` query parameter is not leaked into the OpenAPI docs. + """ + + async def wrapped(request: Request, item_id: Any) -> Any: + return await handler(request, item_id=item_id) + + wrapped.__annotations__ = { + "request": Request, + "item_id": Any, + "return": Response if returns_response else Any, + } + wrapped.__name__ = getattr(handler, "__name__", "api_response") + wrapped.__doc__ = getattr(handler, "__doc__", None) + return wrapped + + def _register_model_routes(router: APIRouter, registered: Any) -> None: """Register CRUD routes for a single model using view classes.""" table_name = registered.table_name - prefix = f"/{table_name}" # DIP: resolve view classes from ModelAdmin config admin = registered.admin @@ -74,61 +164,56 @@ def _register_model_routes(router: APIRouter, registered: Any) -> None: schemas = get_or_build_schemas(registered) response_schema = schemas["response"] list_response_schema = schemas["list_response"] - - @router.get(prefix, response_model=list_response_schema) - async def list_items( - request: Request, - page: int = Query(1, ge=1), - per_page: int = Query(25, ge=1, le=100), - q: str = Query(""), - order: str = Query(""), - after: str | None = Query(None), - before: str | None = Query(None), - ): - user = await _get_current_user(request) - await _check_permission(request, user, table_name, "view") - result = await list_v.api_response( - request, - page=page, - per_page=per_page, - q=q, - order=order, - after=after, - before=before, - ) - if isinstance(result, JSONResponse): - return result - return result - - @router.post(prefix, response_model=response_schema, status_code=201) - async def create_item(request: Request): - user = await _get_current_user(request) - await _check_permission(request, user, table_name, "create") - result = await create_v.api_response(request) - if isinstance(result, JSONResponse): - return result - return result - - @router.get(f"{prefix}/{{item_id}}", response_model=response_schema) - async def retrieve_item(request: Request, item_id: str): - user = await _get_current_user(request) - await _check_permission(request, user, table_name, "view") - result = await edit_v.api_response(request, item_id=item_id) - if isinstance(result, JSONResponse): - return result - return result - - @router.put(f"{prefix}/{{item_id}}", response_model=response_schema) - async def update_item(request: Request, item_id: str): - user = await _get_current_user(request) - await _check_permission(request, user, table_name, "edit") - result = await edit_v.api_response(request, item_id=item_id) - if isinstance(result, JSONResponse): - return result - return result - - @router.delete(f"{prefix}/{{item_id}}", status_code=204) - async def delete_item(request: Request, item_id: str): - user = await _get_current_user(request) - await _check_permission(request, user, table_name, "delete") - return await delete_v.api_response(request, item_id=item_id) + create_schema = schemas["create"] + update_schema = schemas["update"] + + # Add routes with both "api-crud" and model verbose_name tags + router.add_api_route( + "", + list_v.api_response if hasattr(list_v, "api_response") else list_v, + methods=["GET"], + response_model=list_response_schema, + tags=["api-crud", registered.verbose_name], + dependencies=[Depends(require_api_permission(table_name, "view"))], + ) + router.add_api_route( + "", + _wrap_body_handler(create_v.api_response, create_schema), + methods=["POST"], + response_model=response_schema, + status_code=201, + tags=["api-crud", registered.verbose_name], + dependencies=[Depends(require_api_permission(table_name, "create"))], + ) + router.add_api_route( + "/{item_id}", + _wrap_item_handler(edit_v.api_response), + methods=["GET"], + response_model=response_schema, + tags=["api-crud", registered.verbose_name], + dependencies=[Depends(require_api_permission(table_name, "view"))], + ) + router.add_api_route( + "/{item_id}", + _wrap_body_handler(edit_v.api_response, update_schema, include_item_id=True), + methods=["PUT"], + response_model=response_schema, + tags=["api-crud", registered.verbose_name], + dependencies=[Depends(require_api_permission(table_name, "edit"))], + ) + router.add_api_route( + "/{item_id}", + _wrap_body_handler(edit_v.api_response, update_schema, include_item_id=True), + methods=["PATCH"], + response_model=response_schema, + tags=["api-crud", registered.verbose_name], + dependencies=[Depends(require_api_permission(table_name, "edit"))], + ) + router.add_api_route( + "/{item_id}", + _wrap_item_handler(delete_v.api_response, returns_response=True), + methods=["DELETE"], + status_code=204, + tags=["api-crud", registered.verbose_name], + dependencies=[Depends(require_api_permission(table_name, "delete"))], + ) diff --git a/fastapi_admin_kit/api/deps.py b/fastapi_admin_kit/api/deps.py index c166634..eb2dde3 100644 --- a/fastapi_admin_kit/api/deps.py +++ b/fastapi_admin_kit/api/deps.py @@ -1,4 +1,11 @@ -"""API dependencies — JWT-based permission checking without DB hits.""" +"""API dependencies — JWT-based permission checking with DB fallback. + +Primary source of truth is the permission snapshot embedded in the JWT at +login time (fast, no DB hit). When the snapshot does not grant the action, +we fall back to a live :class:`PermissionChecker` query so permissions that +were granted *after* the token was issued take effect immediately instead of +requiring the user to log in again. +""" from __future__ import annotations @@ -26,6 +33,46 @@ async def get_api_current_user(request: Request) -> dict[str, Any]: return payload +async def _check_live_permission( + request: Request, + user: dict[str, Any], + table_name: str, + action: str, +) -> bool: + """Check *action* on *table_name* against the database. + + Resolves the current user from the JWT subject via the configured auth + backend and runs a fresh :class:`PermissionChecker`. Used as a fallback + when the JWT-embedded permission snapshot is stale. + """ + sub = user.get("sub") + if sub is None: + return False + try: + user_id: int | str = int(sub) + except (TypeError, ValueError): + return False + + from fastapi_admin_kit.auth.identity import resolve_user + from fastapi_admin_kit.auth.permissions import PermissionChecker + from fastapi_admin_kit.db import get_db_session + + session = get_db_session(request) + if session is None: + return False + + resolved = await resolve_user(request, user_id) + if resolved is None: + return False + + checker = PermissionChecker( + session=session, + user=resolved, + user_snapshot=getattr(request.state, "admin_user_snapshot", None), + ) + return await checker.has_permission(table_name, action) + + def require_api_permission(table_name: str, action: str): """Return a dependency that checks JWT-embedded permissions. @@ -34,26 +81,30 @@ def require_api_permission(table_name: str, action: str): @router.get("/") async def list_view(user=Depends(require_api_permission("products", "view"))): ... + + Superusers always pass. When the JWT snapshot does not grant the action, + a live DB check is performed so newly-granted permissions take effect + without requiring the user to re-authenticate. """ - async def _check( - user: dict[str, Any] = None, - request: Request = None, - ) -> dict[str, Any]: - if user is None: - user = await get_api_current_user(request) + async def _check(request: Request) -> dict[str, Any]: + user = await get_api_current_user(request) if user.get("is_superuser"): return user permissions = user.get("permissions", {}) table_perms = permissions.get(table_name, []) - if action not in table_perms: - raise HTTPException( - status_code=403, - detail=f"You do not have permission to {action} {table_name}.", - ) - return user + if action in table_perms: + return user + + if await _check_live_permission(request, user, table_name, action): + return user + + raise HTTPException( + status_code=403, + detail=f"You do not have permission to {action} {table_name}.", + ) return _check @@ -61,12 +112,8 @@ async def _check( def require_api_superuser(): """Return a dependency that enforces superuser access from JWT.""" - async def _check( - user: dict[str, Any] = None, - request: Request = None, - ) -> dict[str, Any]: - if user is None: - user = await get_api_current_user(request) + async def _check(request: Request) -> dict[str, Any]: + user = await get_api_current_user(request) if not user.get("is_superuser"): raise HTTPException(status_code=403, detail="Superuser access required.") diff --git a/fastapi_admin_kit/api/roles.py b/fastapi_admin_kit/api/roles.py index b2623c4..832a4a0 100644 --- a/fastapi_admin_kit/api/roles.py +++ b/fastapi_admin_kit/api/roles.py @@ -7,6 +7,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel from sqlalchemy import select +from sqlalchemy.orm import selectinload from fastapi_admin_kit.api.deps import require_api_superuser from fastapi_admin_kit.auth.models import Role @@ -39,7 +40,7 @@ async def list_roles( ) -> list[RoleResponse]: """GET /api/roles/ — list all roles (superuser only).""" db_session = get_db_session(request) - roles = await db_session.all(select(Role)) + roles = await db_session.all(select(Role).options(selectinload(Role.users))) return [ RoleResponse( id=r.id, diff --git a/fastapi_admin_kit/api/schema_generator.py b/fastapi_admin_kit/api/schema_generator.py index 6962f10..24f27e8 100644 --- a/fastapi_admin_kit/api/schema_generator.py +++ b/fastapi_admin_kit/api/schema_generator.py @@ -72,6 +72,31 @@ def _collect_fields(registered: Any, *, exclude_pk: bool = False) -> list[Any]: return columns +def _collect_relationships(registered: Any) -> list[Any]: + """Collect relationships to include in a schema, respecting ModelAdmin config.""" + admin = registered.admin + relationships = list(registered.relationships) + + if admin.fields is not None: + field_names = set(admin.fields) + relationships = [r for r in relationships if r.name in field_names] + + if admin.exclude: + relationships = [r for r in relationships if r.name not in admin.exclude] + + return relationships + + +def _relationship_python_type(rel: Any) -> type: + """Get the Python type for a relationship field in a request schema. + + MANYTOONE → the target primary key (int); MANYTOMANY → a list of PKs. + """ + if rel.direction == "MANYTOMANY": + return list[int] + return int + + def build_create_schema(registered: Any) -> type[BaseModel]: """Build a Pydantic model for create requests. @@ -96,6 +121,12 @@ def build_create_schema(registered: Any) -> type[BaseModel]: fields[col.name] = field_info + for rel in _collect_relationships(registered): + if rel.name in readonly or rel.name in fields: + continue + python_type = _relationship_python_type(rel) + fields[rel.name] = (python_type | None, Field(default=None)) + model_name = f"{registered.verbose_name.replace(' ', '')}Create" return create_model(model_name, __config__=None, **fields) @@ -118,6 +149,12 @@ def build_update_schema(registered: Any) -> type[BaseModel]: field_info = (python_type | None, Field(default=None)) fields[col.name] = field_info + for rel in _collect_relationships(registered): + if rel.name in readonly or rel.name in fields: + continue + python_type = _relationship_python_type(rel) + fields[rel.name] = (python_type | None, Field(default=None)) + model_name = f"{registered.verbose_name.replace(' ', '')}Update" return create_model(model_name, __config__=None, **fields) diff --git a/fastapi_admin_kit/api/search.py b/fastapi_admin_kit/api/search.py index 6940adc..f02f1fb 100644 --- a/fastapi_admin_kit/api/search.py +++ b/fastapi_admin_kit/api/search.py @@ -63,6 +63,9 @@ async def get_search_suggestions( for registered in registry.all(): if getattr(registered.admin, "skip_auto_routes", False): continue + # API-only models have no admin HTML pages — hide from suggestions. + if getattr(registered.admin, "export_endpoint", None) == "api": + continue table_name: str = registered.table_name verbose_name: str = registered.verbose_name verbose_name_plural: str = registered.verbose_name_plural diff --git a/fastapi_admin_kit/api/security.py b/fastapi_admin_kit/api/security.py new file mode 100644 index 0000000..19e5fe6 --- /dev/null +++ b/fastapi_admin_kit/api/security.py @@ -0,0 +1,23 @@ +"""OpenAPI security schemes for the Admin JSON API (Swagger docs). + +These are used as documented dependencies so Swagger UI shows "Authorize" +buttons for both HTTP Basic and HTTP Bearer. ``auto_error=False`` keeps them +purely declarative — the real auth enforcement lives in +:mod:`fastapi_admin_kit.api.deps` / the view handlers. +""" + +from __future__ import annotations + +from fastapi.security import HTTPBasic, HTTPBearer + +basic_scheme = HTTPBasic( + auto_error=False, + scheme_name="BasicAuth", + description="Admin credentials (email:password). Accepted by POST /api/auth/token.", +) + +bearer_scheme = HTTPBearer( + auto_error=False, + scheme_name="BearerAuth", + description="JWT access token returned by POST /api/auth/token.", +) diff --git a/fastapi_admin_kit/auth/backend.py b/fastapi_admin_kit/auth/backend.py index 83a577a..0c218ea 100644 --- a/fastapi_admin_kit/auth/backend.py +++ b/fastapi_admin_kit/auth/backend.py @@ -57,6 +57,7 @@ async def authenticate( login_field: str = "email", ) -> AdminUserProtocol | None: from sqlalchemy import select + from sqlalchemy.orm import selectinload session = as_session_backend(session) model = self._get_model() @@ -65,9 +66,13 @@ async def authenticate( field = getattr(model, "email", None) if field is None: return None - user = await session.scalar_one_or_none( - select(model).where(field == credential, model.is_active.is_(True)) - ) + query = select(model).where(field == credential, model.is_active.is_(True)) + + # Eagerly load roles if the model has a roles relationship + if hasattr(model, "roles"): + query = query.options(selectinload(model.roles)) + + user = await session.scalar_one_or_none(query) if not user: return None diff --git a/fastapi_admin_kit/auth/identity.py b/fastapi_admin_kit/auth/identity.py index 4c3a714..4014f3b 100644 --- a/fastapi_admin_kit/auth/identity.py +++ b/fastapi_admin_kit/auth/identity.py @@ -91,10 +91,10 @@ async def resolve_user(request: Request, user_id: int | str | None) -> AdminUser if auth_backend is None or session is None: return None - from fastapi_admin_kit.db import rollback_if_needed - - await rollback_if_needed(session) - + # Deliberately do NOT call rollback_if_needed here unconditionally — + # it would roll back the current request's session, destroying data such as + # newly inserted objects in API create/update flows. The pending-rollback + # state is better handled by SessionMiddleware or explicit error handling. user = await auth_backend.get_user(user_id, session) if user is None or not getattr(user, "is_active", False): return None diff --git a/fastapi_admin_kit/auth/views.py b/fastapi_admin_kit/auth/views.py index 27198e2..e2cae8f 100644 --- a/fastapi_admin_kit/auth/views.py +++ b/fastapi_admin_kit/auth/views.py @@ -39,7 +39,7 @@ def _is_safe_url(url: str | None) -> bool: return not (parsed.scheme or parsed.netloc) -@router.get("/login", response_class=HTMLResponse) +@router.get("/login", response_class=HTMLResponse, include_in_schema=False) async def login_get( request: Request, next: str | None = None, @@ -97,7 +97,7 @@ async def login_get( ) -@router.post("/login", response_model=None) +@router.post("/login", response_model=None, include_in_schema=False) async def login_post( request: Request, username: str = Form(...), @@ -193,7 +193,7 @@ async def login_post( ) -@router.post("/logout") +@router.post("/logout", include_in_schema=False) async def logout_post( request: Request, session_payload: dict[str, Any] | None = Depends(get_session), diff --git a/fastapi_admin_kit/db.py b/fastapi_admin_kit/db.py index bf80027..bd2f945 100644 --- a/fastapi_admin_kit/db.py +++ b/fastapi_admin_kit/db.py @@ -96,15 +96,9 @@ async def __call__(self, scope: dict, receive: Any, send: Any) -> None: raise else: if hasattr(session, "commit"): - try: - result = session.commit() - if hasattr(result, "__await__"): - await result - except Exception: - if hasattr(session, "rollback"): - result = session.rollback() - if hasattr(result, "__await__"): - await result + result = session.commit() + if hasattr(result, "__await__"): + await result finally: if hasattr(session, "close"): result = session.close() @@ -117,14 +111,20 @@ async def rollback_if_needed(session: Any) -> None: SQLAlchemy marks a session with a ``PendingRollbackError`` after a flush raises: every later operation on that session fails until it is rolled - back. This helper calls ``rollback()`` unconditionally because the - cost on a clean session is negligible, while the benefit of clearing a - pending-rollback state is essential. + back. This helper calls ``rollback()`` only when a + ``PendingRollbackError`` is detected, to avoid unnecessary rollbacks on + clean sessions. """ + from sqlalchemy.exc import PendingRollbackError + try: result = session.rollback() if hasattr(result, "__await__"): await result + except PendingRollbackError: + # Session already has pending rollback state from a previous flush + # failure — no action needed; the session is already marked for rollback. + pass except Exception: pass diff --git a/fastapi_admin_kit/inspection.py b/fastapi_admin_kit/inspection.py index 50628fa..3a7cde3 100644 --- a/fastapi_admin_kit/inspection.py +++ b/fastapi_admin_kit/inspection.py @@ -5,6 +5,7 @@ import re from typing import Any +from fastapi import HTTPException from sqlalchemy import inspect from fastapi_admin_kit.inspection.types import ColumnMeta, RelationMeta @@ -111,3 +112,40 @@ def model_display_name(obj: Any) -> str: return str(obj) pk = getattr(obj, "id", None) return f"{type(obj).__name__}:{pk}" if pk is not None else type(obj).__name__ + + +async def validate_related_id( + model: type, + related_model: type, + id_value: Any, + field_name: str = "id", +) -> Any: + """Validate that a related model ID exists in the database. + + Casts the ID value to the correct type for the related model's primary key, + fetches the related object, and raises ``HTTPException(404)`` if not found. + + Args: + model: The model containing the foreign key field (for type resolution). + related_model: The model whose ID is being validated. + id_value: The ID value from the request (typically a string from form data). + field_name: Name of the field for error messaging. + + Returns: + The cast primary key value if the related object exists. + + Raises: + HTTPException: 404 if the related object does not exist. + """ + from fastapi_admin_kit.db import get_db_session + from fastapi_admin_kit.inspection import cast_pk_value + + pk_value = cast_pk_value(related_model, id_value) + session = get_db_session(None) + obj = await session.get(related_model, pk_value) + if obj is None: + raise HTTPException( + status_code=404, + detail=f"{related_model.__name__} with {field_name}={id_value} not found", + ) + return pk_value diff --git a/fastapi_admin_kit/inspection/__init__.py b/fastapi_admin_kit/inspection/__init__.py index d07bd45..3041d6e 100644 --- a/fastapi_admin_kit/inspection/__init__.py +++ b/fastapi_admin_kit/inspection/__init__.py @@ -9,6 +9,8 @@ import re from typing import Any +from fastapi import Request + from fastapi_admin_kit.backends.sqlalchemy import SqlAlchemyIntrospectionAdapter from fastapi_admin_kit.inspection.types import ColumnMeta, RelationMeta @@ -22,7 +24,7 @@ def inspect_model(model: type) -> tuple[list[ColumnMeta], list[RelationMeta]]: def is_abstract(model: type) -> bool: """Check if a model is abstract and should be skipped during auto-discovery.""" - return _inspector.is_abstract(model) + return getattr(model, "__abstract__", False) def get_pk_field(model: type) -> str | None: @@ -120,3 +122,59 @@ def model_display_name(obj: Any) -> str: return str(label) pk = getattr(obj, "id", None) return f"{type(obj).__name__}:{pk}" if pk is not None else type(obj).__name__ + + +async def validate_related_id( + model: type, + related_model: type, + id_value: Any, + field_name: str = "id", + request: Request | None = None, +) -> tuple[Any, dict[str, Any] | None]: + """Validate that a related model ID exists in the database. + + Casts the ID value to the correct type for the related model's primary key, + fetches the related object, and returns an error context if not found. + + Args: + model: The model containing the foreign key field (for type resolution). + related_model: The model whose ID is being validated. + id_value: The ID value from the request (typically a string from form data). + field_name: Name of the field for error messaging. + request: FastAPI Request object. If not provided, session will be + obtained from the application state (legacy mode). + + Returns: + A tuple of (pk_value, error_ctx): + - pk_value: The cast primary key value if the related object exists, otherwise None + - error_ctx: None if valid, or dict with error context if invalid (for field validation). + The dict contains keys: "msg", "input", "ctx" compatible with + FastAPI's validation error format. Callers should check if error_ctx + is not None and raise HTTPException(422, detail=[error_ctx]) + + Note: + The error_ctx can be directly used in FastAPI's HTTPException detail list: + raise HTTPException( + status_code=422, + detail=[{ + "loc": ["body", field_name], + "msg": error_ctx["msg"], + "type": "value_error", + "input": id_value, + "ctx": error_ctx["ctx"], + }] + ) + """ + from fastapi_admin_kit.db import get_db_session + from fastapi_admin_kit.inspection import cast_pk_value + + pk_value = cast_pk_value(related_model, id_value) + session = get_db_session(request) if request else get_db_session(None) + obj = await session.get(related_model, pk_value) + if obj is None: + return None, { + "msg": f"{related_model.__name__} with {id_value} not found", + "input": id_value, + "ctx": {"value": id_value}, + } + return pk_value, None diff --git a/fastapi_admin_kit/modeladmin.py b/fastapi_admin_kit/modeladmin.py index 050c0a6..a8339b1 100644 --- a/fastapi_admin_kit/modeladmin.py +++ b/fastapi_admin_kit/modeladmin.py @@ -102,6 +102,10 @@ def get_ordering(request_params: dict, admin_ordering: list[str] | None) -> list # Route generation skip_auto_routes: bool = False + # Endpoint export control: None (default) → both HTML + JSON API routers + # are auto-built; "html" → only the admin HTML router is auto-built; + # "api" → only the JSON API router is auto-built. + export_endpoint: str | None = None # Custom display functions (dict-based fallback) display_functions: dict[str, Any] | None = None @@ -109,6 +113,65 @@ def get_ordering(request_params: dict, admin_ordering: list[str] | None) -> list # Decorator for custom column display column = staticmethod(column) + # ── Standalone router export (no admin.register required) ─────── + + def export_api_route(self, model: Any, prefix: str = "") -> Any: + """Build a standalone JSON API router for ``model``. + + Unlike ``admin.register()`` this does not write to the singleton + registry — the returned router can be mounted directly:: + + app.include_router(ProductAdmin().export_api_route(Product)) + + Args: + model: A SQLAlchemy declarative model class. + prefix: Optional extra path prefix prepended to the router. + + Returns: + An ``APIRouter`` exposing the model's JSON CRUD endpoints. + """ + from fastapi import APIRouter + + from fastapi_admin_kit.api.crud import build_api_router_for_model + from fastapi_admin_kit.registry import build_registered_model + + registered = build_registered_model(model, self) + router = build_api_router_for_model(registered) + if prefix: + wrapper = APIRouter(prefix=prefix) + wrapper.include_router(router) + return wrapper + return router + + def export_admin_route(self, model: Any, prefix: str = "") -> Any: + """Build a standalone HTML admin router for ``model``. + + Unlike ``admin.register()`` this does not write to the singleton + registry — the returned router can be mounted directly:: + + app.include_router(ProductAdmin().export_admin_route(Product), prefix="/admin") + + Args: + model: A SQLAlchemy declarative model class. + prefix: Optional extra path prefix prepended to the router. + + Returns: + An ``APIRouter`` exposing the model's HTML admin views. + """ + from fastapi import APIRouter + + from fastapi_admin_kit.registry import build_registered_model + from fastapi_admin_kit.router import build_model_router + + registered = build_registered_model(model, self) + # Explicit call — build regardless of the model's export_endpoint. + router = build_model_router(registered, force=True) + if prefix: + wrapper = APIRouter(prefix=prefix) + wrapper.include_router(router) + return wrapper + return router + # Badge hook — return str e.g. "12" or None def get_nav_badge(self, request: Any = None) -> str | None: return None diff --git a/fastapi_admin_kit/nav.py b/fastapi_admin_kit/nav.py index 65c16eb..398620d 100644 --- a/fastapi_admin_kit/nav.py +++ b/fastapi_admin_kit/nav.py @@ -102,6 +102,9 @@ def build( for registered in registry: if getattr(registered.admin, "skip_auto_routes", False): continue + # API-only models have no admin HTML pages — hide from sidebar. + if getattr(registered.admin, "export_endpoint", None) == "api": + continue tags = self._get_tags(registered) for tag in tags: buckets.setdefault(tag, []).append( diff --git a/fastapi_admin_kit/notifications/plugin.py b/fastapi_admin_kit/notifications/plugin.py index 60dd89a..c7b1afd 100644 --- a/fastapi_admin_kit/notifications/plugin.py +++ b/fastapi_admin_kit/notifications/plugin.py @@ -46,4 +46,7 @@ def configure_notifications(app: Any, service: Any, prefix: str = "/api/notifica current_api = getattr(admin.config, "notifications_api_path", None) if current_api in (None, default_api): admin.config.notifications_api_path = prefix.rstrip("/") - admin.config.notifications_list_path = f"{admin.config.notifications_api_path}/" + if admin.config.notifications_list_path is None: + admin.config.notifications_list_path = ( + f"{getattr(admin.router, 'admin_path', '/admin')}/admin_notifications/" + ) diff --git a/fastapi_admin_kit/notifications/router.py b/fastapi_admin_kit/notifications/router.py index c228310..39e6c47 100644 --- a/fastapi_admin_kit/notifications/router.py +++ b/fastapi_admin_kit/notifications/router.py @@ -15,6 +15,7 @@ from __future__ import annotations +import ast import asyncio import json import logging @@ -381,9 +382,23 @@ def _serialize(n: Any) -> NotificationOut: id=n.id, title=n.title, body=n.body, - channels=n.channels or [], - data=n.data, + channels=_as_json(n.channels, default=[]), + data=_as_json(n.data, default=None), status=n.status, is_read=bool(n.is_read), created_at=n.created_at.isoformat() if n.created_at else None, ) + + +def _as_json(value: Any, default: Any) -> Any: + """Parse a JSON column that some backends return as a string.""" + if isinstance(value, str): + try: + return json.loads(value) + except (TypeError, ValueError): + pass + try: + return ast.literal_eval(value) + except (ValueError, SyntaxError): + return default + return value if value is not None else default diff --git a/fastapi_admin_kit/notifications/service.py b/fastapi_admin_kit/notifications/service.py index 396ce34..a262956 100644 --- a/fastapi_admin_kit/notifications/service.py +++ b/fastapi_admin_kit/notifications/service.py @@ -27,6 +27,8 @@ from __future__ import annotations +import ast +import json import logging from collections.abc import Sequence from dataclasses import dataclass, field @@ -42,6 +44,20 @@ logger = logging.getLogger("fastapi_admin_kit.notifications") +def _decode_json(value: Any, default: Any) -> Any: + """Parse a JSON column that some backends return as a string.""" + if isinstance(value, str): + try: + return json.loads(value) + except (TypeError, ValueError): + pass + try: + return ast.literal_eval(value) + except (ValueError, SyntaxError): + return default + return value if value is not None else default + + @dataclass class ChannelResult: """Result of delivering a notification over a single channel.""" @@ -411,8 +427,8 @@ async def _push_in_app(self, user_id: str, notif: Any) -> None: "id": notif.id, "title": notif.title, "body": notif.body, - "channels": notif.channels, - "data": notif.data, + "channels": _decode_json(notif.channels, default=[]), + "data": _decode_json(notif.data, default=None), "created_at": (notif.created_at.isoformat() if notif.created_at else None), }, } diff --git a/fastapi_admin_kit/pagination/cursor.py b/fastapi_admin_kit/pagination/cursor.py index 33cfc44..bf06f0c 100644 --- a/fastapi_admin_kit/pagination/cursor.py +++ b/fastapi_admin_kit/pagination/cursor.py @@ -21,8 +21,18 @@ 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)) + """Decode base64 cursor to Python value, or parse as plain value.""" + try: + return json.loads(base64.b64decode(cursor)) + except Exception: + # Fallback: try to parse as plain value (integer, float, etc.) + try: + return int(cursor) + except ValueError: + try: + return float(cursor) + except ValueError: + return cursor def _encode_cursor(self, value: Any) -> str: """Encode Python value to base64 cursor string.""" diff --git a/fastapi_admin_kit/registry/__init__.py b/fastapi_admin_kit/registry/__init__.py index b50a262..02e942f 100644 --- a/fastapi_admin_kit/registry/__init__.py +++ b/fastapi_admin_kit/registry/__init__.py @@ -1,5 +1,9 @@ """Registry package — AdminRegistry and related components.""" -from fastapi_admin_kit.registry.core import AdminRegistry, RegisteredModel +from fastapi_admin_kit.registry.core import ( + AdminRegistry, + RegisteredModel, + build_registered_model, +) -__all__ = ["AdminRegistry", "RegisteredModel"] +__all__ = ["AdminRegistry", "RegisteredModel", "build_registered_model"] diff --git a/fastapi_admin_kit/registry/core.py b/fastapi_admin_kit/registry/core.py index beb9e06..1551f12 100644 --- a/fastapi_admin_kit/registry/core.py +++ b/fastapi_admin_kit/registry/core.py @@ -102,6 +102,74 @@ def get_widget(self, field_name: str, resolver: WidgetResolver | None = None) -> ) +def build_registered_model( + model: type, + admin: ModelAdmin | None = None, +) -> RegisteredModel: + """Build a :class:`RegisteredModel` without writing to the singleton registry. + + Used by the standalone ``ModelAdmin.export_api_route`` / + ``export_admin_route`` helpers so a model's routers can be built without + calling ``admin.register()``. + + Args: + model: A SQLAlchemy declarative model class. + admin: Optional ModelAdmin instance that configures the model. + + Returns: + The built registered model configuration. + + Raises: + ValueError: If the model is not a valid SQLAlchemy model. + """ + from fastapi_admin_kit.views import ModelAdmin as _ModelAdmin + + registry = AdminRegistry() + + # Validate using the injected validator + admin_class = admin.__class__ if admin is not None else None + registry._validator.validate_model_registration(model, admin_class) + + # Inspect using the injected inspector + columns, relationships = registry._inspector.inspect_model(model) + + if admin is None: + admin = _ModelAdmin() + table_name = model.__tablename__ + if admin.verbose_name: + verbose_name = admin.verbose_name + elif getattr(model, "verbose_name", None): + verbose_name = model.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 getattr(model, "verbose_name_plural", None): + verbose_name_plural = model.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" + + return RegisteredModel( + model=model, + admin=admin, + table_name=table_name, + verbose_name=verbose_name, + verbose_name_plural=verbose_name_plural, + columns=columns, + relationships=relationships, + ) + + class AdminRegistry: """Singleton registry for admin models. @@ -167,45 +235,10 @@ def register( # 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 - elif getattr(model, "verbose_name", None): - verbose_name = model.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 getattr(model, "verbose_name_plural", None): - verbose_name_plural = model.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, - ) + registered = build_registered_model(model, admin) - self._models[table_name] = registered + self._models[registered.table_name] = registered return registered def get(self, table_name: str) -> RegisteredModel | None: diff --git a/fastapi_admin_kit/router.py b/fastapi_admin_kit/router.py index 313f09c..05a910b 100644 --- a/fastapi_admin_kit/router.py +++ b/fastapi_admin_kit/router.py @@ -20,8 +20,20 @@ ) -def build_model_router(registered: RegisteredModel) -> APIRouter: - router = APIRouter(prefix=f"/{registered.table_name}") +def build_model_router(registered: RegisteredModel, *, force: bool = False) -> APIRouter | None: + """Build the HTML admin router for a model. + + Returns ``None`` when the model is API-only (``export_endpoint == "api"``) + so callers know no admin router should be mounted. Pass ``force=True`` to + build the router regardless (used by the standalone + ``ModelAdmin.export_admin_route`` helper). + """ + # Models with export_endpoint="api" only expose their JSON API router — + # the admin HTML router is skipped entirely. + if not force and getattr(registered.admin, "export_endpoint", None) == "api": + return None + + router = APIRouter(prefix=f"/{registered.table_name}", tags=[registered.verbose_name]) # DIP: view classes resolved from ModelAdmin config with defaults admin = registered.admin @@ -37,12 +49,14 @@ def build_model_router(registered: RegisteredModel) -> APIRouter: list_v.html_response, methods=["GET"], dependencies=[Depends(require_permission(registered.table_name, "view"))], + include_in_schema=False, ) router.add_api_route( "/create", create_v.html_response, methods=["GET"], dependencies=[Depends(require_permission(registered.table_name, "create"))], + include_in_schema=False, ) router.add_api_route( "/create", @@ -52,9 +66,10 @@ def build_model_router(registered: RegisteredModel) -> APIRouter: Depends(require_permission(registered.table_name, "create")), Depends(require_csrf_token), ], + include_in_schema=False, ) - @router.get("/search") + @router.get("/search", include_in_schema=False) async def search_view( request: Request, q: str = "", @@ -99,11 +114,12 @@ async def search_view( Depends(require_permission(registered.table_name, "edit")), Depends(require_csrf_token), ], + include_in_schema=False, ) # ── Export Endpoint ───────────────────────────────────────────── - @router.get("/export/") + @router.get("/export/", include_in_schema=False) async def export_data( request: Request, format: str = "csv", @@ -192,7 +208,7 @@ async def export_data( # ── Import Endpoint ───────────────────────────────────────────── - @router.post("/import/") + @router.post("/import/", include_in_schema=False) async def import_data( request: Request, format: str = "csv", @@ -309,7 +325,7 @@ async def import_data( html = templates.TemplateResponse(request, "partials/list_table.html", ctx) return html - @router.post("/validate-field") + @router.post("/validate-field", include_in_schema=False) async def validate_field_endpoint( request: Request, _csrf: bool = Depends(require_csrf_token), @@ -356,6 +372,7 @@ async def validate_field_endpoint( edit_v.html_response, methods=["GET"], dependencies=[Depends(require_permission(registered.table_name, "edit"))], + include_in_schema=False, ) router.add_api_route( "/{id}", @@ -365,6 +382,7 @@ async def validate_field_endpoint( Depends(require_permission(registered.table_name, "edit")), Depends(require_csrf_token), ], + include_in_schema=False, ) router.add_api_route( "/{id}/delete", @@ -374,11 +392,12 @@ async def validate_field_endpoint( Depends(require_permission(registered.table_name, "delete")), Depends(require_csrf_token), ], + include_in_schema=False, ) # ── Inline Edit ─────────────────────────────────────────────── - @router.get("/{id}/inline-edit/") + @router.get("/{id}/inline-edit/", include_in_schema=False) async def inline_edit_form( request: Request, id: str, @@ -450,7 +469,7 @@ async def inline_edit_form( }, ) - @router.post("/{id}/inline-edit/") + @router.post("/{id}/inline-edit/", include_in_schema=False) async def inline_edit_save( request: Request, id: str, @@ -606,7 +625,7 @@ async def inline_edit_save( # ── Custom Actions ───────────────────────────────────────────── - @router.post("/action/{action_name}") + @router.post("/action/{action_name}", include_in_schema=False) async def execute_list_action( request: Request, action_name: str, @@ -640,7 +659,7 @@ async def execute_list_action( return HTMLResponse(content="OK") - @router.post("/action/{action_name}/{id}") + @router.post("/action/{action_name}/{id}", include_in_schema=False) async def execute_row_action( request: Request, action_name: str, @@ -674,7 +693,7 @@ async def execute_row_action( # ── Sortable Endpoint ────────────────────────────────────────── - @router.post("/sort") + @router.post("/sort", include_in_schema=False) async def sort_items( request: Request, _csrf: bool = Depends(require_csrf_token), @@ -699,7 +718,7 @@ async def sort_items( # ── Autocomplete Endpoint ────────────────────────────────────── - @router.get("/autocomplete/") + @router.get("/autocomplete/", include_in_schema=False) async def autocomplete( request: Request, q: str = "", @@ -730,7 +749,7 @@ async def autocomplete( return JSONResponse(content=results) # ── Inline Field Update (existing) ───────────────────────────── - @router.patch("/{id}/field") + @router.patch("/{id}/field", include_in_schema=False) async def update_field( request: Request, id: str, diff --git a/fastapi_admin_kit/templates/partials/topbar.html b/fastapi_admin_kit/templates/partials/topbar.html index 08984e0..b9d5a79 100644 --- a/fastapi_admin_kit/templates/partials/topbar.html +++ b/fastapi_admin_kit/templates/partials/topbar.html @@ -97,7 +97,7 @@

Notifications

- View all + View all
diff --git a/fastapi_admin_kit/views.py b/fastapi_admin_kit/views.py index 1b263f5..12bd2d3 100644 --- a/fastapi_admin_kit/views.py +++ b/fastapi_admin_kit/views.py @@ -94,24 +94,29 @@ def create_model_router(registered: RegisteredModel) -> APIRouter: """Create all CRUD routes for a registered model.""" router = APIRouter(prefix=f"/{registered.table_name}", tags=[registered.verbose_name]) - router.add_api_route("/", create_list_view(registered), methods=["GET"], name="list") + router.add_api_route( + "/", create_list_view(registered), methods=["GET"], name="list", include_in_schema=False + ) router.add_api_route( "/create", create_create_view(registered), methods=["GET", "POST"], name="create", + include_in_schema=False, ) router.add_api_route( "/{item_id}", create_edit_view(registered), methods=["GET", "POST"], name="edit", + include_in_schema=False, ) router.add_api_route( "/{item_id}/delete", create_delete_view(registered), methods=["POST"], name="delete", + include_in_schema=False, ) return router diff --git a/fastapi_admin_kit/views/class_views.py b/fastapi_admin_kit/views/class_views.py index 5fd3be2..ccd4133 100644 --- a/fastapi_admin_kit/views/class_views.py +++ b/fastapi_admin_kit/views/class_views.py @@ -94,14 +94,30 @@ def _get_extra_context(self, request: Request) -> dict[str, Any]: def _serialize(self, obj: Any) -> dict[str, Any]: """Serialize an object to a dict using registered columns.""" + from fastapi_admin_kit.audit.diff import serialize_value + if self.api_renderer and hasattr(self.api_renderer, "serialize"): return self.api_renderer.serialize(obj) item_dict: dict[str, Any] = {"id": getattr(obj, "id", None)} for col in self.registered.columns: if col.name != "id": - item_dict[col.name] = str(getattr(obj, col.name, "")) + item_dict[col.name] = serialize_value(getattr(obj, col.name, None)) return item_dict + async def _serialize_for_api(self, obj: Any, request: Request) -> dict[str, Any]: + """Refresh *obj* inside the async greenlet, then serialize it. + + After ``flush()``/``commit()`` server-side defaults (e.g. columns with + ``onupdate``/``server_default``) may be left expired by SQLAlchemy. + ``serialize()`` is synchronous, so touching an expired attribute there + triggers a lazy load outside the greenlet and raises + ``MissingGreenlet``. Reloading first guarantees every attribute is + populated before the sync ``serialize()`` reads it. + """ + session = get_db_session(request) + await session.refresh(obj) + return self._serialize(obj) + async def html_response(self, request: Request) -> Response: raise NotImplementedError @@ -281,7 +297,7 @@ async def api_response( next_cursor, has_next, pagination_mode, - ) = await self.query_provider.get_list(request, q, page) + ) = await self.query_provider.get_list(request, q, page, order=order) item_list = [self._serialize(item) for item in items] return await self.api_renderer.render( request, @@ -633,6 +649,38 @@ async def api_response(self, request: Request) -> Any: if errors: raise HTTPException(status_code=422, detail=errors) session = get_db_session(request) + # Validate related model IDs exist + from fastapi_admin_kit.inspection import validate_related_id + + for rel in self.registered.relationships: + raw_val = parsed.get(rel.name) + if raw_val is None: + continue + # Skip many-to-many relationships — they are handled separately + # by extract_m2m/apply_m2m, not by per-ID validation. + if rel.direction == "MANYTOMANY": + continue + # Validate the related ID exists + pk_value, error = await validate_related_id( + model=self.registered.model, + related_model=rel.target_model, + id_value=raw_val, + field_name=rel.name, + request=request, + ) + if error: + raise HTTPException( + status_code=422, + detail=[ + { + "loc": ["body", rel.name], + "msg": error["msg"], + "type": "value_error", + "input": error["input"], + "ctx": error["ctx"], + } + ], + ) m2m_data = self.model_saver.extract_m2m(self.registered.model, parsed, request) resolved = self.model_saver.resolve_rel_keys(parsed, request) resolved = self.admin.prepare_create_data(resolved, request) @@ -649,7 +697,7 @@ async def api_response(self, request: Request) -> Any: obj=obj, ) await flush_pending_perm_ops(request) - return await self.api_renderer.render(request, self._serialize(obj)) + return await self.api_renderer.render(request, await self._serialize_for_api(obj, request)) class EditView(BaseView): @@ -1155,6 +1203,38 @@ async def api_response( # PUT parser = JSONBodyParser(self.registered) parsed, _ = await parser.parse(request, obj) + # Validate related model IDs exist + from fastapi_admin_kit.inspection import validate_related_id + + for rel in self.registered.relationships: + raw_val = parsed.get(rel.name) + if raw_val is None: + continue + # Skip many-to-many relationships — they are handled separately + # by extract_m2m/apply_m2m, not by per-ID validation. + if rel.direction == "MANYTOMANY": + continue + # Validate the related ID exists + pk_value, error = await validate_related_id( + model=self.registered.model, + related_model=rel.target_model, + id_value=raw_val, + field_name=rel.name, + request=request, + ) + if error: + raise HTTPException( + status_code=422, + detail=[ + { + "loc": ["body", rel.name], + "msg": error["msg"], + "type": "value_error", + "input": error["input"], + "ctx": error["ctx"], + } + ], + ) try: m2m_data = self.model_saver.extract_m2m(obj, parsed, request) self.model_saver.apply_parsed(obj, parsed, request) @@ -1174,7 +1254,7 @@ async def api_response( session = get_db_session(request) await session.rollback() raise - return self._serialize(obj) + return await self._serialize_for_api(obj, request) class DeleteView(BaseView): diff --git a/fastapi_admin_kit/views/profile.py b/fastapi_admin_kit/views/profile.py index 5665f2d..1506d65 100644 --- a/fastapi_admin_kit/views/profile.py +++ b/fastapi_admin_kit/views/profile.py @@ -17,7 +17,7 @@ router = APIRouter() -@router.get("/profile", response_class=HTMLResponse) +@router.get("/profile", response_class=HTMLResponse, include_in_schema=False) async def profile_view( request: Request, user: AdminUserProtocol = Depends(get_current_admin_user), @@ -106,7 +106,7 @@ async def profile_update( ) -@router.get("/profile/password", response_class=HTMLResponse) +@router.get("/profile/password", response_class=HTMLResponse, include_in_schema=False) async def password_change_view( request: Request, _: AdminUserProtocol = Depends(get_current_admin_user), diff --git a/fastapi_admin_kit/views/protocols.py b/fastapi_admin_kit/views/protocols.py index 3bfb3a1..926c500 100644 --- a/fastapi_admin_kit/views/protocols.py +++ b/fastapi_admin_kit/views/protocols.py @@ -17,7 +17,7 @@ class QueryProvider(Protocol): """Single responsibility: build and execute database queries.""" async def get_list( - self, request: Request, q: str, page: int + self, request: Request, q: str, page: int, order: str = "" ) -> tuple[list[Any], int, int, int]: """Return (items, total, page, per_page).""" ... diff --git a/fastapi_admin_kit/views/renderers.py b/fastapi_admin_kit/views/renderers.py index 4258756..31873e7 100644 --- a/fastapi_admin_kit/views/renderers.py +++ b/fastapi_admin_kit/views/renderers.py @@ -11,6 +11,7 @@ from fastapi import Request from fastapi.responses import Response +from fastapi_admin_kit.audit.diff import serialize_value from fastapi_admin_kit.auth.dependencies import ( resolve_permission_checker as _resolve_permission_checker, # noqa: F401 ) @@ -154,7 +155,7 @@ def serialize(self, obj: Any) -> dict[str, Any]: item_dict: dict[str, Any] = {"id": getattr(obj, "id", None)} for col in self.registered.columns: if col.name != "id": - item_dict[col.name] = str(getattr(obj, col.name, "")) + item_dict[col.name] = serialize_value(getattr(obj, col.name, None)) return item_dict async def render(self, request: Request, data: Any) -> Any: @@ -243,7 +244,11 @@ async def parse( ) -> tuple[dict[str, Any], dict[str, list[str]]]: from sqlalchemy import inspect as sa_inspect - body = await request.json() + # Pre-parsed body supplied by the API wrapper handler (so FastAPI can + # document the request body schema in Swagger/OpenAPI). + body = getattr(request.state, "_api_payload", None) + if body is None: + body = await request.json() valid_fields = {col.name for col in self.registered.columns} # Relationship keys (FK / many-to-many) are handled separately by the # view (resolved to FK columns or applied as m2m collections), so they @@ -348,7 +353,7 @@ def _build_filter_clauses( return clauses async def get_list( - self, request: Request, q: str = "", page: int = 1 + self, request: Request, q: str = "", page: int = 1, order: str = "" ) -> tuple[list[Any], int, int, int]: """Execute list query with filtering, search, pagination. @@ -511,7 +516,7 @@ async def get_list( if q and registered.admin.search_fields: base = apply_search_filter(request, base, model, registered.admin.search_fields, q) - query_ordering = request.query_params.get("ordering", "") + query_ordering = request.query_params.get("ordering", "") or order order = registered.admin.get_ordering( {"ordering": query_ordering}, registered.admin.ordering ) @@ -534,7 +539,7 @@ async def get_list( base = base.order_by(desc(col) if order[0].startswith("-") else asc(col)) - per_page = registered.admin.per_page + per_page = int(request.query_params.get("per_page", registered.admin.per_page)) from fastapi_admin_kit.pagination import ( OffsetPagination, diff --git a/fastapi_admin_kit/views/roles.py b/fastapi_admin_kit/views/roles.py index a1b421a..39114c5 100644 --- a/fastapi_admin_kit/views/roles.py +++ b/fastapi_admin_kit/views/roles.py @@ -75,7 +75,7 @@ async def permissions_search( ) -@router.get("/roles", response_class=HTMLResponse) +@router.get("/roles", response_class=HTMLResponse, include_in_schema=False) async def role_list_view( request: Request, _: AdminUserProtocol = Depends(require_superuser), @@ -108,7 +108,7 @@ async def role_list_view( ) -@router.get("/roles/create", response_class=HTMLResponse) +@router.get("/roles/create", response_class=HTMLResponse, include_in_schema=False) async def role_create_view( request: Request, _: AdminUserProtocol = Depends(require_superuser), @@ -132,7 +132,7 @@ async def role_create_view( ) -@router.post("/roles", response_class=RedirectResponse) +@router.post("/roles", response_class=RedirectResponse, include_in_schema=False) async def role_create_save_view( request: Request, _: AdminUserProtocol = Depends(require_superuser), @@ -179,7 +179,7 @@ async def role_create_save_view( ) -@router.get("/roles/{role_id}", response_class=HTMLResponse) +@router.get("/roles/{role_id}", response_class=HTMLResponse, include_in_schema=False) async def role_edit_view( request: Request, role_id: int, @@ -213,7 +213,7 @@ async def role_edit_view( ) -@router.post("/roles/{role_id}", response_class=RedirectResponse) +@router.post("/roles/{role_id}", response_class=RedirectResponse, include_in_schema=False) async def role_save_view( request: Request, role_id: int, @@ -257,7 +257,7 @@ async def role_save_view( ) -@router.post("/roles/{role_id}/delete", response_class=RedirectResponse) +@router.post("/roles/{role_id}/delete", response_class=RedirectResponse, include_in_schema=False) async def role_delete_view( request: Request, role_id: int, diff --git a/fastapi_admin_kit/views/settings.py b/fastapi_admin_kit/views/settings.py index 53cd20b..bf1518e 100644 --- a/fastapi_admin_kit/views/settings.py +++ b/fastapi_admin_kit/views/settings.py @@ -12,7 +12,7 @@ router = APIRouter() -@router.get("/settings/theme", response_class=HTMLResponse) +@router.get("/settings/theme", response_class=HTMLResponse, include_in_schema=False) async def theme_settings( request: Request, current_user: Any = Depends(get_current_admin_user), diff --git a/fastapi_admin_kit/views/totp.py b/fastapi_admin_kit/views/totp.py index 0ce3f96..4f4e6f9 100644 --- a/fastapi_admin_kit/views/totp.py +++ b/fastapi_admin_kit/views/totp.py @@ -25,7 +25,7 @@ router = APIRouter() -@router.get("/profile/2fa", response_class=HTMLResponse) +@router.get("/profile/2fa", response_class=HTMLResponse, include_in_schema=False) async def totp_setup_view( request: Request, user: AdminUserProtocol = Depends(get_current_admin_user), @@ -228,7 +228,7 @@ async def totp_regenerate_backup_codes( ) -@router.get("/verify-2fa", response_class=HTMLResponse) +@router.get("/verify-2fa", response_class=HTMLResponse, include_in_schema=False) async def totp_verify_view( request: Request, temp_token: str | None = None, diff --git a/tests/test_api_crud.py b/tests/test_api_crud.py new file mode 100644 index 0000000..3123c92 --- /dev/null +++ b/tests/test_api_crud.py @@ -0,0 +1,239 @@ +"""Integration tests for the JSON API CRUD endpoints (PATCH partial updates).""" + +from __future__ import annotations + +import base64 +import os +import tempfile + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import Column, DateTime, Integer, String, create_engine, func +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.orm import declarative_base +from sqlalchemy.pool import StaticPool + +from fastapi_admin_kit import Admin +from fastapi_admin_kit.auth.backend import BuiltinAuthBackend +from fastapi_admin_kit.migrations.models import Role, User +from fastapi_admin_kit.models.base import Base as AdminBase +from tests.conftest import SECRET_KEY, run_async +from tests.test_registry import Product + + +@pytest.fixture(autouse=True) +def _clear_registry(): + from fastapi_admin_kit.registry import AdminRegistry + + AdminRegistry().clear() + yield + AdminRegistry().clear() + + +@pytest.fixture +def engine(): + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + sync_engine = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False}) + AdminBase.metadata.create_all(sync_engine) + Product.metadata.create_all(sync_engine) + sync_engine.dispose() + async_engine = create_async_engine( + f"sqlite+aiosqlite:///{path}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + yield async_engine + run_async(async_engine.dispose()) + os.unlink(path) + + +@pytest.fixture +def client(engine): + async def _seed(): + async with AsyncSession(engine) as session: + role = Role(name="SuperAdmin") + session.add(role) + await session.flush() + user = User( + email="test@example.com", + hashed_password="$2b$12$DOXzSwSZYp0Y1pTzEvWjO.KOLQg3wA/Ez1RkN4RHMiLqngoLM2lMG", + full_name="Test User", + is_superuser=True, + is_active=True, + ) + user.roles.append(role) + session.add(user) + await session.commit() + + run_async(_seed()) + + admin = Admin( + engine=engine, + auth_model=User, + auth_backend=BuiltinAuthBackend(), + secret_key=SECRET_KEY, + auto_discover=False, + session_secure=False, + ) + admin.register(Product) + app = FastAPI() + run_async(admin.setup(app)) + return TestClient(app) + + +@pytest.fixture +def product(engine): + async def _create(): + async with AsyncSession(engine) as session: + p = Product(name="Initial", price=10, is_active=True) + session.add(p) + await session.commit() + await session.refresh(p) + return p + + return run_async(_create()) + + +@pytest.fixture +def auth_headers(client): + creds = base64.b64encode(b"test@example.com:secret").decode() + token = client.post("/api/auth/token", headers={"Authorization": f"Basic {creds}"}).json()[ + "access_token" + ] + return {"Authorization": f"Bearer {token}"} + + +def test_patch_endpoint_registered(client): + schema = client.get("/openapi.json").json() + methods = set(schema["paths"]["/api/products/{item_id}"]) + assert "patch" in methods + assert "put" in methods + + +def test_patch_partially_updates_only_provided_fields(client, product, auth_headers): + resp = client.patch(f"/api/products/{product.id}", headers=auth_headers, json={"price": 99}) + assert resp.status_code == 200 + body = resp.json() + assert body["price"] == 99 + assert body["name"] == "Initial" + + get_resp = client.get(f"/api/products/{product.id}", headers=auth_headers) + assert get_resp.json()["price"] == 99 + assert get_resp.json()["name"] == "Initial" + + +def test_patch_missing_item_404(client, auth_headers): + resp = client.patch("/api/products/99999", headers=auth_headers, json={"price": 1}) + assert resp.status_code == 404 + + +# A model whose row has a server-side ``onupdate`` timestamp. SQLAlchemy +# leaves such columns expired after ``flush()``; serializing the object then +# used to trigger a lazy load outside the async greenlet (MissingGreenlet). +_OnUpdateBase = declarative_base() + + +class OnUpdateModel(_OnUpdateBase): + __tablename__ = "onupdate_models" + + id = Column(Integer, primary_key=True) + name = Column(String(200), nullable=False) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +@pytest.fixture +def onupdate_engine(): + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + sync_engine = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False}) + AdminBase.metadata.create_all(sync_engine) + _OnUpdateBase.metadata.create_all(sync_engine) + sync_engine.dispose() + async_engine = create_async_engine( + f"sqlite+aiosqlite:///{path}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + yield async_engine + run_async(async_engine.dispose()) + os.unlink(path) + + +@pytest.fixture +def onupdate_client(onupdate_engine): + async def _seed(): + async with AsyncSession(onupdate_engine) as session: + user = User( + email="upd@example.com", + hashed_password="$2b$12$DOXzSwSZYp0Y1pTzEvWjO.KOLQg3wA/Ez1RkN4RHMiLqngoLM2lMG", + full_name="Upd User", + is_superuser=True, + is_active=True, + ) + session.add(user) + await session.commit() + + run_async(_seed()) + + admin = Admin( + engine=onupdate_engine, + auth_model=User, + auth_backend=BuiltinAuthBackend(), + secret_key=SECRET_KEY, + auto_discover=False, + session_secure=False, + ) + admin.register(OnUpdateModel) + app = FastAPI() + run_async(admin.setup(app)) + return TestClient(app) + + +@pytest.fixture +def onupdate_obj(onupdate_engine): + async def _create(): + async with AsyncSession(onupdate_engine) as session: + p = OnUpdateModel(name="Initial") + session.add(p) + await session.commit() + await session.refresh(p) + return p + + return run_async(_create()) + + +@pytest.fixture +def onupdate_headers(onupdate_client): + creds = base64.b64encode(b"upd@example.com:secret").decode() + token = onupdate_client.post( + "/api/auth/token", headers={"Authorization": f"Basic {creds}"} + ).json()["access_token"] + return {"Authorization": f"Bearer {token}"} + + +def test_patch_with_server_side_onupdate_serializes_without_missing_greenlet( + onupdate_client, onupdate_obj, onupdate_headers +): + """PATCH on a model with an ``onupdate`` timestamp must not raise MissingGreenlet. + + Regression test: after ``flush()`` SQLAlchemy leaves the ``updated_at`` + column expired, and the synchronous serializer used to trigger a lazy load + outside the async greenlet. + """ + resp = onupdate_client.patch( + f"/api/onupdate_models/{onupdate_obj.id}", + headers=onupdate_headers, + json={"name": "Renamed"}, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["name"] == "Renamed" + assert body["updated_at"] is not None + + get_resp = onupdate_client.get( + f"/api/onupdate_models/{onupdate_obj.id}", headers=onupdate_headers + ) + assert get_resp.status_code == 200 + assert get_resp.json()["name"] == "Renamed" diff --git a/tests/test_api_security.py b/tests/test_api_security.py new file mode 100644 index 0000000..c1a581e --- /dev/null +++ b/tests/test_api_security.py @@ -0,0 +1,137 @@ +"""Integration tests — /api/auth/token accepts HTTP Basic creds; Swagger +documents both Basic and Bearer security schemes.""" + +from __future__ import annotations + +import base64 +import os +import tempfile + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine +from sqlalchemy.pool import StaticPool + +from fastapi_admin_kit import Admin +from fastapi_admin_kit.auth.backend import BuiltinAuthBackend +from fastapi_admin_kit.migrations.models import Role, User +from fastapi_admin_kit.models import Base +from tests.conftest import SECRET_KEY, run_async + + +@pytest.fixture(autouse=True) +def _clear_registry(): + from fastapi_admin_kit.registry import AdminRegistry + + AdminRegistry().clear() + yield + AdminRegistry().clear() + + +@pytest.fixture +def engine(): + fd, path = tempfile.mkstemp() + os.close(fd) + sync_engine = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False}) + Base.metadata.create_all(sync_engine) + sync_engine.dispose() + async_engine = create_async_engine( + f"sqlite+aiosqlite:///{path}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + yield async_engine + run_async(async_engine.dispose()) + os.unlink(path) + + +@pytest.fixture +def client(engine): + async def _seed(): + async with AsyncSession(engine) as session: + role = Role(name="SuperAdmin") + session.add(role) + await session.flush() + user = User( + email="test@example.com", + hashed_password="$2b$12$DOXzSwSZYp0Y1pTzEvWjO.KOLQg3wA/Ez1RkN4RHMiLqngoLM2lMG", + full_name="Test User", + is_superuser=True, + is_active=True, + ) + user.roles.append(role) + session.add(user) + await session.commit() + + run_async(_seed()) + + admin = Admin( + engine=engine, + auth_model=User, + auth_backend=BuiltinAuthBackend(), + secret_key=SECRET_KEY, + auto_discover=False, + session_secure=False, + ) + app = FastAPI() + run_async(admin.setup(app)) + return TestClient(app) + + +def test_openapi_exposes_basic_and_bearer_schemes(client): + schema = client.get("/openapi.json").json() + schemes = schema["components"]["securitySchemes"] + + assert schemes["BasicAuth"]["type"] == "http" + assert schemes["BasicAuth"]["scheme"] == "basic" + assert schemes["BearerAuth"]["type"] == "http" + assert schemes["BearerAuth"]["scheme"] == "bearer" + + # Token endpoint requires Basic; protected CRUD/roles require Bearer. + assert schema["paths"]["/api/auth/token"]["post"]["security"] == [{"BasicAuth": []}] + assert schema["paths"]["/api/roles/"]["get"]["security"] == [{"BearerAuth": []}] + + +def test_token_endpoint_accepts_basic_auth(client): + creds = base64.b64encode(b"test@example.com:secret").decode() + response = client.post("/api/auth/token", headers={"Authorization": f"Basic {creds}"}) + assert response.status_code == 200 + body = response.json() + assert body["access_token"] + assert body["token_type"] == "bearer" + + +def test_token_endpoint_rejects_bad_basic_auth(client): + creds = base64.b64encode(b"test@example.com:wrong").decode() + response = client.post("/api/auth/token", headers={"Authorization": f"Basic {creds}"}) + assert response.status_code == 401 + + +def test_token_endpoint_still_accepts_json_body(client): + response = client.post( + "/api/auth/token", + json={"email": "test@example.com", "password": "secret"}, + ) + assert response.status_code == 200 + assert response.json()["access_token"] + + +def test_token_endpoint_without_credentials_is_422(client): + response = client.post("/api/auth/token") + assert response.status_code == 422 + + +def test_protected_route_without_bearer_is_401(client): + response = client.get("/api/roles/") + assert response.status_code == 401 + + +def test_bearer_token_grants_access_to_protected_route(client): + creds = base64.b64encode(b"test@example.com:secret").decode() + token_resp = client.post("/api/auth/token", headers={"Authorization": f"Basic {creds}"}) + token = token_resp.json()["access_token"] + + response = client.get("/api/roles/", headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 200 diff --git a/tests/test_export_endpoint.py b/tests/test_export_endpoint.py new file mode 100644 index 0000000..b457bed --- /dev/null +++ b/tests/test_export_endpoint.py @@ -0,0 +1,170 @@ +"""Tests for per-model endpoint export control (export_endpoint) and standalone router export.""" + +from __future__ import annotations + +import os +import tempfile + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.ext.asyncio import create_async_engine +from sqlalchemy.pool import StaticPool + +from fastapi_admin_kit import Admin +from fastapi_admin_kit.auth.backend import BuiltinAuthBackend +from fastapi_admin_kit.migrations.models import User +from fastapi_admin_kit.modeladmin import ModelAdmin +from fastapi_admin_kit.models.base import Base as AdminBase +from tests.conftest import SECRET_KEY, run_async +from tests.test_registry import Product + + +@pytest.fixture(autouse=True) +def _clear_registry(): + from fastapi_admin_kit.registry import AdminRegistry + + AdminRegistry().clear() + yield + AdminRegistry().clear() + + +@pytest.fixture +def engine(): + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + sync_engine = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False}) + AdminBase.metadata.create_all(sync_engine) + Product.metadata.create_all(sync_engine) + sync_engine.dispose() + async_engine = create_async_engine( + f"sqlite+aiosqlite:///{path}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + yield async_engine + run_async(async_engine.dispose()) + os.unlink(path) + + +def _build_app(engine, admin_class): + """Build a fully-setup FastAPI app with *admin_class* registered for Product.""" + admin = Admin( + engine=engine, + auth_model=User, + auth_backend=BuiltinAuthBackend(), + secret_key=SECRET_KEY, + auto_discover=False, + session_secure=False, + ) + admin.register(Product, admin_class) + app = FastAPI() + run_async(admin.setup(app)) + return TestClient(app) + + +def _openapi_paths(client) -> dict: + return client.get("/openapi.json").json()["paths"] + + +class TestDefaultExportEndpoint: + def test_none_builds_both_routers_but_only_api_in_schema(self, engine): + class ProductAdmin(ModelAdmin): + pass + + client = _build_app(engine, ProductAdmin) + paths = _openapi_paths(client) + + # JSON API routes are present in the schema. + assert "/api/products" in paths + assert "/api/products/{item_id}" in paths + + # Admin HTML model routes exist at runtime… + resp = client.get("/admin/products/") + assert resp.status_code != 404 + + # …but never leak into the OpenAPI schema. + assert not any(p.startswith("/admin/products") for p in paths) + + +class TestApiOnlyExportEndpoint: + def test_api_excludes_admin_routes(self, engine): + class ProductAdmin(ModelAdmin): + export_endpoint = "api" + + client = _build_app(engine, ProductAdmin) + paths = _openapi_paths(client) + + # JSON API routes are present. + assert "/api/products" in paths + assert "/api/products/{item_id}" in paths + + # No admin HTML routes for the model — neither mounted nor in schema. + assert not any(p.startswith("/admin/products") for p in paths) + assert client.get("/admin/products/").status_code == 404 + + +class TestHtmlOnlyExportEndpoint: + def test_html_excludes_api_routes(self, engine): + class ProductAdmin(ModelAdmin): + export_endpoint = "html" + + client = _build_app(engine, ProductAdmin) + paths = _openapi_paths(client) + + # No JSON API routes for the model. + assert not any(p.startswith("/api/products") for p in paths) + + # Admin HTML model routes are mounted… + assert client.get("/admin/products/").status_code != 404 + + # …but hidden from the OpenAPI schema. + assert not any(p.startswith("/admin/products") for p in paths) + + +class TestStandaloneExport: + def test_export_api_route_works_without_register(self): + class ProductAdmin(ModelAdmin): + export_endpoint = "api" + + router = ProductAdmin().export_api_route(Product) + assert router is not None + + app = FastAPI() + app.include_router(router) + paths = app.openapi()["paths"] + assert "/products" in paths + assert "/products/{item_id}" in paths + + def test_export_api_route_with_prefix(self): + class ProductAdmin(ModelAdmin): + pass + + router = ProductAdmin().export_api_route(Product, prefix="/api") + app = FastAPI() + app.include_router(router) + paths = app.openapi()["paths"] + assert "/api/products" in paths + + def test_export_admin_route_works_without_register(self): + class ProductAdmin(ModelAdmin): + export_endpoint = "api" + + router = ProductAdmin().export_admin_route(Product, prefix="/admin") + assert router is not None + + app = FastAPI() + app.include_router(router) + # Admin HTML routes are mounted but hidden from the schema entirely. + assert app.openapi()["paths"] == {} + + def test_standalone_export_does_not_write_to_registry(self): + from fastapi_admin_kit.registry import AdminRegistry + + class ProductAdmin(ModelAdmin): + export_endpoint = "api" + + ProductAdmin().export_api_route(Product) + ProductAdmin().export_admin_route(Product) + assert AdminRegistry().get("products") is None diff --git a/tests/test_jwt.py b/tests/test_jwt.py index 90181bc..4102833 100644 --- a/tests/test_jwt.py +++ b/tests/test_jwt.py @@ -120,3 +120,28 @@ def test_wrong_secret_returns_none(self): token = pyjwt.encode(payload, secret, algorithm="HS256") result = decode_access_token(token, "wrong-secret") assert result is None + + +class TestOpenAPISecuritySchemes: + """Swagger docs expose Basic + Bearer auth for the JSON API.""" + + def _schemes(self): + from fastapi_admin_kit.api.security import basic_scheme, bearer_scheme + + return { + "basic": { + "name": basic_scheme.scheme_name, + "auto_error": basic_scheme.auto_error, + }, + "bearer": { + "name": bearer_scheme.scheme_name, + "auto_error": bearer_scheme.auto_error, + }, + } + + def test_schemes_are_purely_documentational(self): + schemes = self._schemes() + assert schemes["basic"]["name"] == "BasicAuth" + assert schemes["bearer"]["name"] == "BearerAuth" + assert schemes["basic"]["auto_error"] is False + assert schemes["bearer"]["auto_error"] is False diff --git a/tests/test_notifications_admin_integration.py b/tests/test_notifications_admin_integration.py index 6956ffc..229311c 100644 --- a/tests/test_notifications_admin_integration.py +++ b/tests/test_notifications_admin_integration.py @@ -195,7 +195,7 @@ def test_admin_paths_synced_to_mount_prefix(api_prefix_app, admin_user): """The admin template/JS must point at the real mount prefix, not /admin/notifications.""" app, admin = api_prefix_app assert admin.config.notifications_api_path == "/api/notifications" - assert admin.config.notifications_list_path == "/api/notifications/" + assert admin.config.notifications_list_path == "/admin/admin_notifications/" client = TestClient(app) client.cookies.set("admin_session", create_session_cookie(admin_user.id)) @@ -232,7 +232,7 @@ def test_explicit_admin_path_respected(engine, async_session_factory, admin_user os.environ.pop("SKIP_CREATE_TABLES", None) assert admin.config.notifications_api_path == "/custom/notifications" - assert admin.config.notifications_list_path == "/custom/notifications/" + assert admin.config.notifications_list_path == "/admin/admin_notifications/" # --------------------------------------------------------------------------- diff --git a/tests/test_user_management.py b/tests/test_user_management.py index 167f67b..6302267 100644 --- a/tests/test_user_management.py +++ b/tests/test_user_management.py @@ -52,3 +52,124 @@ def test_soft_delete_sets_inactive(self): assert user.is_active is True user.is_active = False assert user.is_active is False + + +class TestDirectPermissionRemoval: + """Regression — removing all direct permissions from a user must persist. + + Previously ``perm_data`` was forwarded from the form only when it was + truthy; an empty list (all permissions removed via the cross button) was + treated as "no change", so stale ``UserPermission`` rows survived the save + and the user kept access. + """ + + def test_removing_all_direct_permissions_clears_rows(self): + import os + import tempfile + + from fastapi import FastAPI + from fastapi.testclient import TestClient + from sqlalchemy import create_engine + from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + from sqlalchemy.pool import StaticPool + + from fastapi_admin_kit import Admin + from fastapi_admin_kit.auth.backend import BuiltinAuthBackend + from fastapi_admin_kit.auth.csrf import generate_csrf_token + from fastapi_admin_kit.migrations.models import Permission, User, UserPermission + from fastapi_admin_kit.models.base import Base as AdminBase + from tests.conftest import SECRET_KEY, create_session_cookie, run_async + + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + sync_engine = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False}) + AdminBase.metadata.create_all(sync_engine) + sync_engine.dispose() + async_engine = create_async_engine( + f"sqlite+aiosqlite:///{path}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + + async def _seed(): + async with AsyncSession(async_engine, expire_on_commit=False) as session: + admin = User( + email="admin@test.com", + hashed_password="$2b$12$HQlaDF1uaZvpsppxtnwD5uXp1VxiNXsiS5OCEkXRn7G0xNjUEo8cG", + full_name="Admin", + is_superuser=True, + is_active=True, + ) + session.add(admin) + target = User( + email="target@test.com", + hashed_password="hashed", + full_name="Target", + is_superuser=False, + is_active=True, + ) + session.add(target) + await session.flush() + perm = Permission( + name="products_view", + table_name="products", + can_view=True, + ) + session.add(perm) + await session.flush() + session.add(UserPermission(user_id=target.id, permission_id=perm.id)) + await session.commit() + return admin.id, target.id + + admin_id, target_id = run_async(_seed()) + + app = FastAPI() + admin = Admin( + app=app, + engine=async_engine, + secret_key=SECRET_KEY, + auth_backend=BuiltinAuthBackend(), + auto_discover=False, + ) + run_async(admin.setup(app)) + client = TestClient(app) + + csrf_token = generate_csrf_token(SECRET_KEY) + cookie = create_session_cookie(admin_id) + resp = client.post( + f"/admin/admin_users/{target_id}", + data={ + "email": "target@test.com", + "full_name": "Target", + "password": "", + "role_ids": "[]", + "perm_data": "[]", + "is_superuser": "", + "is_active": "on", + "csrf_token": csrf_token, + }, + cookies={"admin_session": cookie, "admin_csrf_token": csrf_token}, + follow_redirects=False, + ) + assert resp.status_code == 303 + + async def _remaining_rows(): + from sqlalchemy import select + + async with AsyncSession(async_engine, expire_on_commit=False) as session: + rows = ( + ( + await session.execute( + select(UserPermission).where(UserPermission.user_id == target_id) + ) + ) + .scalars() + .all() + ) + return list(rows) + + rows = run_async(_remaining_rows()) + assert rows == [] + + run_async(async_engine.dispose()) + os.unlink(path)