diff --git a/docs/guide/features.md b/docs/guide/features.md index 50ef0b3..7981ff6 100644 --- a/docs/guide/features.md +++ b/docs/guide/features.md @@ -149,6 +149,7 @@ Everything FastAPI Admin Kit offers, in one place. | Ordering | Default sort order with clickable column headers | [Model Registration](model-registration.md) | | Pagination | Offset, cursor, or dynamic strategies per model | [Pagination](pagination.md) | | Custom Columns | `@column` decorator for computed columns with formatting, icons, width | [Model Registration](model-registration.md) | +| Custom Endpoints | `@endpoint` decorator for arbitrary FastAPI routes under `//...` with RBAC | [Model Registration](model-registration.md) | | Column Export | CSV export support per column | [Model Registration](model-registration.md) | --- diff --git a/docs/guide/model-registration.md b/docs/guide/model-registration.md index d4e4d62..cb136a6 100644 --- a/docs/guide/model-registration.md +++ b/docs/guide/model-registration.md @@ -400,6 +400,73 @@ class ProductAdmin(ModelAdmin): | `exportable` | `bool` | Include in CSV export (default: `True`) | | `icon` | `str` | Material icon name | +## Custom API Endpoints + +Use the `@endpoint()` decorator to add arbitrary FastAPI routes to a model's +admin router. Every endpoint is served under `//` (mounted +under the admin prefix), so a `health-check` endpoint on the `Product` model +is reachable at `/admin/products/health-check`. + +```python +from fastapi_admin_kit import endpoint + +@admin.register(Product) +class ProductAdmin(ModelAdmin): + @endpoint( + path="/health-check", + methods=["GET"], + tags=["monitoring"], + description="Health check endpoint", + summary="Health summary", + response_description="Healthy status", + permission="view", + ) + async def health_check(self, request): + return {"status": "healthy"} + + @endpoint(path="/stats", methods=["GET"]) + async def stats(self, request, days: int = 7): + return {"stats": {"days": days}} +``` + +The `request` parameter is injected automatically (annotated `Request` or a +bare `request` argument both work); other parameters are handled by FastAPI as +usual (query/path params, bodies, etc.). + +### Endpoint Options + +| Option | Type | Required | Default | Description | +|--------|------|----------|---------|-------------| +| `path` | `str` | Yes | - | Path within the model router (e.g. `/health-check`) | +| `methods` | `list[str]` | No | `["GET"]` | HTTP methods to expose | +| `tags` | `list[str]` | No | `[]` | OpenAPI tags (appended to the model tag) | +| `description` | `str` | No | `""` | Route description | +| `name` | `str` | No | auto-generated | Route name (defaults to `_`) | +| `dependencies` | `list[Any]` | No | `[]` | Extra FastAPI `Depends()` dependencies | +| `status_code` | `int` | No | `200` | Response status code | +| `response_model` | `Type[BaseModel]` | No | `None` | Response schema for validation/docs | +| `summary` | `str` | No | `""` | OpenAPI summary | +| `response_description` | `str` | No | `""` | OpenAPI response description | +| `permission` | `str` | No | `None` | RBAC action enforced via `require_permission` | + +### RBAC + +Setting `permission` enforces the same RBAC used by the built-in routes (a +`require_permission("", "")` dependency). You can also pass +arbitrary dependencies directly: + +```python +from fastapi import Depends +from fastapi_admin_kit.auth.dependencies import require_permission + +@endpoint( + path="/stats", + dependencies=[Depends(require_permission("products", "view"))], +) +async def stats(self, request): + pass +``` + ## Customizing Built-in Admin Models FastAPI Admin Kit ships with default admin classes for built-in models (users, roles, audit logs, etc.). You can customize these by inheriting from the default classes. diff --git a/fastapi_admin_kit/__init__.py b/fastapi_admin_kit/__init__.py index 3881453..97c7c31 100644 --- a/fastapi_admin_kit/__init__.py +++ b/fastapi_admin_kit/__init__.py @@ -1,7 +1,7 @@ """FastAPI Admin Kit — Drop-in admin panel for FastAPI + SQLAlchemy apps.""" from fastapi_admin_kit.admin import Admin -from fastapi_admin_kit.admin.decorators import column +from fastapi_admin_kit.admin.decorators import column, endpoint from fastapi_admin_kit.auth.mixins import AuthModelMixin from fastapi_admin_kit.config import DatabaseConfig, DatabaseType from fastapi_admin_kit.exceptions import ConfigError @@ -74,6 +74,7 @@ "RegisteredModel", "ModelAdmin", "column", + "endpoint", "BuiltNavGroup", "BuiltNavItem", "DefaultSidebarBuilder", diff --git a/fastapi_admin_kit/admin/decorators.py b/fastapi_admin_kit/admin/decorators.py index 78813bf..13b9582 100644 --- a/fastapi_admin_kit/admin/decorators.py +++ b/fastapi_admin_kit/admin/decorators.py @@ -3,7 +3,33 @@ from __future__ import annotations from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from pydantic import BaseModel + + +@dataclass +class EndpointOptions: + """Metadata for @endpoint() decorator.""" + + path: str + methods: list[str] = field(default_factory=lambda: ["GET"]) + tags: list[str] = field(default_factory=list) + description: str = "" + name: str = "" + dependencies: list[Any] = field(default_factory=list) + status_code: int = 200 + response_model: type[BaseModel] | None = None + summary: str = "" + response_description: str = "" + permission: str | None = None + include_in_schema: bool = True + + def __call__(self, func: Callable) -> Callable: + func._admin_endpoint = self + return func @dataclass @@ -68,3 +94,55 @@ def price_display(self, obj): exportable=exportable, icon=icon, ) + + +def endpoint( + path: str, + methods: list[str] | None = None, + tags: list[str] | None = None, + description: str = "", + name: str = "", + dependencies: list[Any] | None = None, + status_code: int = 200, + response_model: type[BaseModel] | None = None, + summary: str = "", + response_description: str = "", + permission: str | None = None, + include_in_schema: bool = True, +) -> EndpointOptions: + """Decorator to register a custom FastAPI endpoint on a ModelAdmin. + + Endpoints are auto-registered on the model's admin router by + ``build_model_router()`` via ``APIRouter.add_api_route()``, keeping full + FastAPI configuration support (path, methods, tags, dependencies, + status code, response model, ...). + + Usage:: + + from fastapi_admin_kit import endpoint + + class ProductAdmin(ModelAdmin): + @endpoint( + path="/health-check", + methods=["GET"], + tags=["monitoring"], + description="Health check endpoint", + permission="view", + ) + async def health_check(self, request): + return {"status": "healthy"} + """ + return EndpointOptions( + path=path, + methods=methods or ["GET"], + tags=tags or [], + description=description, + name=name, + dependencies=dependencies or [], + status_code=status_code, + response_model=response_model, + summary=summary, + response_description=response_description, + permission=permission, + include_in_schema=include_in_schema, + ) diff --git a/fastapi_admin_kit/modeladmin.py b/fastapi_admin_kit/modeladmin.py index a8339b1..d1ec483 100644 --- a/fastapi_admin_kit/modeladmin.py +++ b/fastapi_admin_kit/modeladmin.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Any -from fastapi_admin_kit.admin.decorators import column +from fastapi_admin_kit.admin.decorators import column, endpoint from fastapi_admin_kit.form.types import ExtraField, FieldMeta if TYPE_CHECKING: @@ -113,6 +113,9 @@ def get_ordering(request_params: dict, admin_ordering: list[str] | None) -> list # Decorator for custom column display column = staticmethod(column) + # Decorator for custom FastAPI endpoints + endpoint = staticmethod(endpoint) + # ── Standalone router export (no admin.register required) ─────── def export_api_route(self, model: Any, prefix: str = "") -> Any: diff --git a/fastapi_admin_kit/router.py b/fastapi_admin_kit/router.py index 05a910b..b811e5a 100644 --- a/fastapi_admin_kit/router.py +++ b/fastapi_admin_kit/router.py @@ -20,6 +20,41 @@ ) +def _endpoint_handler(fn): + """Prepare an ``@endpoint`` method for FastAPI registration. + + FastAPI only injects the ``Request`` object into parameters annotated as + ``Request``. To honour the documented ``async def health_check(self, request)`` + usage, any parameter literally named ``request`` without an annotation is + re-annotated as ``Request`` via a thin forwarding wrapper (other parameters, + e.g. query/path params, are preserved). + """ + import inspect + + from fastapi import Request + + sig = inspect.signature(fn) + params = list(sig.parameters.values()) + if not any(p.name == "request" and p.annotation is inspect.Parameter.empty for p in params): + return fn + + new_params = [ + p.replace(annotation=Request) + if p.name == "request" and p.annotation is inspect.Parameter.empty + else p + for p in params + ] + new_sig = sig.replace(parameters=new_params) + + async def wrapper(*args, **kwargs): + return await fn(*args, **kwargs) + + wrapper.__signature__ = new_sig + wrapper.__name__ = getattr(fn, "__name__", "endpoint") + wrapper.__doc__ = getattr(fn, "__doc__", None) + return wrapper + + def build_model_router(registered: RegisteredModel, *, force: bool = False) -> APIRouter | None: """Build the HTML admin router for a model. @@ -367,6 +402,37 @@ async def validate_field_endpoint( }, ) + # ── Custom @endpoint routes ──────────────────────────────────── + # Registered before the ``/{id}`` catch-all so custom paths (e.g. + # ``/health-check``) are never swallowed by the edit-view route. + + for name in dir(admin): + if name.startswith("__"): + continue + fn = getattr(admin, name, None) + opts = getattr(fn, "_admin_endpoint", None) + if opts is None: + continue + + dependencies = list(opts.dependencies or []) + if opts.permission: + dependencies.append(Depends(require_permission(registered.table_name, opts.permission))) + + router.add_api_route( + opts.path, + _endpoint_handler(fn), + methods=opts.methods or ["GET"], + tags=opts.tags or None, + description=opts.description or None, + name=opts.name or f"{registered.table_name}_{name}", + dependencies=dependencies or None, + status_code=opts.status_code, + response_model=opts.response_model, + summary=opts.summary or None, + response_description=opts.response_description or None, + include_in_schema=opts.include_in_schema, + ) + router.add_api_route( "/{id}", edit_v.html_response, diff --git a/fastapi_admin_kit/views.py b/fastapi_admin_kit/views.py index 12bd2d3..fe15c3a 100644 --- a/fastapi_admin_kit/views.py +++ b/fastapi_admin_kit/views.py @@ -7,7 +7,7 @@ from fastapi import APIRouter, Request from fastapi.responses import HTMLResponse -from fastapi_admin_kit.admin.decorators import column +from fastapi_admin_kit.admin.decorators import column, endpoint if TYPE_CHECKING: from fastapi_admin_kit.registry import RegisteredModel @@ -39,6 +39,9 @@ class ModelAdmin: # Decorator for custom column display column = staticmethod(column) + # Decorator for custom FastAPI endpoints + endpoint = staticmethod(endpoint) + def __str__(self, obj: Any) -> str: """How to display an object in dropdowns/links.""" return str(getattr(obj, "name", None) or getattr(obj, "title", None) or f"#{obj.id}") diff --git a/tests/test_endpoint_decorator.py b/tests/test_endpoint_decorator.py new file mode 100644 index 0000000..da714be --- /dev/null +++ b/tests/test_endpoint_decorator.py @@ -0,0 +1,301 @@ +"""Tests for the @endpoint decorator — custom ModelAdmin FastAPI routes.""" + +from __future__ import annotations + +import asyncio +import os +import tempfile + +import pytest +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient +from pydantic import BaseModel +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, ModelAdmin, endpoint +from fastapi_admin_kit.admin.decorators import EndpointOptions +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, create_session_cookie, 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 admin_user(engine): + async def _create(): + async with AsyncSession(engine) as session: + role = Role(name="SuperAdmin") + session.add(role) + await session.flush() + user = User( + email="admin@test.com", + hashed_password="$2b$12$HQlaDF1uaZvpsppxtnwD5uXp1VxiNXsiS5OCEkXRn7G0xNjUEo8cG", + full_name="Admin", + is_superuser=True, + is_active=True, + ) + user.roles.append(role) + session.add(user) + await session.commit() + await session.refresh(user) + return user + + return run_async(_create()) + + +class HealthResponse(BaseModel): + status: str + + +class ProductEndpointAdmin(ModelAdmin): + @endpoint( + path="/health-check", + methods=["GET"], + tags=["monitoring"], + description="Health check endpoint", + summary="Health summary", + response_description="Healthy status", + permission="view", + ) + async def health_check(self, request): + return {"status": "healthy"} + + @endpoint( + path="/typed", + methods=["GET"], + response_model=HealthResponse, + status_code=201, + ) + async def typed(self, request: Request, limit: int = 3): + assert request is not None + return HealthResponse(status=f"ok-{limit}") + + @endpoint(path="/submit", methods=["POST"]) + async def submit(self, request): + return {"submitted": True} + + +@pytest.fixture +def client(engine, admin_user): + app = FastAPI() + admin = Admin( + app=app, + engine=engine, + secret_key=SECRET_KEY, + auth_backend=BuiltinAuthBackend(), + auto_discover=False, + ) + admin.register(Product, ProductEndpointAdmin) + asyncio.run(admin.setup(app)) + return TestClient(app), admin, engine + + +# =========================================================================== +# Decorator metadata +# =========================================================================== + + +class TestEndpointDecoratorMetadata: + def test_sets_admin_endpoint_attribute(self): + opts = ProductEndpointAdmin.__dict__["health_check"]._admin_endpoint + assert isinstance(opts, EndpointOptions) + assert opts.path == "/health-check" + assert opts.methods == ["GET"] + assert opts.tags == ["monitoring"] + assert opts.description == "Health check endpoint" + assert opts.summary == "Health summary" + assert opts.response_description == "Healthy status" + assert opts.permission == "view" + + def test_defaults(self): + class DefaultAdmin(ModelAdmin): + @endpoint(path="/foo") + async def foo(self, request): + return {} + + opts = DefaultAdmin.__dict__["foo"]._admin_endpoint + assert opts.methods == ["GET"] + assert opts.tags == [] + assert opts.description == "" + assert opts.name == "" + assert opts.dependencies == [] + assert opts.status_code == 200 + assert opts.response_model is None + assert opts.permission is None + assert opts.include_in_schema is True + + def test_auto_name(self): + class AutoAdmin(ModelAdmin): + @endpoint(path="/auto") + async def my_endpoint(self, request): + return {} + + assert AutoAdmin.__dict__["my_endpoint"]._admin_endpoint.name == "" + + def test_exported_from_package(self): + from fastapi_admin_kit import endpoint as pkg_endpoint + + assert pkg_endpoint is endpoint + + +# =========================================================================== +# Router registration +# =========================================================================== + + +class TestEndpointRouterRegistration: + def test_routes_registered(self, client): + _, _, _ = client + from fastapi_admin_kit.registry import AdminRegistry + + registered = AdminRegistry().get("products") + from fastapi_admin_kit.router import build_model_router + + router = build_model_router(registered) + paths = {r.path: r.methods for r in router.routes} + assert "/products/health-check" in paths + assert "/products/typed" in paths + assert "/products/submit" in paths + assert "GET" in paths["/products/health-check"] + assert "POST" in paths["/products/submit"] + + def test_name_generated(self, client): + from fastapi_admin_kit.registry import AdminRegistry + from fastapi_admin_kit.router import build_model_router + + registered = AdminRegistry().get("products") + router = build_model_router(registered) + named = {r.name for r in router.routes} + assert "products_health_check" in named + + def test_existing_routes_unchanged(self, client): + from fastapi_admin_kit.registry import AdminRegistry + from fastapi_admin_kit.router import build_model_router + + registered = AdminRegistry().get("products") + router = build_model_router(registered) + paths = {r.path for r in router.routes} + assert "/products/" in paths + assert "/products/create" in paths + assert "/products/bulk" in paths + assert "/products/export/" in paths + + +# =========================================================================== +# End-to-end HTTP behaviour +# =========================================================================== + + +class TestEndpointHTTP: + def test_health_check_requires_auth(self, client): + test_client, _, _ = client + resp = test_client.get("/admin/products/health-check") + assert resp.status_code in {401, 403} + + def test_health_check_authenticated(self, client, admin_user): + test_client, _, _ = client + cookie = create_session_cookie(admin_user.id) + resp = test_client.get("/admin/products/health-check", cookies={"admin_session": cookie}) + assert resp.status_code == 200 + assert resp.json() == {"status": "healthy"} + + def test_untyped_request_injected(self, client, admin_user): + test_client, _, _ = client + cookie = create_session_cookie(admin_user.id) + resp = test_client.get("/admin/products/health-check", cookies={"admin_session": cookie}) + assert resp.json() == {"status": "healthy"} + + def test_typed_request_and_query_param(self, client, admin_user): + test_client, _, _ = client + cookie = create_session_cookie(admin_user.id) + resp = test_client.get("/admin/products/typed?limit=7", cookies={"admin_session": cookie}) + assert resp.status_code == 201 + assert resp.json() == {"status": "ok-7"} + + def test_default_query_param(self, client, admin_user): + test_client, _, _ = client + cookie = create_session_cookie(admin_user.id) + resp = test_client.get("/admin/products/typed", cookies={"admin_session": cookie}) + assert resp.status_code == 201 + assert resp.json() == {"status": "ok-3"} + + def test_post_method(self, client, admin_user): + test_client, _, _ = client + cookie = create_session_cookie(admin_user.id) + resp = test_client.post("/admin/products/submit", cookies={"admin_session": cookie}) + assert resp.status_code == 200 + assert resp.json() == {"submitted": True} + + def test_rbac_permission_denied_for_regular_user(self, client): + """A non-superuser without the view permission gets 403.""" + test_client, _, engine = client + + async def _create_user(): + async with AsyncSession(engine) as session: + role = Role(name="Viewer") + session.add(role) + await session.flush() + user = User( + email="viewer@test.com", + hashed_password="$2b$12$HQlaDF1uaZvpsppxtnwD5uXp1VxiNXsiS5OCEkXRn7G0xNjUEo8cG", + full_name="Viewer", + is_superuser=False, + is_active=True, + ) + user.roles.append(role) + session.add(user) + await session.commit() + await session.refresh(user) + return user + + user = run_async(_create_user()) + cookie = create_session_cookie(user.id) + resp = test_client.get("/admin/products/health-check", cookies={"admin_session": cookie}) + assert resp.status_code == 403 + + def test_openapi_includes_endpoint(self, client, admin_user): + test_client, _, _ = client + schema = test_client.get("/openapi.json").json() + assert "/admin/products/health-check" in schema["paths"] + get_op = schema["paths"]["/admin/products/health-check"]["get"] + assert get_op["tags"] == ["Product", "monitoring"] + assert get_op["summary"] == "Health summary" + assert get_op["description"] == "Health check endpoint" + assert get_op["responses"]["200"]["description"] == "Healthy status" + + def test_response_model_documented(self, client, admin_user): + test_client, _, _ = client + schema = test_client.get("/openapi.json").json() + path = "/admin/products/typed" + assert "201" in schema["paths"][path]["get"]["responses"] + assert schema["paths"][path]["get"]["responses"]["201"]["content"]