Skip to content

Endpoints and routers

This page exposes the TaskAction from Actions and CRUD over HTTP, mounts it in the app, and tests it. It follows the same shape MSFLib modules use for their own routers, so what you build here is how you will mount msflib-account, msflib-workspaces and the rest.

A request-scoped dependency

The router needs to know who is calling, to fill owner on create and to scope every read. In a real app that comes from msflib-auth (see Authentication). To keep this Quick Start self-contained, a header stands in for it:

# app/deps.py
from fastapi import Header


def get_owner(x_owner: str = Header(default="anonymous")) -> str:
    return x_owner

Warning

A client-supplied header is not authentication. It is here only so the example has a caller. Replace get_owner with a dependency that resolves the current account from a verified token.

The router factory

MSFLib routers are factory functions. The factory takes the dependencies it needs as arguments and returns an APIRouter, instead of importing a global session or user. The same router code then works with any engine and any auth setup, and tests can pass fakes.

# app/router.py
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query
from msflib.api import create_enhanced_router
from sqlmodel import Session

from .actions import task_action
from .models import TaskCreate, TaskRead, TaskUpdate


def task_router(*, get_session, get_owner, create_type=TaskCreate, update_type=TaskUpdate):
    api_router = APIRouter(prefix="/tasks", tags=["tasks"])
    router = create_enhanced_router(api_router)

    @router.get("/", response_model=list[TaskRead])
    def list_tasks(
        session: Session = Depends(get_session),
        owner: str = Depends(get_owner),
        offset: int = Query(0, ge=0),
        limit: int = Query(100, ge=1, le=500),
    ) -> Any:
        return task_action.get_multi_by_all(session, offset=offset, limit=limit, owner=owner)

    @router.post(
        "/",
        response_model=TaskRead,
        status_code=201,
        validation_types={"data": create_type},
    )
    def create_task(
        data: Any,
        session: Session = Depends(get_session),
        owner: str = Depends(get_owner),
    ) -> Any:
        return task_action.create(session, data=data, update={"owner": owner})

    @router.get("/{task_id}", response_model=TaskRead)
    def read_task(
        task_id: int,
        session: Session = Depends(get_session),
        owner: str = Depends(get_owner),
    ) -> Any:
        task = task_action.get_by_all(session, id=task_id, owner=owner)
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task

    @router.patch(
        "/{task_id}",
        response_model=TaskRead,
        validation_types={"data": update_type},
    )
    def update_task(
        task_id: int,
        data: Any,
        session: Session = Depends(get_session),
        owner: str = Depends(get_owner),
    ) -> Any:
        task = task_action.get_by_all(session, id=task_id, owner=owner)
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task_action.update(session, model=task, data=data)

    @router.delete("/{task_id}", response_model=TaskRead)
    def delete_task(
        task_id: int,
        session: Session = Depends(get_session),
        owner: str = Depends(get_owner),
    ) -> Any:
        task = task_action.get_by_all(session, id=task_id, owner=owner)
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task_action.delete(session, task.id)

    return api_router

Things worth noting:

  • Scoping is in the query, not the check. Every handler looks the row up with id=task_id, owner=owner. A task owned by someone else is indistinguishable from one that does not exist, and both give 404.
  • response_model=TaskRead controls what leaves the API, whatever the action returns.
  • create_enhanced_router and validation_types. create_enhanced_router(api_router) wraps the router so each HTTP-method decorator accepts validation_types={"param": SomeModel}. At registration time the named parameter's annotation is replaced with that model, and FastAPI validates the request body against it. The handler body still receives the validated object as a plain argument. This is what makes the request schema a parameter of the factory (create_type, update_type) rather than something hard-coded: a host app can mount the same router with a stricter or extended schema. Parameters not listed, including every Depends(...), are left as they are. The wrapper delegates everything else to the underlying router, which is why the factory returns api_router, the original APIRouter, for include_router.
  • You do not need create_enhanced_router for your own routes. In an app where the schema never varies, annotate data: TaskCreate and use a plain APIRouter. Use it when you write routers meant to be reused with substitutable schemas, as MSFLib modules do.
  • The data: Any parameter with no default must come before parameters that have defaults, which is why it is listed first.

Substitute a schema

Because the schema is an argument, a caller can tighten it without touching the router:

from pydantic import Field

from app.models import TaskCreate


class StrictTaskCreate(TaskCreate):
    title: str = Field(min_length=3)


router = task_router(get_session=get_session, get_owner=get_owner, create_type=StrictTaskCreate)

