Skip to Content
Proof and receiptsA scaffolded resource, in full

A scaffolded resource, in full

A new user-owned resource starts as six files copied from a template set and substituted: model, migration, schema, repository, router, tests. There is no generator command — the templates are files you copy, which is deliberate. A generator is a second thing to maintain, and a buyer who reads the template once understands what they now own.

Below is the complete output for a fictional SavedSearch resource. It does not run: the factory primitives it imports are not published here. It is here to be read.

What it is meant to prove: that generated code arrives with its security properties already in place, rather than as a shape you are expected to harden afterwards.

repository.py

Start here. The ownership filter is one SQL WHERE clause in one place, and every read of this resource goes through it. AsyncBaseRepository deliberately exposes no ownership-aware read, so the way to reach a row by id is get_for_user — a router that calls repo.get(...) or session.get(Model, id) has gone around the filter, which is the IDOR red flag reviewers look for. Either way, a reviewer has one method to read rather than every endpoint.

repository.py
"""Async repository for SavedSearch.
 
INSPECTION RENDERING — not runnable as-is.
 
Generated from The Fabrica's private resource scaffold for a fictional
`SavedSearch` resource. The factory primitive it extends
(`AsyncBaseRepository`) is intentionally omitted from this repository:
it is meant to be inspected, not executed.
 
This file is the one worth reading closely. The router performs every
database read through it, so the `WHERE user_id = ...` predicate lives
in ONE place instead of being restated at each endpoint. Omitting that
predicate in a single endpoint is the most common way an owned-resource
API grows an IDOR hole; centralising it means a reviewer confirms
ownership by reading one file rather than auditing every handler.
 
Ownership is filtered IN SQL — not fetched and then compared in Python.
A row belonging to another user is never loaded into the process.
"""
 
from __future__ import annotations
 
import uuid
from collections.abc import Sequence
 
from sqlalchemy import func, select
 
from src.models._domain.saved_search import SavedSearch
from src.services.repositories_async.base import AsyncBaseRepository
 
 
class SavedSearchRepository(AsyncBaseRepository[SavedSearch]):
    """One repository, one model. Every query filters by user_id."""
 
    model = SavedSearch
 
    async def get_for_user(
        self,
        resource_id: uuid.UUID,
        *,
        user_id: uuid.UUID,
    ) -> SavedSearch | None:
        """Fetch by id WITH ownership check. Returns None for foreign rows.
 
        This is the ONLY way the router fetches a single row. Calling
        `session.get(SavedSearch, id)` directly would skip the ownership
        filter — the classic IDOR slip.
        """
        stmt = select(SavedSearch).where(
            SavedSearch.id == resource_id,
            SavedSearch.user_id == user_id,
        )
        return (await self.session.execute(stmt)).scalar_one_or_none()
 
    async def list_for_user(
        self,
        *,
        user_id: uuid.UUID,
        limit: int = 50,
        offset: int = 0,
    ) -> Sequence[SavedSearch]:
        """List newest-first. Hits the (user_id, created_at) composite index."""
        stmt = (
            select(SavedSearch)
            .where(SavedSearch.user_id == user_id)
            .order_by(SavedSearch.created_at.desc())
            .limit(limit)
            .offset(offset)
        )
        return (await self.session.execute(stmt)).scalars().all()
 
    async def count_for_user(self, *, user_id: uuid.UUID) -> int:
        """Total rows for one user. Powers list pagination + tier-limit checks."""
        stmt = (
            select(func.count())
            .select_from(SavedSearch)
            .where(SavedSearch.user_id == user_id)
        )
        return (await self.session.execute(stmt)).scalar_one()

test_resource.py

Then this. test_foreign_row_returns_404_not_403 is the IDOR regression test, and it ships with every scaffolded resource. It is the test that fails the moment someone re-implements a read without the ownership filter — and it checks the write paths too, not only the read.

test_resource.py
"""Integration tests for the SavedSearch CRUD endpoints.
 
INSPECTION RENDERING — not runnable as-is.
 
Generated from The Fabrica's private resource scaffold for a fictional
`SavedSearch` resource. The fixtures these tests consume
(`authed_client`, `other_user_client`, `anon_client`) live in the
private conftest and are intentionally omitted from this repository:
it is meant to be inspected, not executed.
 
These are the security regression tests the scaffold generates, kept
here deliberately — they are the point of the example. Three of them
are mandatory in the private repo and are not allowed to be deleted:
 
    1. Happy path: proves the endpoint works at all.
    2. Foreign row -> 404: proves the ownership filter holds. IDOR is
       the dominant bug class for owned-by-user resources; this test
       fails noisily the moment someone drops the user_id filter.
    3. Anonymous -> 401: proves the authentication dependency is wired.
 
A generated CRUD endpoint is easy. A generated CRUD endpoint that
arrives with its own IDOR regression test is the part worth inspecting.
"""
 
