Skip to content

Model projects and todo items

You will now see how ordinary Pydantic validation becomes table metadata, constraints, indexes, and a named relationship.

%% entities
erDiagram
    PROJECT ||--o{ TODO : contains
    PROJECT {
        string id PK
        string name
        datetime created_at
    }
    TODO {
        string id PK
        string project FK
        string title
        string status
        integer priority
        datetime due_at
    }
"""Persisted models for the Todo example."""

from __future__ import annotations

from datetime import datetime, timezone
from enum import Enum
from typing import overload
from uuid import UUID, uuid4

from pydantic import BaseModel, ConfigDict, Field, field_validator

from ormdantic import TableColumn

from .database import db


def utc_now() -> datetime:
    """Return the current time as a timezone-aware UTC datetime."""
    return datetime.now(timezone.utc)


def _uuid_string() -> str:
    return str(uuid4())


def _canonical_uuid(value: str) -> str:
    try:
        return str(UUID(value))
    except (AttributeError, TypeError, ValueError) as error:
        raise ValueError("value must be a valid UUID") from error


@overload
def _aware_to_utc(value: datetime) -> datetime: ...


@overload
def _aware_to_utc(value: None) -> None: ...


def _aware_to_utc(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None or value.utcoffset() is None:
        raise ValueError("datetime must be timezone-aware")
    return value.astimezone(timezone.utc)


class TodoStatus(str, Enum):
    """Lifecycle states persisted for a todo item."""

    pending = "pending"
    in_progress = "in_progress"
    completed = "completed"


@db.table("project", pk="id", indexed=["name"])
class Project(BaseModel):
    """A named collection of todo items."""

    model_config = ConfigDict(str_strip_whitespace=True, validate_default=True)

    id: str = Field(default_factory=_uuid_string)
    name: str = Field(min_length=1, max_length=120)
    created_at: datetime = Field(default_factory=utc_now)

    @field_validator("id")
    @classmethod
    def normalize_id(cls, value: str) -> str:
        """Validate and canonicalize the persisted identifier."""
        return _canonical_uuid(value)

    @field_validator("created_at")
    @classmethod
    def normalize_created_at(cls, value: datetime) -> datetime:
        """Require the creation time to be aware and normalized to UTC."""
        return _aware_to_utc(value)


@db.table(
    "todo",
    pk="id",
    indexed=["title"],
    column_options={
        "project": TableColumn(
            foreign_key_name="todo_project_fk",
            on_delete="cascade",
        )
    },
)
class Todo(BaseModel):
    """A prioritized unit of work belonging to a project."""

    model_config = ConfigDict(str_strip_whitespace=True, validate_default=True)

    id: str = Field(default_factory=_uuid_string)
    project: Project | str
    title: str = Field(min_length=1, max_length=200)
    description: str | None = None
    status: TodoStatus = TodoStatus.pending
    priority: int = Field(default=3, ge=1, le=5)
    due_at: datetime | None = None
    created_at: datetime = Field(default_factory=utc_now)
    updated_at: datetime = Field(default_factory=utc_now)

    @field_validator("id")
    @classmethod
    def normalize_id(cls, value: str) -> str:
        """Validate and canonicalize the persisted identifier."""
        return _canonical_uuid(value)

    @field_validator("project")
    @classmethod
    def normalize_project(cls, value: Project | str) -> Project | str:
        """Validate shallow project identifiers while preserving nested models."""
        if isinstance(value, Project):
            return value
        return _canonical_uuid(value)

    @field_validator("created_at", "updated_at", "due_at")
    @classmethod
    def normalize_timestamp(cls, value: datetime | None) -> datetime | None:
        """Require persisted timestamps to be aware and normalized to UTC."""
        return _aware_to_utc(value)

Both identifiers are canonical UUID strings. This keeps response shapes identical between SQLite and PostgreSQL. Time factories produce aware UTC values, and validators reject naive datetimes before persistence.

Todo.project accepts either a shallow Project identifier or a loaded Project. The todo_project_fk constraint uses ON DELETE CASCADE; deleting a Project at the database level removes its Todos. Naming the constraint also makes generated migration SQL easier to audit.

Pydantic field constraints produce database checks where the driver supports them. Here they protect name/title lengths, enum values, and the priority range. Application validation gives clients early feedback, while constraints protect writes that bypass the API.

The HTTP schemas remain separate:

"""Validated request and response shapes for the Todo API."""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Annotated, Any
from uuid import UUID

from pydantic import (
    BaseModel,
    BeforeValidator,
    ConfigDict,
    Field,
    WithJsonSchema,
    field_validator,
)
from pydantic.json_schema import SkipJsonSchema

from .models import Project, Todo, TodoStatus


def _canonical_uuid(value: str | UUID) -> str:
    if isinstance(value, UUID):
        return str(value)
    try:
        return str(UUID(value))
    except (AttributeError, TypeError, ValueError) as error:
        raise ValueError("value must be a valid UUID") from error


UUIDString = Annotated[
    str,
    BeforeValidator(_canonical_uuid),
    WithJsonSchema({"type": "string", "format": "uuid"}),
]


def _aware_to_utc(value: datetime) -> datetime:
    if value.tzinfo is None or value.utcoffset() is None:
        raise ValueError("datetime must be timezone-aware")
    return value.astimezone(timezone.utc)


def _patch_schema(schema: dict[str, Any]) -> None:
    properties = schema.get("properties", {})
    for name in ("title", "status", "priority"):
        field_schema = properties.get(name)
        if isinstance(field_schema, dict):
            field_schema.pop("default", None)


class ProjectCreate(BaseModel):
    """Input accepted when creating a project."""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    name: str = Field(
        min_length=1,
        max_length=120,
        description="Human-readable project name.",
    )


class ProjectResponse(BaseModel):
    """Public representation of a persisted project."""

    model_config = ConfigDict(from_attributes=True)

    id: UUIDString = Field(description="Canonical project UUID.")
    name: str = Field(description="Human-readable project name.")
    created_at: datetime = Field(description="UTC project creation time.")

    @field_validator("created_at")
    @classmethod
    def normalize_created_at(cls, value: datetime) -> datetime:
        """Require and normalize the project creation time."""
        return _aware_to_utc(value)


class TodoCreate(BaseModel):
    """Input accepted when creating a todo item."""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    title: str = Field(
        min_length=1,
        max_length=200,
        description="Short todo title.",
    )
    description: str | None = Field(
        default=None,
        max_length=4000,
        description="Optional todo details.",
    )
    priority: int = Field(
        default=3,
        ge=1,
        le=5,
        description="Priority from 1 (highest) to 5 (lowest).",
    )
    due_at: datetime | None = Field(
        default=None,
        description="Optional timezone-aware due time, normalized to UTC.",
    )

    @field_validator("due_at")
    @classmethod
    def normalize_due_at(cls, value: datetime | None) -> datetime | None:
        """Require supplied due times to be aware and normalize them to UTC."""
        return None if value is None else _aware_to_utc(value)


class TodoUpdate(BaseModel):
    """Optional fields accepted when partially updating a todo item."""

    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True,
        json_schema_extra=_patch_schema,
    )

    title: str | SkipJsonSchema[None] = Field(
        default=None,
        min_length=1,
        max_length=200,
        description="Replacement title when supplied.",
    )
    description: str | None = Field(
        default=None,
        max_length=4000,
        description="Replacement details; null clears them.",
    )
    status: TodoStatus | SkipJsonSchema[None] = Field(
        default=None,
        description="Replacement lifecycle status.",
    )
    priority: int | SkipJsonSchema[None] = Field(
        default=None,
        ge=1,
        le=5,
        description="Replacement priority from 1 to 5.",
    )
    due_at: datetime | None = Field(
        default=None,
        description="Replacement due time; null clears it.",
    )

    @field_validator("title", "status", "priority", mode="before")
    @classmethod
    def reject_null_non_nullable_patch_field(cls, value: Any) -> Any:
        """Allow omission but reject null for fields that cannot be cleared."""
        if value is None:
            raise ValueError("field cannot be null")
        return value

    @field_validator("due_at")
    @classmethod
    def normalize_due_at(cls, value: datetime | None) -> datetime | None:
        """Require non-null due times to be aware and normalize them to UTC."""
        return None if value is None else _aware_to_utc(value)


class TodoResponse(BaseModel):
    """Stable public representation of a persisted todo item."""

    id: UUIDString = Field(description="Canonical todo UUID.")
    project_id: UUIDString = Field(description="Canonical UUID of the owning project.")
    title: str = Field(description="Todo title.")
    description: str | None = Field(description="Optional todo details.")
    status: TodoStatus = Field(description="Current lifecycle status.")
    priority: int = Field(ge=1, le=5, description="Priority from 1 to 5.")
    due_at: datetime | None = Field(description="Optional UTC due time.")
    created_at: datetime = Field(description="UTC creation time.")
    updated_at: datetime = Field(description="UTC last-update time.")
    project: ProjectResponse | None = Field(
        default=None,
        description="Loaded project data when the relationship was expanded.",
    )

    @field_validator("created_at", "updated_at")
    @classmethod
    def normalize_required_timestamp(cls, value: datetime) -> datetime:
        """Require response timestamps to be aware and normalize them to UTC."""
        return _aware_to_utc(value)

    @field_validator("due_at")
    @classmethod
    def normalize_optional_timestamp(cls, value: datetime | None) -> datetime | None:
        """Normalize an optional due time to UTC."""
        return None if value is None else _aware_to_utc(value)

    @classmethod
    def from_model(cls, todo: Todo) -> TodoResponse:
        """Convert a persisted todo with a shallow or loaded project relation."""
        relation = todo.project
        if isinstance(relation, Project):
            project_id = relation.id
            project = ProjectResponse.model_validate(relation)
        else:
            project_id = relation
            project = None

        return cls(
            id=todo.id,
            project_id=project_id,
            title=todo.title,
            description=todo.description,
            status=todo.status,
            priority=todo.priority,
            due_at=todo.due_at,
            created_at=todo.created_at,
            updated_at=todo.updated_at,
            project=project,
        )


class ProjectPage(BaseModel):
    """A bounded page of projects."""

    items: list[ProjectResponse]
    total: int = Field(ge=0, description="Total matching projects.")
    limit: int = Field(ge=1, le=100, description="Maximum returned items.")
    offset: int = Field(ge=0, description="Number of matching items skipped.")


class TodoPage(BaseModel):
    """A bounded page of todo items."""

    items: list[TodoResponse]
    total: int = Field(ge=0, description="Total matching todo items.")
    limit: int = Field(ge=1, le=100, description="Maximum returned items.")
    offset: int = Field(ge=0, description="Number of matching items skipped.")

This separation prevents persistence-only fields from becoming writable by accident. PATCH distinguishes omitted values from explicit nulls, and UUID fields appear as format: uuid in OpenAPI.

Read Field types and metadata for the complete mapping and Loading strategies for relationship behavior.