FastAPI + SQLAlchemy Best Practices

Most long-term problems, both in development and in production come from rushed model design, scattered logics, and patterns that feel convenient at first but become expensive later.

As Robert C. Martin writes in Clean Architecture, “getting something to work—once—just isn’t that hard, getting it right is another matter entirely.” That is exactly the point here. Making an API work is not the hard part. Keeping a codebase readable and easy to maintain as it grows is the harder part. In this post I go through a set of practical best practices for FastAPI + SQLAlchemy applications that I have learnt from projects. They are patterns that helps keeping codebases maintainable and pleasant to work with over time.

This post covers best practices in these topics:

1. Optimal number of models per bounded context
2. Keep reusable query logic close to the model
3. Empty values in string columns
4. Keep Alembic migrations clean
5. Public UUIDs for API-facing identifiers
6. SQLAlchemy expression tools instead of Python loops
7. Generic pk access on the base model
8. Model and relationship naming
9. Use a yield dependency with async_sessionmaker
10. Keep transaction boundaries explicit
11. Avoid accidental lazy loading and N+1 queries
12. Do not commit inside low-level helpers
13. Avoid COUNT(*) when it is unnecessary
14. Keep business logic in one place
15. Use with_for_update() wisely
16. Enforce data consistency at the database level
17. Make Full Use of Dependency Injection to Keep Routes Clean and Thin
18. Set Explicit Naming Conventions for Database Constraints

1. Keep The Number Of Models Per Bounded Context Reasonable

If you have twenty unrelated models in a single models.py, your codebase is trying to tell you something.

In FastAPI applications, it is common to start with a flat structure and then let it grow without boundaries. That works for very small apps, but it becomes messy fast. Prefer splitting your codebase by domain or bounded context:

app/
  billing/
    models.py
    schemas.py
    service.py
    routes.py
  catalog/
    models.py
    schemas.py
    service.py
    routes.py

There is no magical hard limit, but once a module holds around ten models, it is worth asking whether more than one business domain has been mixed together. See this link for preferred fastapi project structure

2. Encapsulate Reusable Queries

In Django, custom managers are a natural place for repeated query logic. In SQLAlchemy the equivalent idea is to centralize reusable statements instead of scattering ad-hoc select() blocks across route handlers.

A nice pattern is to keep small query helpers close to the model as @classmethods that return a Select object. That gives you manager-like ergonomics without making the model responsible for session lifecycle, transactions, or endpoint behavior.

For example, avoid this repeated pattern:

@app.get("/posts/published")
async def get_published_posts(session: SessionDep):
    stmt = select(Post).where(Post.status == "published")
    return (await session.scalars(stmt)).all()


@app.get("/authors/{author_id}/posts")
async def get_author_posts(author_id: int, session: SessionDep):
    stmt = (
        select(Post)
        .where(Post.status == "published", Post.author_id == author_id)
        .order_by(Post.created_at.desc())
    )
    return (await session.scalars(stmt)).all()

Reusable statement builders, and putting them on the model is often a nice option:

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    status: Mapped[str] = mapped_column(String(20))
    title: Mapped[str] = mapped_column(String(200))
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), nullable=False
    )

    @classmethod
    def published_stmt(cls):
        return select(cls).where(cls.status == "published")

    @classmethod
    def published_by_author_stmt(cls, author_id: int):
        return (
            cls.published_stmt()
            .where(cls.author_id == author_id)
            .order_by(cls.created_at.desc(), cls.id.desc())
        )

Then let a service use those model helpers:

async def get_published_posts_by_author(
    session: AsyncSession,
    author_id: int,
) -> list[Post]:
    stmt = Post.published_by_author_stmt(author_id)
    return (await session.scalars(stmt)).all()

And keep the route thin:

@app.get("/authors/{author_id}/posts")
async def get_author_posts(author_id: int, session: SessionDep):
    return await get_published_posts_by_author(session, author_id)

This keeps separation of concerns clean. The route calls a service, and the service composes model-level query helpers. It gives you several benefits:

• query intent stays close to the model it belongs to
• routes stay focused on HTTP concerns
• services coordinate application use cases
• the returned object is still just a SQLAlchemy statement, so it composes well
• query behavior is easier to test and harder to accidentally fork into slightly different versions

What we should avoid is putting session-bound methods directly on the model, such as:

@classmethod
async def get_published(cls, session: AsyncSession):
    return (await session.scalars(select(cls).where(cls.status == "published"))).all()

That starts mixing model definition with I/O and transaction concerns. If a model method executes the session internally and returns a raw list of objects, you lose composability, it means you can not apply more filter on get_published(). Returning a statement is usually the cleaner boundary, and it keeps the code easier to test and maintain.

3. Avoid Two Different Meanings Of “No String Value”

One of the most common data quality problems in APIs is allowing both NULL and "" to mean “missing text”. Pick one representation and normalize input consistently.

In SQLAlchemy, using None / NULL is usually the cleanest choice for optional string data:

class CustomerCreate(BaseModel):
    middle_name: str | None = None

    @field_validator("middle_name")
    @classmethod
    def normalize_middle_name(cls, value: str | None) -> str | None:
        if value is None:
            return None
        value = value.strip()
        return value or None


class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    middle_name: Mapped[str | None] = mapped_column(String(100), nullable=True)

For fields where empty string is a meaningful value, keep nullable=False and store "" consistently. The important part is not which convention you choose. The important part is avoiding both at the same time.

For optional unique string fields, NULL is usually the better choice than "".

4. Keep Alembic Migrations Clean

FastAPI + SQLAlchemy projects usually rely on Alembic for schema changes. A common mistake is letting migration history become noisy and chaotic during active feature development.

For example, one feature branch might generate these revisions:

001_add_order_table.py
002_fix_order_table.py
003_fix_order_table_again.py
004_add_missing_index.py

Before merging a private branch, squash all migration generated in the brnach into one clean revision. However, there exist cases where the migration should be applied in more than one phase, in that cases having more than one migration file is inevitable.

Typical flow:

alembic revision --autogenerate -m "add orders table"
alembic upgrade head

Best practice:

• keep branch-local migrations tidy
• write descriptive revision names
• do not rewrite migrations already used by teammates or production
• prefer one clean migration per coherent change instead of five correction migrations in a row

Alembic does not have the same built-in “squash migrations” workflow Django has, so cleanup usually means reorganizing revisions before they become shared history.

5. Use Two Identifiers: Private PK And Public UUID

In API-driven applications, it is often useful to keep a compact internal primary key and expose a public UUID externally.

Example:

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    public_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        default=uuid.uuid4,
        unique=True,
        nullable=False,
        index=True,
    )
    status: Mapped[str] = mapped_column(String(30), nullable=False)

Then your public routes can use public_id:

@app.get("/orders/{public_id}")
async def get_order(public_id: uuid.UUID, session: SessionDep):
    stmt = select(Order).where(Order.public_id == public_id)
    return await session.scalar(stmt)

This pattern gives us:

• small and efficient internal joins on integer keys
• stable public identifiers
• less leakage of record counts and ordering

Also note the use of SQLAlchemy’s backend-agnostic Uuid type. In SQLAlchemy 2.x, you no longer need custom GUID hacks for common cross-database UUID storage.

Although there are valid distributed-system cases for UUID or ULID primary keys, for many business apps, integer primary key plus public UUID is a very pragmatic default.

6. Use SQL Expressions Instead Of Processing Rows In Python

If the database can do the filtering, comparison, aggregation, or update, let it do it.

Bad practice:

students = []
result = await session.scalars(select(Student))
for student in result:
    if student.math_score > student.english_score:
        students.append(student)

Better:

stmt = select(Student).where(Student.math_score > Student.english_score)
students = (await session.scalars(stmt)).all()

SQLAlchemy gives you rich expression tools such as:

• func for SQL functions
• subqueries and CTEs
• SQL expressions in updates
• relationship loader options

Another useful example is atomic updates:

post = await session.get(Post, post_id)
post.view_count = Post.view_count + 1
await session.commit()

Letting the database express the change is safer than reading a value in Python, incrementing it, and writing it back later.

7. Add A Generic pk Property On Your Base Model

One of the nicest ideas in Django models is obj.pk. It lets you refer to the primary key generically without coupling your code to a specific column name like id, uuid, or invoice_number.

You can get the same convenience in SQLAlchemy with a hybrid property on your declarative base:

from sqlalchemy import inspect
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    @classmethod
    def _get_pk_field(cls):
        mapper = inspect(cls)
        fields = mapper.primary_key

        if len(fields) != 1:
            raise ValueError(f"{cls.__name__} has composite primary key")

        return fields[0]

    @hybrid_property
    def pk(self):
        pk_field = type(self)._get_pk_field()
        return getattr(self, pk_field.key)

    @pk.expression
    def pk(cls):
        return cls._get_pk_field()

This has the same benefits Django developers enjoy:

• your code stays generic even when different models use different primary-key names
• pk expresses intent better than hardcoding id
• the same API works on instances and inside SQL expressions
• schema refactors are less painful because calling code does not care what the PK column is called

Example model:

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

Instance access:

todo = Todo(title="Write the post")
todo.pk

Class-level SQL expression:

Todo.pk

Comparison expression:

Todo.pk == 1

Selecting the primary key generically:

select(Todo.pk)

Filtering by primary key without hardcoding id:

stmt = select(Todo).where(Todo.pk == 1)
todo = await session.scalar(stmt)

When the primary key is not named id:

class Invoice(Base):
    __tablename__ = "invoices"

    invoice_number: Mapped[str] = mapped_column(String(50), primary_key=True)
    amount_cents: Mapped[int]

Now the calling code stays consistent:

stmt = select(Invoice).where(Invoice.pk == "INV-2026-0001")
invoice = await session.scalar(stmt)

session.get() is still a great API for direct primary-key retrieval when you already have the PK value in hand. But adding Base.pk gives your models the same expressive, generic primary-key interface that makes Django code pleasant to write and maintain.

One improvement would be to define two reusable base classes: a primary base class without an id field, which still provides the pk property described above, and a second base class that inherits from it and adds id as the default primary key. This gives each model the flexibility to either use the conventional integer id primary key or define its own primary key, such as a UUID, slug, or another natural key.

from sqlalchemy import inspect
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    @classmethod
    def _get_pk_field(cls):
        mapper = inspect(cls)
        fields = mapper.primary_key

        if len(fields) != 1:
            raise ValueError(f"{cls.__name__} has composite primary key")

        return fields[0]

    @hybrid_property
    def pk(self):
        pk_field = type(self)._get_pk_field()
        return getattr(self, pk_field.key)

    @pk.expression
    def pk(cls):
        return cls._get_pk_field()


class BaseWithId(Base):
    __abstract__ = True

    id: Mapped[int] = mapped_column(primary_key=True)


class User(BaseWithId):
    __tablename__ = "users"

    name: Mapped[str]


class Country(Base):
    __tablename__ = "countries"

    code: Mapped[str] = mapped_column(primary_key=True)
    name: Mapped[str]

In this example, User uses the default integer id, while Country uses code as its primary key. Both models can still access their primary-key value through the shared pk property.

8. Name Models And Relationships Clearly

Naming matters because ORM code is read far more often than it is written. A few practical conventions:

• class names should be singular: User, Order, BlogPost
• table names should be predictable: users, orders, blog_posts
• scalar relationships should be singular: author, profile, customer
• collection relationships should be plural: posts, items, roles

Example:

class Author(Base):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    posts: Mapped[list["BlogPost"]] = relationship(back_populates="author")