from __future__ import annotations
 
import uuid
 
import pytest
from httpx import AsyncClient
 
pytestmark = pytest.mark.integration
 
 
SAMPLE_PAYLOAD = {
    "name": "Remote Python roles",
    "query": "python remote",
    "filters": {"seniority": "senior"},
}
 
 
async def test_create_then_fetch_happy_path(authed_client: AsyncClient) -> None:
    """POST returns 201 + GET by id returns the same row."""
    resp = await authed_client.post(
        "/api/v1/saved_searches",
        json=SAMPLE_PAYLOAD,
    )
    assert resp.status_code == 201, resp.text
    created = resp.json()
    resource_id = created["id"]
 
    fetched = await authed_client.get(
        f"/api/v1/saved_searches/{resource_id}",
    )
    assert fetched.status_code == 200
    assert fetched.json()["id"] == resource_id
 
 
async def test_foreign_row_returns_404_not_403(
    authed_client: AsyncClient,
    other_user_client: AsyncClient,
) -> None:
    """Another user's row must look like it doesn't exist.
 
    REGRESSION GUARD: this is the IDOR test. If someone removes the
    user_id filter in the repository, this test fails immediately
    because the request would return 200 instead of 404.
 
    404 rather than 403 is deliberate: 403 confirms the id exists,
    which is itself a disclosure.
    """
    create_resp = await other_user_client.post(
        "/api/v1/saved_searches",
        json=SAMPLE_PAYLOAD,
    )
    foreign_id = create_resp.json()["id"]
 
    get_resp = await authed_client.get(
        f"/api/v1/saved_searches/{foreign_id}",
    )
    assert get_resp.status_code == 404, (
        f"IDOR REGRESSION: user got status {get_resp.status_code} "
        f"for foreign row {foreign_id}. Must be 404 (not 403 — don't "
        f"leak existence)."
    )
 
    # Same posture for PATCH + DELETE — a write path that skipped the
    # ownership filter would be worse than a leaky read path.
    patch_resp = await authed_client.patch(
        f"/api/v1/saved_searches/{foreign_id}",
        json=SAMPLE_PAYLOAD,
    )
    assert patch_resp.status_code == 404
 
    del_resp = await authed_client.delete(
        f"/api/v1/saved_searches/{foreign_id}",
    )
    assert del_resp.status_code == 404
 
 
async def test_anonymous_request_returns_401(
    anon_client: AsyncClient,
) -> None:
    """No JWT -> 401 from the authentication dependency."""
    resp = await anon_client.get("/api/v1/saved_searches")
    assert resp.status_code == 401
 
 
async def test_list_returns_only_my_rows(
    authed_client: AsyncClient,
    other_user_client: AsyncClient,
) -> None:
    """The list endpoint filters by user_id even with no explicit filter."""
    # Other user has 1 row.
    await other_user_client.post(
        "/api/v1/saved_searches",
        json=SAMPLE_PAYLOAD,
    )
    # I have 0 rows.
    resp = await authed_client.get("/api/v1/saved_searches")
    assert resp.status_code == 200
    body = resp.json()
    assert body["total"] == 0
    assert body["items"] == []
 
 
async def test_get_nonexistent_returns_404(
    authed_client: AsyncClient,
) -> None:
    """Random UUID -> 404, not 500."""
    resp = await authed_client.get(
        f"/api/v1/saved_searches/{uuid.uuid4()}",
    )
    assert resp.status_code == 404

model.py

The user foreign key cascades on delete, because erasure is a real code path rather than a policy document. The (user_id, created_at) composite index exists because listing “my rows, newest first” is the query that will actually run.

model.py
"""SavedSearch — owned-by-user resource.
 
INSPECTION RENDERING — not runnable as-is.
 
Generated from The Fabrica's private resource scaffold for a fictional
`SavedSearch` resource, so the engineering contract can be read without
publishing the scaffold itself. The factory primitives this imports
(`Base`, `TimestampMixin`) are intentionally omitted from this
repository: it is meant to be inspected, not executed.
 
Two invariants the scaffold refuses to generate without:
 
    - CASCADE on user delete (GDPR Art. 17 right-to-erasure).
    - `(user_id, created_at)` composite index — every "my rows,
      newest-first" list query hits this index rather than sorting
      the user's whole partition.
"""
 
