Fastcrud

FastCRUD

FastCRUD generates async CRUD methods (and optionally CRUD endpoints) for SQLAlchemy 2.0 models inside a FastAPI app. The two entry points are:

• FastCRUD(Model) — the data-access class. Use this for per-row CRUD operations (single-record get / create / update / delete, paginated lists, joins with relationships). Real codebases use this in services, workers, and custom endpoints.
• crud_router(...) — returns an APIRouter with create/read/update/delete endpoints auto-wired. Use this only when you want generated endpoints; many apps skip it and hand-roll FastAPI routes on top of FastCRUD instances.

This skill covers the public API across fastcrud >= 0.22; features marked (v0.23+) need fastcrud >= 0.23. SQLAlchemy is the default ORM in examples; SQLModel works the same way except where noted.

Before reaching for FastCRUD, check the When NOT to use FastCRUD section below. Aggregate roll-ups, CTEs, GROUP BY projections, and bulk writes are SQLAlchemy's job — FastCRUD is for per-row CRUD.

Canonical setup

The minimal viable pattern. Always start here, then add filters/joins/relationships as needed.

pythonCopy code
# models.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, relationship

class Base(DeclarativeBase): pass

class Tier(Base):
    __tablename__ = "tier"
    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)
    tier_id = Column(Integer, ForeignKey("tier.id"))
    tier = relationship("Tier")

pythonCopy code
# schemas.py
from pydantic import BaseModel, ConfigDict

class UserCreate(BaseModel):
    name: str
    email: str
    tier_id: int

class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    email: str
    tier_id: int

class UserUpdate(BaseModel):
    name: str | None = None
    email: str | None = None

pythonCopy code
# main.py
from fastapi import FastAPI
from fastcrud import crud_router
from .db import get_async_session
from .models import User
from .schemas import UserCreate, UserRead, UserUpdate

app = FastAPI()

user_router = crud_router(
    session=get_async_session,    # callable returning AsyncSession (FastAPI dependency)
    model=User,
    create_schema=UserCreate,
    update_schema=UserUpdate,
    select_schema=UserRead,        # used for read responses
    path="/users",
    tags=["users"],
)
app.include_router(user_router)

This produces POST /users, GET /users/{id}, PATCH /users/{id}, DELETE /users/{id}, DELETE /users/db_delete/{id} (hard delete), and GET /users (paginated list).

Choosing the right method

When using FastCRUD(Model) directly, pick the method that matches the data shape you actually need. Do not loop over get_multi() and call get_joined() per row — that's the canonical N+1.

Avoiding N+1 — the most common mistake

FastCRUD has two mechanisms to prevent N+1 when fetching related data. Use one of them — never iterate.

1. Auto-detect relationships (zero config, default for v0.21+)

Set include_relationships=True on crud_router (or auto_detect_relationships=True on the FastCRUD method directly). FastCRUD inspects the SQLAlchemy mapper, builds the joins, and threads them through a single query.

pythonCopy code
crud_router(
    session=get_async_session,
    model=User,
    create_schema=UserCreate,
    update_schema=UserUpdate,
    select_schema=UserWithTier,           # schema includes nested tier field
    include_relationships=True,           # auto-include all relationships
    path="/users",
)

Or pass a list to include only specific relationships:

pythonCopy code
include_relationships=["tier", "department"]   # whitelist

Gotcha: One-to-many relationships are excluded by default because they can return unbounded data. Opt in explicitly:

pythonCopy code
include_relationships=True,
include_one_to_many=True,
default_nested_limit=10,    # cap nested rows per parent (uses SQL window function)

default_nested_limit uses row_number() OVER (PARTITION BY ...) at the database level — it does not fetch everything and slice in Python.

2. Manual JoinConfig (when you need explicit control)

For self-joins, aliases, custom join conditions, or per-join schemas:

pythonCopy code
from fastcrud import FastCRUD, JoinConfig

orders_with_user = await crud.get_multi_joined(
    db=session,
    joins_config=[
        JoinConfig(
            model=User,
            join_on=Order.user_id == User.id,
            join_prefix="user_",         # avoid column collisions
            schema_to_select=UserRead,
        ),
    ],
    offset=0,
    limit=20,
)

For one-to-many with per-parent capping:

pythonCopy code
JoinConfig(
    model=Article,
    join_on=Article.author_id == Author.id,
    relationship_type="one-to-many",
    sort_columns="created_at",
    sort_orders="desc",
    nested_limit=5,         # 5 most recent articles per author, computed in SQL
)

See references/joins.md for the full JoinConfig spec, polymorphic inheritance, and the auto-detection rules.

limit=None — the second biggest footgun

get_multi(...) / get_multi_joined(...) accept limit=None to fetch every matching row. This is the single biggest cause of latency cliffs and OOMs in real-world FastCRUD code.

Rule: limit=None is only safe when a WHERE clause domain-bounds the result to a known-small set. Otherwise, always pass an explicit numeric limit.

Safe — scoped by a WHERE clause

pythonCopy code
# OK: scoped to a single project → bounded by N(clips per project)
clips = await crud_clips.get_multi(db, limit=None, project_id=project.id)

# OK: scoped to a small known set of IDs
tiers = await crud_tiers.get_multi(db, limit=None, id__in=tier_ids)

Unsafe — unbounded growth

pythonCopy code
# BAD: grows linearly with the user table forever
users = await crud_users.get_multi(db, limit=None)

# GOOD: HTTP endpoint with a hard ceiling
limit = min(requested_limit, MAX_PAGE_LIMIT)
users = await crud_users.get_multi(db, offset=offset, limit=limit)

Pattern: sanity-check limit=None results

When limit=None is justified, log when results unexpectedly explode so drift is caught before it becomes an incident:

pythonCopy code
def handle_query_sanity(items: list, threshold: int, context: str) -> None:
    if len(items) > threshold:
        logger.warning("query growth: %d > %d threshold (%s)",
                       len(items), threshold, context)

entitlements = await crud_entitlements.get_multi(db, limit=None, user_id=user.id)
handle_query_sanity(entitlements["data"], threshold=100, context="user_entitlements")

Three-tier default

When NOT to use FastCRUD

FastCRUD is the right tool for per-row CRUD on a single model, optionally with relationship joins. Drop to raw SQLAlchemy when the query shape is anything else. In a typical app the split is roughly 80/20 — most data access is FastCRUD's territory, but the 20% that isn't is firmly SQLAlchemy's.

crud_router itself is optional. Production codebases often use FastCRUD instances as the data-access layer behind hand-written FastAPI routes, skipping crud_router entirely so they can compose auth, validation, and business logic without subclassing EndpointCreator.

Anti-patterns to refactor on sight

1. Per-ID loop calling crud.get() — classic N+1.

   pythonCopy code
   # BAD
   for tier_id in tier_ids:
       tier = await crud_tiers.get(db, id=tier_id)
   # GOOD — one query
   tiers = await crud_tiers.get_multi(db, id__in=tier_ids, limit=None)
   

2. db.scalar(select(Model).where(...).exists()) — use await crud.exists(**filters).

3. get_multi(limit=None) then Python filtering — push the filter into kwargs so the database does the work.

4. crud.get() in a hot path without schema_to_select — selects every column. Pass a minimal schema to cut payload size.

5. Loop of crud.create() — calls go one-at-a-time. For >10 rows use crud.upsert_multi(...) or raw db.execute(insert(Model), records).

6. Hand-written select(...).where(...) for an OR group, a column comparison or an EXISTS (v0.23+) — pass the condition as where= to the FastCRUD method instead: await crud.get_multi(db, where=(Post.owner_id == me) | Post.published).

where= and row-level policies (v0.23+)

Every read and write takes where=: one SQLAlchemy condition or a list of them, AND-ed with the keyword filters. Use it for what keyword filters can't express — OR groups, column-to-column comparisons, .any() / .has() / exists() subqueries:

pythonCopy code
active_or_mine = await post_crud.get_multi(
    db, where=Post.published | (Post.owner_id == me), is_deleted=False
)

JoinConfig.where chooses which joined rows attach to each parent, never which parents come back (JoinConfig.filters does that); CountConfig.where chooses which rows are counted.

FastCRUD(Model, policy=PolicyConfig(...)) — or crud.with_policy(policy) per request — restricts every operation the way Postgres row-level security does: rows outside the policy behave as absent, writes can only reach rows the caller can read, and create_check / update_check validate written rows. crud_router(policy=...) resolves a policy per request and turns AccessDenied into a 403. Read references/policies.md before writing policy code.

Filter syntax

FastCRUD accepts filters as keyword arguments on every read/update/delete method, and via FilterConfig for crud_router's GET /list endpoint.

Operator suffix on the field name

The operator goes after a double underscore:

pythonCopy code
await crud.get_multi(db, price__gte=10, name__ilike="%admin%", id__in=[1, 2, 3])

Joined filters

Walk relationships with .:

pythonCopy code
await crud.get_multi_joined(db, **{"tier.name__eq": "premium"})

Or, equivalently, build the kwargs dict literal-style. In FilterConfig:

pythonCopy code
FilterConfig({
    "name__ilike": None,           # query string param: ?name__ilike=foo
    "price__gte": None,
    "tier.name": None,             # joined filter — auto-includes the tier relationship
})

Bool coercion

?filter=false is correctly parsed (case-insensitive: False, FALSE, 0, true, TRUE, 1). No need to pre-coerce.

See references/filters.md for FilterConfig deep-dive, Depends(...) filters (filter by current user automatically), and custom operator registration.

Return semantics — read carefully

The biggest footgun. Several methods have non-obvious defaults that changed in v0.20:

create()

• Without schema_to_select: returns None (since v0.20.0 — was the model in earlier versions).
• With schema_to_select: returns a dict.
• With schema_to_select + return_as_model=True: returns a Pydantic instance.

pythonCopy code
await crud.create(db, UserCreate(...))                                     # → None
await crud.create(db, UserCreate(...), schema_to_select=UserRead)          # → dict
await crud.create(db, UserCreate(...), schema_to_select=UserRead,
                  return_as_model=True)                                    # → UserRead

return_as_model=True requires schema_to_select (else raises ValueError).

get() / get_multi() / get_joined() / etc.

Same pattern: return_as_model=True requires schema_to_select. Without schema_to_select, all columns are returned as a dict.

schema_to_select propagates the subclass type (v0.22+)

If you pass a subclass override per-call, the return type narrows correctly — no manual cast() needed:

pythonCopy code
class UserAdminRead(UserRead):
    role: str

result = await crud.get(db, id=1, schema_to_select=UserAdminRead, return_as_model=True)
# result is typed as UserAdminRead | None, not UserRead | None

update(return_columns=...) must be a list (v0.22+)

return_columns=True raises ValueError. Pass a list:

pythonCopy code
await crud.update(db, schema=UserUpdate(name="X"), return_columns=["id", "name"], id=1)

commit=False for transactions

All write methods take commit=False. Use it when chaining multiple operations under a single transaction; commit once at the end with await db.commit().

Soft delete

Configure at the FastCRUD level:

pythonCopy code
crud = FastCRUD(User, is_deleted_column="is_deleted", deleted_at_column="deleted_at")

await crud.delete(db, id=1)       # soft: sets is_deleted=True, deleted_at=now
                                  #       (requires both columns on the model)
await crud.db_delete(db, id=1)    # hard: actual DELETE — always removes the row

Reads do NOT auto-filter soft-deleted rows. This is the most common surprise — you have to filter explicitly:

pythonCopy code
active = await crud.get_multi(db, is_deleted=False)

If you want every read to exclude soft-deleted rows automatically, wrap FastCRUD in a subclass that overrides get/get_multi/etc. to add the filter, or pass is_deleted=False as a dependency-based default via FilterConfig on crud_router.

Gotchas (read these before writing code)

1. limit=None fetches every matching row. Only safe when a WHERE clause domain-bounds the result. See the dedicated section.
2. Async session only. All methods require AsyncSession. Sync Session support for count()/exists() is in flight (PR #333) but not merged.
3. One-to-many relationships excluded from auto-detect. Add include_one_to_many=True and set default_nested_limit (or nested_limit per JoinConfig).
4. Filter operator separator is __ (two underscores). price_gte is just a column name price_gte; price__gte is price >= ....
5. create() returns None by default (no schema_to_select). Don't expect the model back.
6. return_as_model=True without schema_to_select raises. Always pair them.
7. update(return_columns=True) raises since v0.22.0. Use a list of column names.
8. Joined filters need . not __ for the relationship part. "tier.name__eq", not "tier__name__eq".
9. session parameter on crud_router is a callable (a FastAPI dependency that yields/returns an AsyncSession), not a session instance.
10. include_relationships and joins_config are mutually exclusive on crud_router. Pick one.
11. Joined-table polymorphic inheritance is supported. v0.22.0 fixed auto-aliasing when primary and joined share a base table; v0.22.2 fixed inherited columns missing from create() responses.
12. SQLModel joined-table inheritance is fragile — redeclaring id in a subclass breaks the SQLModel metaclass mapping. Stick to plain SQLAlchemy if you need polymorphism.
13. Python 3.10+ required. Python 3.14 works since v0.22.1 (earlier versions crashed at import under PEP 649).
14. update / delete / db_delete refuse to run with no filters (v0.23+), including where=[]. Pass where=true() to reach every row on purpose.
15. A condition naming a model the query doesn't join raises ValueError (v0.23+) instead of returning a cartesian product. Join the model, or use .any() / .has().
16. db_delete raises NoResultFound when nothing matches (v0.23+), like delete and update.
17. where and server_data are reserved keyword arguments (v0.23+): where on every read and write, so a column named where can't be filtered by keyword (use where=Model.where == value); server_data only on update, the one method that takes both it and keyword filters.

When to drill into references

Load these on demand:

• references/methods.md — full signatures and overloads for every FastCRUD method
• references/filters.md — every operator, FilterConfig, dependency-based filters, custom operators
• references/joins.md — JoinConfig fields, auto-detection rules, nested_limit mechanics, polymorphism
• references/pagination.md — offset vs cursor pagination, paginated_response, CursorPaginatedRequestQuery
• references/endpoints.md — crud_router full signature, EndpointCreator subclassing, included_methods, per-method dependencies, soft delete, custom endpoint names
• references/policies.md — PolicyConfig, AccessDenied, with_policy, column rules, server_data, upserts and endpoints under a policy (v0.23+)

SQLModel notes (when the project uses SQLModel instead)

The library works the same way; substitute the model definition:

pythonCopy code
from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    __tablename__ = "user"
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)
    tier_id: int | None = Field(default=None, foreign_key="tier.id")

Schemas can also be SQLModel (without table=True) instead of BaseModel. The CRUD layer is identical because SQLModel inherits from SQLAlchemy. One caveat:

• Joined-table inheritance is unreliable — see gotcha #11. For polymorphic models, drop to plain SQLAlchemy.
• Conditions need col() for type checkers (v0.23+) — SQLModel types Hero.age > 18 as bool, so write col(Hero.age) > 18 in where=, JoinConfig.where and policy rules.