class BlogPost(Base):
    __tablename__ = "blog_posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    title: Mapped[str] = mapped_column(String(200))

    author: Mapped[Author] = relationship(back_populates="posts")

This is clearer than ambiguous names like post_data, author_ref, or blogpostlist.

9. Use A yield Dependency With async_sessionmaker

FastAPI’s documentation recommends using dependencies with yield when you need setup and teardown around a request. For database sessions, that maps perfectly to SQLAlchemy sessions: create the session before yield, and make sure it is closed after the request finishes.

For modern async SQLAlchemy code, the cleanest version is:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine


engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_pre_ping=True,
)

SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def get_session():
    async with SessionLocal() as session:
        yield session

It is better to use FastAPI’s yield dependency pattern for request-scoped setup and cleanup. And use SQLAlchemy’s async_sessionmaker, which is the dedicated SQLAlchemy 2.x factory for AsyncSession. It’s clearer and more modern than using generic sessionmaker(..., class_=AsyncSession)

Benefits:

• each request gets its own session
• the session is closed automatically after the request
• the code is small and easy to understand

If you prefer manual cleanup, this is also valid:

async def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        await session.close()

But async with SessionLocal() as session: is usually cleaner and less error-prone.

Also, an AsyncSession should not be shared across concurrent tasks. The safe mental model is one session per request, and if you spawn concurrent tasks, each task should have its own session.

10. Keep Transaction Boundaries Explicit

This is one of the most important SQLAlchemy habits to get right in FastAPI apps: understand when you need flush() and when you need commit().

• flush() sends pending SQL to the database inside the current transaction
• commit() permanently finishes the transaction

Many bugs come from calling commit() too early just to get a generated primary key or to make a later insert work.

Bad:

@app.post("/orders")
async def create_order(payload: OrderCreate, session: SessionDep):
    order = Order(customer_id=payload.customer_id, status="pending")
    session.add(order)
    await session.commit()

    item = OrderItem(order_id=order.id, sku=payload.sku, quantity=payload.quantity)
    session.add(item)
    await session.commit()
    return order

Good:

@app.post("/orders")
async def create_order(payload: OrderCreate, session: SessionDep):
    order = Order(customer_id=payload.customer_id, status="pending")
    session.add(order)
    await session.flush()

    item = OrderItem(order_id=order.id, sku=payload.sku, quantity=payload.quantity)
    session.add(item)

    await session.commit()
    await session.refresh(order)
    return order

It is better because:

• one request owns one transaction
• you still get order.id after flush()
• if creating the item fails, the whole unit of work can roll back together
• transaction behavior stays predictable

Use flush() when you need database-generated values or constraint checks before the end of the request. Use commit() at the use-case boundary.

11. Avoid Accidental Lazy Loading And N+1 Queries

FastAPI endpoints serialize lists of ORM objects into JSON. That is exactly where accidental lazy loading can turn one query into fifty (N+1 query issue).

Example problem:

@app.get("/posts")
async def get_posts(session: SessionDep):
    posts = (await session.scalars(select(Post))).all()

    return [
        {
            "id": post.id,
            "title": post.title,
            "author_name": post.author.name,
        }
        for post in posts
    ]

If author is lazily loaded, every post.author access can trigger another query hitting DB.

Prefer explicit loader strategies:

from sqlalchemy.orm import selectinload


@app.get("/posts")
async def get_posts(session: SessionDep):
    stmt = select(Post).options(joinedload(Post.author))
    posts = (await session.scalars(stmt)).all()

    return [
        {
            "id": post.id,
            "title": post.title,
            "author_name": post.author.name,
        }
        for post in posts
    ]

joinedload() fetches the parent and related rows in one SQL query using a JOIN, while selectinload() fetches the parent rows first and then loads related rows in a second query with IN (...).
A practical rule of thumb is this: prefer selectinload() for collections like posts.comments, users.roles, or orders.items; consider joinedload() for scalar relationships like post.author or order.customer, especially when the result set is small.
If a JOIN would duplicate parent rows heavily, selectinload() is usually the better choice; if you need everything in one compact query and the relationship is small, joinedload() can be a very good fit.