from __future__ import annotations
 
import uuid
 
import sqlalchemy as sa
from sqlalchemy import ForeignKey, Index, String, Text
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.orm import Mapped, mapped_column
 
from src.db.base import Base, TimestampMixin
 
 
class SavedSearch(Base, TimestampMixin):
    """A search a user has saved for reuse.
 
    Conventions inherited from the factory:
        - CASCADE on user delete (GDPR Art. 17 right-to-erasure).
        - `(user_id, created_at)` composite index.
        - TimestampMixin gives created_at + updated_at automatically.
    """
 
    __tablename__ = "saved_searches"
 
    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        default=uuid.uuid4,
    )
    user_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("users.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
        comment="Owner. CASCADE so this row disappears on GDPR delete.",
    )
 
    # === BEGIN FIELDS_BLOCK ====================================================
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    query: Mapped[str] = mapped_column(Text, nullable=False)
    filters: Mapped[dict[str, object]] = mapped_column(
        JSONB,
        nullable=False,
        server_default=sa.text("'{}'::jsonb"),
    )
    # === END FIELDS_BLOCK ======================================================
 
    __table_args__ = (
        Index(
            "ix_saved_searches_user_id_created_at",
            "user_id",
            "created_at",
        ),
    )
 
    def __repr__(self) -> str:  # pragma: no cover - debug aid
        return f"<SavedSearch id={self.id} user={self.user_id}>"

schema.py

Request and response shapes are separate from the ORM model, so the client-facing contract can diverge from the table without either one dragging the other along.

schema.py
"""Pydantic schemas for the SavedSearch router.
 
INSPECTION RENDERING — not runnable as-is.
 
Generated from The Fabrica's private resource scaffold for a fictional
`SavedSearch` resource. Surrounding factory primitives are intentionally
omitted from this repository: it is meant to be inspected, not executed.
 
Note the split between what a client may SET and what the API RETURNS.
`user_id` appears only on the response model — a client never supplies
ownership, the router takes it from the authenticated principal.
"""
 
from __future__ import annotations
 
import uuid
from datetime import datetime
from typing import Any
 
from pydantic import BaseModel, ConfigDict, Field
 
 
class SavedSearchCreate(BaseModel):
    """Body for POST /api/v1/saved_searches."""
 
    name: str = Field(..., max_length=200)
    query: str
    # `default_factory`, never `default={}` — a shared mutable default
    # would be aliased across every request that omits the field.
    filters: dict[str, Any] = Field(default_factory=dict)
 
 
class SavedSearchUpdate(BaseModel):
    """Body for PATCH /api/v1/saved_searches/{id}. All fields optional."""
 
    name: str | None = Field(default=None, max_length=200)
    query: str | None = Field(default=None)
    filters: dict[str, Any] | None = Field(default=None)
 
 
class SavedSearchResponse(BaseModel):
    """One SavedSearch row in GET / list responses."""
 
    model_config = ConfigDict(from_attributes=True)
 
    id: uuid.UUID
    user_id: uuid.UUID
    name: str
    query: str
    filters: dict[str, Any]
    created_at: datetime
    updated_at: datetime
 
 
class SavedSearchList(BaseModel):
    """Paginated list response."""
 
    items: list[SavedSearchResponse]
    total: int
    page: int
    page_size: int

router.py

Every read goes through the repository — there is no session.get() path to a row. Every state change writes an AuditLog row. A row owned by someone else answers 404, not 403, on read, update and delete alike: 403 confirms the row exists, which is what an enumeration attack is trying to learn.

router.py
"""SavedSearch router — full CRUD owned by the current user.
 
INSPECTION RENDERING — not runnable as-is.
 
Generated from The Fabrica's private resource scaffold for a fictional
`SavedSearch` resource. The factory primitives it imports
(`CurrentUserDep`, `DBDep`, `NotFoundError`, `log_action_async`) are
intentionally omitted from this repository: it is meant to be inspected,
not executed.
 
Endpoints:
    POST   /api/v1/saved_searches          — create
    GET    /api/v1/saved_searches          — list (paginated, newest first)
    GET    /api/v1/saved_searches/{id}     — detail
    PATCH  /api/v1/saved_searches/{id}     — partial update
    DELETE /api/v1/saved_searches/{id}     — delete
 
Invariants this shape enforces:
 
    1. Every read goes through `SavedSearchRepository.{get,list}_for_user`,
       so ownership is filtered in SQL rather than checked after loading.
       There is no unscoped `session.get()` path to a row.
    2. Every state change writes an AuditLog row.
    3. AuditLog details NEVER contain user-supplied fields verbatim
       (avoids log injection and PII leakage). What is recorded is a
       curated set of "what changed" keys, not the request body.
    4. A row belonging to another user answers 404, never 403 — a 403
       would confirm that the id exists.
"""
 
from __future__ import annotations
 
import logging
import uuid
 
from fastapi import APIRouter, status
 
from src.api.deps import CurrentUserDep, DBDep
from src.api.exceptions import NotFoundError
from src.api.schemas._domain.saved_search import (
    SavedSearchCreate,
    SavedSearchList,
    SavedSearchResponse,
    SavedSearchUpdate,
)
from src.models._domain.saved_search import SavedSearch
from src.services.audit import log_action_async
from src.services.repositories_async._domain.saved_search import (
    SavedSearchRepository,
)
 
log = logging.getLogger(__name__)
 
router = APIRouter(prefix="/saved_searches", tags=["saved_searches"])
 
 
@router.post(
    "",
    response_model=SavedSearchResponse,
    status_code=status.HTTP_201_CREATED,
)
async def create_saved_search(
    body: SavedSearchCreate,
    user: CurrentUserDep,
    db: DBDep,
) -> SavedSearchResponse:
    """Create a new SavedSearch owned by the current user.
 
    Ownership comes from the authenticated principal, never from the
    request body — `SavedSearchCreate` has no `user_id` field to spoof.
    """
    row = SavedSearch(
        user_id=user.id,
        **body.model_dump(),
    )
    db.add(row)
    await db.flush()
    await log_action_async(
        db,
        actor=f"user:{user.clerk_user_id}",
        action="saved_search.created",
        user_id=user.clerk_user_id,
        resource_type="saved_search",
        resource_id=str(row.id),
        # Audit details are a curated allowlist — never the full body.
        details={"created_via": "api"},
    )
    return SavedSearchResponse.model_validate(row)
 
 
@router.get("", response_model=SavedSearchList)
async def list_saved_searches(
    user: CurrentUserDep,
    db: DBDep,
    page: int = 1,
    page_size: int = 50,
) -> SavedSearchList:
    """List the current user's SavedSearch rows, newest first.
 
    There is no "all rows" code path: the repository method takes
    `user_id` as a required keyword argument.
    """
    if page < 1:
        page = 1
    if page_size < 1 or page_size > 200:
        page_size = 50
    repo = SavedSearchRepository(db)
    rows = await repo.list_for_user(
        user_id=user.id,
        limit=page_size,
        offset=(page - 1) * page_size,
    )
    total = await repo.count_for_user(user_id=user.id)
    return SavedSearchList(
        items=[SavedSearchResponse.model_validate(r) for r in rows],
        total=total,
        page=page,
        page_size=page_size,
    )
 
 
@router.get("/{resource_id}", response_model=SavedSearchResponse)
async def get_saved_search(
    resource_id: uuid.UUID,
    user: CurrentUserDep,
    db: DBDep,
) -> SavedSearchResponse:
    """Detail view. 404 for foreign rows — never 403 (don't leak existence)."""
    repo = SavedSearchRepository(db)
    row = await repo.get_for_user(resource_id, user_id=user.id)
    if row is None:
        raise NotFoundError(f"SavedSearch {resource_id} not found")
    return SavedSearchResponse.model_validate(row)
 
 
@router.patch("/{resource_id}", response_model=SavedSearchResponse)
async def update_saved_search(
    resource_id: uuid.UUID,
    body: SavedSearchUpdate,
    user: CurrentUserDep,
    db: DBDep,
) -> SavedSearchResponse:
    """Partial update. Only fields set on the request body are touched."""
    repo = SavedSearchRepository(db)
    row = await repo.get_for_user(resource_id, user_id=user.id)
    if row is None:
        raise NotFoundError(f"SavedSearch {resource_id} not found")
 
    patch = body.model_dump(exclude_unset=True)
    for k, v in patch.items():
        setattr(row, k, v)
    await db.flush()
 
    await log_action_async(
        db,
        actor=f"user:{user.clerk_user_id}",
        action="saved_search.updated",
        user_id=user.clerk_user_id,
        resource_type="saved_search",
        resource_id=str(row.id),
        # Field NAMES, not values: an audit trail that records what
        # changed without copying user content into the log.
        details={"fields_changed": sorted(patch.keys())},
    )
    return SavedSearchResponse.model_validate(row)
 
 
@router.delete(
    "/{resource_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
async def delete_saved_search(
    resource_id: uuid.UUID,
    user: CurrentUserDep,
    db: DBDep,
) -> None:
    """Delete one row. 404 for foreign rows."""
    repo = SavedSearchRepository(db)
    row = await repo.get_for_user(resource_id, user_id=user.id)
    if row is None:
        raise NotFoundError(f"SavedSearch {resource_id} not found")
    await db.delete(row)
    await log_action_async(
        db,
        actor=f"user:{user.clerk_user_id}",
        action="saved_search.deleted",
        user_id=user.clerk_user_id,
        resource_type="saved_search",
        resource_id=str(resource_id),
        details={},
    )

migration.py

One migration per change, with a downgrade() that works and drops what the upgrade created. Indexes and constraints are named explicitly rather than left to defaults, which is what keeps a downgrade portable between machines.

migration.py
"""Add saved_searches table.
 
INSPECTION RENDERING — not runnable as-is.
 
Generated from The Fabrica's private resource scaffold for a fictional
`SavedSearch` resource.
 
REDACTION NOTE: the revision identifiers below (`revision` /
`down_revision`) have been NORMALISED for this inspection copy. In the
private repository they chain onto the real migration head; publishing
that identifier would leak the private migration graph. Only the
revision metadata was changed — the substantive schema (columns, types,
constraints, indexes, and the downgrade path) is exactly what the
scaffold produces.
 
Note what the migration guarantees at the DATABASE level, not merely in
the ORM: the CASCADE lives in the foreign key itself, so a user delete
removes these rows even if it is issued by a script that never loads
the SQLAlchemy model.
 
Revision ID: 0001_saved_searches
Revises: 0000_inspection_base
Create Date: normalised for inspection
 
"""
 
from __future__ import annotations
 
from collections.abc import Sequence
 
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
 
# revision identifiers, used by Alembic. Normalised — see the note above.
revision: str = "0001_saved_searches"
down_revision: str | Sequence[str] | None = "0000_inspection_base"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
 
 
def upgrade() -> None:
    """Create saved_searches table + composite index."""
    op.create_table(
        "saved_searches",
        sa.Column(
            "id",
            postgresql.UUID(as_uuid=True),
            primary_key=True,
            nullable=False,
        ),
        sa.Column(
            "user_id",
            postgresql.UUID(as_uuid=True),
            sa.ForeignKey("users.id", ondelete="CASCADE"),
            nullable=False,
        ),
        # === BEGIN COLUMNS_BLOCK ===============================================
        sa.Column("name", sa.String(200), nullable=False),
        sa.Column("query", sa.Text, nullable=False),
        sa.Column(
            "filters",
            postgresql.JSONB,
            nullable=False,
            server_default=sa.text("'{}'::jsonb"),
        ),
        # === END COLUMNS_BLOCK =================================================
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("CURRENT_TIMESTAMP"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("CURRENT_TIMESTAMP"),
            nullable=False,
        ),
    )
    op.create_index(
        "ix_saved_searches_user_id",
        "saved_searches",
        ["user_id"],
    )
    op.create_index(
        "ix_saved_searches_user_id_created_at",
        "saved_searches",
        ["user_id", "created_at"],
    )
 
 
def downgrade() -> None:
    """Drop saved_searches + indexes.
 
    The downgrade is written, not stubbed. The factory's CI runs
    `alembic upgrade head` → `downgrade base` → `upgrade head` on every
    change, so a migration that cannot be reversed fails the build.
    """
    op.drop_index(
        "ix_saved_searches_user_id_created_at",
        table_name="saved_searches",
    )
    op.drop_index("ix_saved_searches_user_id", table_name="saved_searches")
    op.drop_table("saved_searches")

Why this page is hand-written

Almost everything else on this site is extracted from the product’s own documentation. This page is not: it is real code, and code is not a passage of prose a document can carry. The same goes for the concurrency test transcript.

Hand-written means drift-prone, which is the failure this whole mechanism exists to remove. So the blocks above are checked against the files in examples/owned-resource/ on every build: edit one without the other and CI fails.

Last updated on
How these documents are maintainedThe concurrency test, and its output