Implement CRUD and typed queries¶

"""Application use cases for projects and todo items."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Generic, TypeVar

from ormdantic import Ormdantic, QueryExpression, column, selectinload

from .database import db
from .errors import ResourceNotFound
from .models import Project, Todo, TodoStatus, utc_now
from .schemas import ProjectCreate, TodoCreate, TodoUpdate

ModelT = TypeVar("ModelT")


@dataclass(frozen=True)
class Page(Generic[ModelT]):
    """A stable slice of matching domain models."""

    items: list[ModelT]
    total: int
    limit: int
    offset: int


class TodoService:
    """Coordinate Todo application persistence operations."""

    def __init__(self, database: Ormdantic) -> None:
        self.db = database

    async def create_project(self, payload: ProjectCreate) -> Project:
        """Create a Project from validated input."""
        return await self.db[Project].insert(Project(name=payload.name))

    async def list_projects(self, *, limit: int, offset: int) -> Page[Project]:
        """List Projects in stable creation order."""
        table = self.db[Project]
        result = await table.find_many(
            order_by=["created_at", "id"],
            limit=limit,
            offset=offset,
        )
        return Page(
            items=result.data,
            total=await table.count(),
            limit=limit,
            offset=offset,
        )

    async def get_project(self, project_id: str) -> Project:
        """Return a Project or raise a stable domain error."""
        project = await self.db[Project].find_one(project_id)
        if project is None:
            raise ResourceNotFound("Project", project_id)
        return project

    async def create_todo(self, project_id: str, payload: TodoCreate) -> Todo:
        """Create a Todo only when its owning Project exists."""
        async with self.db.transaction():
            project = await self.get_project(project_id)
            todo = Todo(
                project=project.id,
                title=payload.title,
                description=payload.description,
                priority=payload.priority,
                due_at=payload.due_at,
            )
            return await self.db[Todo].insert(todo)

    async def list_todos(
        self,
        *,
        project_id: str | None = None,
        status: TodoStatus | None = None,
        priority: int | None = None,
        search: str | None = None,
        limit: int,
        offset: int,
        load_project: bool = False,
    ) -> Page[Todo]:
        """List Todos using composed typed filters."""
        clauses: list[QueryExpression] = []
        if project_id is not None:
            clauses.append(column("project").eq(project_id))
        if status is not None:
            clauses.append(column("status").eq(status.value))
        if priority is not None:
            clauses.append(column("priority").eq(priority))
        if search is not None and search.strip():
            clauses.append(column("title").ilike(f"%{search.strip()}%"))

        where: QueryExpression | None = None
        for clause in clauses:
            where = clause if where is None else where & clause

        table = self.db[Todo]
        loaders = [selectinload("project")] if load_project else None
        result = await table.find_many(
            where=where,
            order_by=["created_at", "id"],
            limit=limit,
            offset=offset,
            load=loaders,
        )
        return Page(
            items=result.data,
            total=await table.count(where),
            limit=limit,
            offset=offset,
        )

    async def get_todo(self, todo_id: str, *, load_project: bool = True) -> Todo:
        """Return one Todo with an optional explicit Project load."""
        loaders = [selectinload("project")] if load_project else None
        todo = await self.db[Todo].find_one(todo_id, load=loaders)
        if todo is None:
            raise ResourceNotFound("Todo", todo_id)
        return todo

    async def update_todo(self, todo_id: str, payload: TodoUpdate) -> Todo:
        """Apply supplied PATCH fields and update the modification time."""
        current = await self.get_todo(todo_id, load_project=False)
        values = current.model_dump()
        values.update(payload.model_dump(exclude_unset=True))
        values["updated_at"] = utc_now()
        updated = Todo.model_validate(values)
        return await self.db[Todo].update(updated)

    async def delete_todo(self, todo_id: str) -> None:
        """Delete an existing Todo."""
        await self.get_todo(todo_id, load_project=False)
        await self.db[Todo].delete(todo_id)


service = TodoService(db)

"""HTTP routes for the Todo reference application."""

from __future__ import annotations

from typing import Annotated

from fastapi import APIRouter, Depends, Path, Query, Request, Response, status

from .models import TodoStatus
from .schemas import (
    ProjectCreate,
    ProjectPage,
    ProjectResponse,
    TodoCreate,
    TodoPage,
    TodoResponse,
    TodoUpdate,
    UUIDString,
)
from .service import TodoService

router = APIRouter()

PageLimit = Annotated[int, Query(ge=1, le=100)]
PageOffset = Annotated[int, Query(ge=0)]
ResourceId = Annotated[UUIDString, Path()]


def get_service(request: Request) -> TodoService:
    """Return the application-scoped service used by an endpoint."""
    return request.app.state.todo_service


Service = Annotated[TodoService, Depends(get_service)]


@router.get("/health", tags=["operations"])
async def health(request: Request) -> dict[str, str]:
    """Report readiness without exposing a database URL or credentials."""
    return {
        "status": "ready",
        "database": request.app.state.database_dialect,
    }


@router.post(
    "/projects",
    response_model=ProjectResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["projects"],
)
async def create_project(
    payload: ProjectCreate,
    service: Service,
) -> ProjectResponse:
    """Create a project."""
    project = await service.create_project(payload)
    return ProjectResponse.model_validate(project)


@router.get("/projects", response_model=ProjectPage, tags=["projects"])
async def list_projects(
    service: Service,
    limit: PageLimit = 50,
    offset: PageOffset = 0,
) -> ProjectPage:
    """Return a stable, bounded page of projects."""
    page = await service.list_projects(limit=limit, offset=offset)
    return ProjectPage(
        items=[ProjectResponse.model_validate(project) for project in page.items],
        total=page.total,
        limit=page.limit,
        offset=page.offset,
    )


@router.get(
    "/projects/{project_id}",
    response_model=ProjectResponse,
    tags=["projects"],
)
async def get_project(
    project_id: ResourceId,
    service: Service,
) -> ProjectResponse:
    """Return one project by its canonical UUID."""
    project = await service.get_project(project_id)
    return ProjectResponse.model_validate(project)


@router.post(
    "/projects/{project_id}/todos",
    response_model=TodoResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["todos"],
)
async def create_todo(
    project_id: ResourceId,
    payload: TodoCreate,
    service: Service,
) -> TodoResponse:
    """Create a todo in an existing project."""
    todo = await service.create_todo(project_id, payload)
    return TodoResponse.from_model(todo)


@router.get("/todos", response_model=TodoPage, tags=["todos"])
async def list_todos(
    service: Service,
    project_id: Annotated[UUIDString | None, Query()] = None,
    todo_status: Annotated[TodoStatus | None, Query(alias="status")] = None,
    priority: Annotated[int | None, Query(ge=1, le=5)] = None,
    search: Annotated[str | None, Query(max_length=200)] = None,
    limit: PageLimit = 50,
    offset: PageOffset = 0,
) -> TodoPage:
    """Filter and paginate todo items."""
    page = await service.list_todos(
        project_id=project_id,
        status=todo_status,
        priority=priority,
        search=search,
        limit=limit,
        offset=offset,
    )
    return TodoPage(
        items=[TodoResponse.from_model(todo) for todo in page.items],
        total=page.total,
        limit=page.limit,
        offset=page.offset,
    )


@router.get("/todos/{todo_id}", response_model=TodoResponse, tags=["todos"])
async def get_todo(todo_id: ResourceId, service: Service) -> TodoResponse:
    """Return one todo with its project relationship loaded."""
    return TodoResponse.from_model(await service.get_todo(todo_id))


@router.patch("/todos/{todo_id}", response_model=TodoResponse, tags=["todos"])
async def update_todo(
    todo_id: ResourceId,
    payload: TodoUpdate,
    service: Service,
) -> TodoResponse:
    """Apply a validated partial update."""
    return TodoResponse.from_model(await service.update_todo(todo_id, payload))


@router.delete(
    "/todos/{todo_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    tags=["todos"],
)
async def delete_todo(todo_id: ResourceId, service: Service) -> Response:
    """Delete one todo."""
    await service.delete_todo(todo_id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

"""FastAPI application factory for the Ormdantic Todo example."""

from __future__ import annotations

from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from ormdantic import Ormdantic

from .database import db
from .errors import TodoApplicationError, error_response
from .routes import router
from .service import TodoService


def create_app(database: Ormdantic | None = None) -> FastAPI:
    """Build an application around an injected or configured database target."""
    runtime_database = database or db

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        await runtime_database.init()
        app.state.todo_service = TodoService(runtime_database)
        app.state.database_dialect = runtime_database.runtime_diagnostics()["backend"]
        yield

    application = FastAPI(
        title="Ormdantic Todo API",
        version="0.1.0",
        lifespan=lifespan,
    )
    application.include_router(router)

    @application.exception_handler(TodoApplicationError)
    async def handle_application_error(
        _request: Request,
        error: TodoApplicationError,
    ) -> JSONResponse:
        status_code, payload = error_response(error)
        return JSONResponse(status_code=status_code, content=payload)

    return application


app = create_app()