12. Do Not Commit Inside Low-Level Helpers

This item is very similar to item 10, just in other words. Don’t call commit() in helper functions or repository methods just because it feels convenient. That convenience becomes a maintenance problem quickly.

Bad:

async def create_user(session: AsyncSession, email: str) -> User:
    user = User(email=email)
    session.add(user)
    await session.commit()
    return user


async def create_profile(session: AsyncSession, user: User) -> Profile:
    profile = Profile(user_id=user.id)
    session.add(profile)
    await session.commit()
    return profile

Now a higher-level flow cannot treat both steps as one atomic operation.

Better:

async def create_user(session: AsyncSession, email: str) -> User:
    user = User(email=email)
    session.add(user)
    await session.flush()
    return user


async def create_profile(session: AsyncSession, user: User) -> Profile:
    profile = Profile(user_id=user.id)
    session.add(profile)
    await session.flush()
    return profile


@app.post("/users")
async def create_user_endpoint(payload: UserCreate, session: SessionDep):
    user = await create_user(session, payload.email)
    await create_profile(session, user)
    await session.commit()
    await session.refresh(user)
    return user

Benefits:

• the request or service layer owns the transaction
• multi-step workflows can succeed or fail as one unit
• helper functions become easier to compose and test
• rollback behavior is much clearer

Low-level helpers should usually add objects, build statements, and maybe flush(). The code that represents the full use case should decide when to commit().

13. Avoid COUNT(*) When You Only Need To Know Whether Something Exists

COUNT(*) is correct when you need the actual number. But for existence checks it is unnecessary work with cost.

Bad:

stmt = select(func.count()).select_from(Post).where(Post.slug == slug)
post_count = await session.scalar(stmt)
exists = post_count > 0

Good:

from sqlalchemy import exists

stmt = select(exists().where(Post.slug == slug))
post_exists = await session.scalar(stmt)

Use COUNT(*) for analytics, dashboards, and pagination totals. Do not use it as a habit for yes/no checks.

14. Keep Business Logic in one place

Item 4 focused on building a reusable query API. This item focuses on something different: reusable domain behavior. Business rules related to a model should have one clear home. They should not be duplicated across route handlers, serializers, background jobs, CLI commands, and utility functions.

For example, suppose creating a Device requires generating a serial number when one is not provided. Every code path that creates a device should use the same implementation of that rule. Otherwise, the behavior will eventually become inconsistent as the application evolves.

Depending on the application’s architecture, this logic can live in a model classmethod, or in a repository method when database access is separated from domain models. The important part is not the specific layer. The important part is that the application exposes one clear entry point for the behavior. This follows several core engineering principles:

• DRY: the rule lives in one place
• single responsibility: routes focus on input/output, not domain behavior
• encapsulation: the model manages its own valid state
• maintainability: future rule changes happen in one place

In SQLAlchemy, the clean equivalent is usually repository method, or model classmethods that perform model-specific work without owning the final transaction boundary (remember Item 10 of the list).

Example:

class Device(Base):
    __tablename__ = "devices"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    serial_number: Mapped[int] = mapped_column(unique=True, nullable=False)

    @classmethod
    async def create(
        cls,
        session: AsyncSession,
        *,
        name: str,
        serial_number: int | None = None,
    ) -> "Device":
        if serial_number is None:
            max_serial_number = await session.scalar(select(func.max(cls.serial_number)))
            serial_number = (max_serial_number or 0) + 1

        device = cls(name=name, serial_number=serial_number)
        session.add(device)
        await session.flush()
        return device

Then let a service own the use case:

async def create_device_service(
    session: AsyncSession,
    payload: DeviceCreate,
) -> Device:
    device = await Device.create(
        session,
        name=payload.name,
        serial_number=payload.serial_number,
    )
    await session.commit()
    await session.refresh(device)
    return device

And again, keep the endpoint thin:

@app.post("/devices")
async def create_device(payload: DeviceCreate, session: SessionDep):
    return await create_device_service(session, payload)

In projects where SQLAlchemy models are used primarily as persistence structures and keep database operations in repositories, the same creation rule can live in a DeviceRepository.

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession


class DeviceRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def create(
        self,
        *,
        name: str,
        serial_number: int | None = None,
    ) -> Device:
        if serial_number is None:
            serial_number = await self._next_serial_number()

        device = Device(
            name=name,
            serial_number=serial_number,
        )

        self.session.add(device)
        await self.session.flush()

        return device

    async def _next_serial_number(self) -> int:
        max_serial_number = await self.session.scalar(
            select(func.max(Device.serial_number))
        )
        return (max_serial_number or 0) + 1

The service uses the repository while retaining control over the transaction:

async def create_device_service(
    session: AsyncSession,
    payload: DeviceCreate,
) -> Device:
    repository = DeviceRepository(session)

    device = await repository.create(
        name=payload.name,
        serial_number=payload.serial_number,
    )

    await session.commit()
    await session.refresh(device)

    return device

Benefits are:

• every caller (API endpoints, CLI commands, tests, and background jobs) goes through the same rule and they stay consistent.
• services/route handlers remain thinner and easier to reason about
• model behavior is easier to discover because it lives near the data it controls

The purpose of the above example is merely to show the business rule should not be scattered across callers. For cases where the rule depends on globally unique number generation under concurrency, max(...) + 1 is usually not strong enough for high-traffic systems. In that case, prefer a database sequence, identity column, or another database-enforced allocation strategy.

15. Use with_for_update() Wisely

Locking is not just an ORM detail. It’s application behavior and it becomes part of your business behavior. The moment you introduce row locks, you are deciding whether other requests should wait, fail, retry, or skip work. In SQLAlchemy, with_for_update() is one of the most useful tools for preventing race conditions, but it is also easy to misuse.

Consider this example:

from sqlalchemy.orm import joinedload

stmt = (
    select(Order)
    .options(joinedload(Order.customer), joinedload(Order.items))
    .where(Order.id == order_id)
    .with_for_update()
)
order = await session.scalar(stmt)

Depending on the backend and generated SQL, broad joins can widen the locking scope more than you expect.

A safer pattern is to lock the base row explicitly and use narrower loading strategies:

from sqlalchemy.orm import selectinload

stmt = (
    select(Order)
    .options(selectinload(Order.items))
    .where(Order.id == order_id)
    .with_for_update(of=Order.__table__)
)
order = await session.scalar(stmt)

Two options matter a lot in real applications:

nowait=True

nowait=True means: try to acquire the lock immediately, and if another transaction already holds it, fail instead of waiting. It is a good fit when waiting several seconds would be worse than returning a clear conflict.

Example use cases:

• an admin tries to edit a record that is currently locked by another operation
• an inventory reservation flow should fail fast and ask the client to retry

from sqlalchemy.orm import selectinload

stmt = (
    select(Order)
    .options(selectinload(Order.items))
    .where(Order.id == order_id)
    .with_for_update(of=Order.__table__, nowait=True)
)
order = await session.scalar(stmt)

In this pattern, if the row is already locked, you can catch the database error and return something like HTTP 409 Conflict instead of tying up the request.

skip_locked=True

skip_locked=True means: if a row is already locked by another transaction, do not wait and do not fail. Just skip that row and move on to other eligible rows.

This is usually a better fit for worker-style flows than for user-facing endpoints.

Example use cases:

• multiple background workers pulling jobs from the same table
• batch processors claiming the next available tasks

Example:

stmt = (
    select(Job)
    .where(Job.status == "pending")
    .order_by(Job.created_at.asc(), Job.id.asc())
    .limit(10)
    .with_for_update(skip_locked=True)
)
jobs = (await session.scalars(stmt)).all()

This lets several workers claim different pending jobs in parallel without blocking each other.

The trade-off is important: skip_locked=True can silently skip work that is currently locked, so it is great for queues but usually wrong for flows where the caller expects a specific row to be processed right now.

Practical advice:

• lock as little as possible
• be careful when mixing row locks with eager joins
• keep the transaction short after acquiring the lock

16. Enforce Data Consistency at the Database Level

Application-layer validation is important, but it is not sufficient to guarantee data consistency. Multiple API instances, direct database access, and race conditions can all bypass application-level checks. Critical constraints must be enforced by the database itself.

Suppose your business rule requires start_at and end_at to occur on the same calendar day. A Pydantic input validation might look like this:

from pydantic import BaseModel, model_validator


class CreateBookingRequest(BaseModel):
    start_at: datetime
    end_at: datetime

    @model_validator(mode="after")
    def validate_same_day(self):
        if self.start_at.date() != self.end_at.date():
            raise ValueError(
                "start_at and end_at must be on the same day"
            )

        return self

Pydantic validation is very useful because it provides immediate, user-friendly feedback. However, it does not guarantee data integrity. Data can still be inserted through Database direct access, migrations, other services, or future code paths that forget to perform the validation. If the rule is truly required by the business domain, the database should enforce it as well.

from sqlalchemy import CheckConstraint
from sqlalchemy.orm import Mapped, mapped_column


class Booking(Base):
    __tablename__ = "bookings"

    id: Mapped[int] = mapped_column(primary_key=True)

    start_at: Mapped[datetime]
    end_at: Mapped[datetime]

    __table_args__ = (
        CheckConstraint(
            "DATE(start_at) = DATE(end_at)",
            name="ck_booking_same_day",
        ),
    )

Using the above constraint, now every write path must respect the rule, regardless of which application or user performs the insert. Catch the DB error regarding the constraint at appropriate place and raise domain-tailored Exception.

---

17. Make Full Use of Dependency Injection to Keep Routes Clean and Thin

FastAPI dependencies are not only for authentication or database sessions. They are a great way to move request-scoped preparation out of route handlers: query parameter validation, pagination, settings, feature flags, tenant validation, and service construction. This keeps routes focused on orchestration instead of low-level work.

from typing import Annotated

from fastapi import Depends, HTTPException, Query, status
from pydantic import BaseModel, Field


class Pagination(BaseModel):
    page: int = Field(default=1, ge=1)
    size: int = Field(default=20, ge=1, le=100)

    @property
    def limit(self) -> int:
        return self.size

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.size


def get_pagination(
    page: Annotated[int, Query(ge=1)] = 1,
    size: Annotated[int, Query(ge=1, le=100)] = 20,
    settings: Settings = Depends(get_settings),
) -> Pagination:
    if size > settings.max_page_size:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Page size cannot exceed {settings.max_page_size}",
        )

    return Pagination(page=page, size=size)

The same idea applies to services. Instead of constructing services inside the route, create them through dependencies and inject already-prepared objects into the endpoint.

class ProductService:
    def __init__(self, settings: Settings):
        self.settings = settings

    async def list_products(self, pagination: Pagination) -> dict:
        return {
            "source": "new-search"
            if self.settings.enable_new_search
            else "classic-search",
            "limit": pagination.limit,
            "offset": pagination.offset,
            "items": [
                {"id": 1, "name": "Keyboard"},
                {"id": 2, "name": "Mouse"},
            ],
        }


def get_product_service(
    settings: Settings = Depends(get_settings),
) -> ProductService:
    return ProductService(settings=settings)


@app.get("/v1/products")
async def list_products_v1(
    tenant_id: str = Depends(validate_tenant_id),
    pagination: Pagination = Depends(get_pagination),
    flags: FeatureFlags = Depends(get_feature_flags),
    service: ProductService = Depends(get_product_service),
):
    result = await service.list_products(pagination)

    return {
        "tenant_id": tenant_id,
        "features": flags,
        "data": result,
    }

Here the route does almost no low-level work. It receives an already-validated tenant_id, a ready-to-use Pagination object, typed feature flags, and a configured service. The route becomes orchestration, dependencies prepare the request context, and services handle business logic. This also makes testing easier because FastAPI allows dependencies to be replaced with fakes through app.dependency_overrides.

class FakeProductService:
    async def list_products(self, pagination: Pagination) -> dict:
        return {
            "source": "fake",
            "limit": pagination.limit,
            "offset": pagination.offset,
            "items": [],
        }


def get_fake_product_service() -> FakeProductService:
    return FakeProductService()


app.dependency_overrides[get_product_service] = get_fake_product_service

So, don’t forget to use DI for every reusable component you need to create and use in routes.

18. Set Explicit Naming Conventions for Database Constraints

By default, SQLAlchemy generally does not assign names to primary-key, foreign-key, unique, or check constraints. If you do not name them, the database generates backend-specific names. Except index created with index=True, for which SQLAlchemy normally generates an ix_... name. Inconsistent or auto-generated names become a maintenance nightmare during migrations and debugging. SQLAlchemy’s MetaData object accepts a naming_convention dictionary that automatically names all constraints in a consistent, readable way. Define this once on your base class:

from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase

convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(column_0_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=convention)

SQLAlchemy applies these conventions to unnamed constraints, including constraints implicitly created through options such as primary_key=True, unique=True, and index=True. Now every constraint gets a predictable name automatically:

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)  # uq_users_email
    age: Mapped[int] = mapped_column(CheckConstraint("age >= 18"))  # ck_users_age

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)  # pk_posts
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))  # fk_posts_user_id_users

• Consistent Naming and no more guessing: When you see fk_orders_customer_id_customers in an error message, you immediately know which table, column, and referenced table are involved
• Smoother migrations: Without your own convention, constraint names may be generated differently by PostgreSQL, MySQL, SQL Server, Oracle, or SQLite. A naming convention makes migrations deterministic and portable. Alembic uses the naming convention from the target_metadata configured in env.py:

 from myapp.models import Base
 target_metadata = Base.metadata

We can use predictable constraint names for operations such as:

op.drop_constraint(
     "fk_orders_user_id_users",
     "orders",
     type_="foreignkey",
 )

• Easier debugging: Constraint violation errors become self-explanatory without cross-referencing your schema

With this pattern, the database speaks in a language you understand. It’s a small setup cost that pays off every single time you run a migration or debug a constraint violation. Read this link to learn more.

---

These practices are not rigid laws. They are defaults that tend to keep FastAPI + SQLAlchemy codebases more readable, more predictable, less fragile under load and as Robert C. Martin said “to get it right”.

The challenge is choosing patterns that remain understandable when the project grows. If a pattern makes your model layer harder to reason about, creates duplicate meanings in the data, or hides transaction behavior, it is usually worth stepping back and simplifying.

Happy coding!

References

1. SQLAlchemy 2.0 documentation: https://docs.sqlalchemy.org/20/
2. SQLAlchemy declarative mapping: https://docs.sqlalchemy.org/20/orm/declarative_mapping.html
3. SQLAlchemy session basics: https://docs.sqlalchemy.org/20/orm/session_basics.html
4. SQLAlchemy relationship loading: https://docs.sqlalchemy.org/20/orm/queryguide/relationships.html
5. SQLAlchemy with_for_update() docs: https://docs.sqlalchemy.org/20/core/selectable.html
6. SQLAlchemy async ORM docs: https://docs.sqlalchemy.org/20/orm/extensions/asyncio.html
7. FastAPI dependencies with yield: https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/
8. Django model best practices: https://mshaeri.com/blog/django-best-practices-part-1/
9. Robert C. Martin, Clean Architecture: A Craftsman’s Guide to Software Structure and Design.
10. https://github.com/zhanymkanov/fastapi-best-practices#project-structure
11. The Importance of Naming Constraints in SQLAlchemy