A POST with {"title": "ab"} now returns 422 and {"title": "abc"} returns 201. The action receives the validated object as data exactly as before. Overriding models covers the wider pattern.

Mount it

# app/main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from msflib.db.migrations import migrate
from msflib.eventbus import bind_app_emitter

from .db import engine, get_session
from .deps import get_owner
from .migration import MIGRATION
from .router import task_router
from .settings import settings

core = settings.scope("CORE")


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    migrate(engine, MIGRATION)
    yield


app = FastAPI(
    title=core.PROJECT_NAME,
    openapi_url=f"{core.API_V1_STR}/openapi.json",
    lifespan=lifespan,
)
app_emitter = bind_app_emitter(app)


@app.get(f"{core.API_V1_STR}/health")
def health() -> dict[str, str]:
    return {"status": "ok", "project": core.PROJECT_NAME}


app.include_router(
    task_router(get_session=get_session, get_owner=get_owner),
    prefix=core.API_V1_STR,
)

The prefix comes from settings (/api/v1 by default), and the router adds its own /tasks. Mounting a module router is the same call with the module's factory and the dependencies it documents. Mounting routes covers prefixes, tags and several routers together.

Run it

uvicorn app.main:app --reload

curl -X POST localhost:8000/api/v1/tasks/ \
  -H 'content-type: application/json' -H 'X-Owner: ann' \
  -d '{"title": "write docs", "priority": "high"}'
# {"id":1,"title":"write docs","done":false,"priority":"high","owner":"ann"}

curl localhost:8000/api/v1/tasks/ -H 'X-Owner: ann'
# [{"id":1,"title":"write docs","done":false,"priority":"high","owner":"ann"}]

curl localhost:8000/api/v1/tasks/ -H 'X-Owner: bob'
# []

/docs now lists the five task routes, and shows TaskCreate as the request body of the POST.

Test it

FastAPI's TestClient drives the app in-process. Replace the session dependency with one bound to a throwaway in-memory database, and the tests never touch tasks.db:

# tests/test_tasks.py
import pytest
from fastapi.testclient import TestClient
from msflib.db.sqlite import enable_savepoints
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine

from app.actions import task_action
from app.db import get_session
from app.main import app
from app.models import TaskCreate


@pytest.fixture
def engine():
    engine = enable_savepoints(
        create_engine(
            "sqlite://",
            connect_args={"check_same_thread": False},
            poolclass=StaticPool,
        )
    )
    SQLModel.metadata.create_all(engine)
    return engine


@pytest.fixture
def client(engine):
    def override_session():
        with Session(engine) as session:
            yield session

    app.dependency_overrides[get_session] = override_session
    yield TestClient(app)
    app.dependency_overrides.clear()


def test_owner_scoping(client):
    body = {"title": "write docs"}
    created = client.post("/api/v1/tasks/", json=body, headers={"X-Owner": "ann"})
    assert created.status_code == 201
    assert created.json()["owner"] == "ann"

    assert client.get("/api/v1/tasks/", headers={"X-Owner": "ann"}).json() != []
    assert client.get("/api/v1/tasks/", headers={"X-Owner": "bob"}).json() == []


def test_action_directly(engine):
    with Session(engine) as session:
        task = task_action.create(
            session, data=TaskCreate(title="from a test"), update={"owner": "ann"}
        )
        assert task_action.get(session, task.id).title == "from a test"

python -m pytest tests

Two details. StaticPool makes every connection share the one in-memory database; without it each thread gets its own empty database, and TestClient runs the app in a different thread from your test. And dependency_overrides is keyed by the exact get_session function that the router was built with, so import it from app.db, not a copy. The TestClient here is not used as a context manager, so the app's lifespan (and migrate) does not run; the fixture creates the tables itself.

Troubleshooting

422 on every write, with data reported as a missing query parameter. The decorator has no validation_types={"data": ...}, so data: Any is treated as a query parameter. Passing validation_types to a plain APIRouter decorator raises TypeError: ... unexpected keyword argument 'validation_types'; register the route on the wrapper returned by create_enhanced_router.

sqlite3.OperationalError: no such table: task in tests. The test engine has no tables. Call SQLModel.metadata.create_all(engine) after importing the models, as the fixture does.

Tests see data from an earlier test. The engine fixture creates a new in-memory database per test. If you made the engine module-level, drop and recreate the tables in the fixture instead.

The override has no effect. The override key must be the same get_session object given to task_router.

Every request returns 404 for rows you just created. The X-Owner header differs between the create and the read. Reads are scoped to the owner.

Where next

You now have the full loop: settings, engine, models, tables, actions, routes and tests, on the core package alone. From here: