FastAPI Project Structure That Scales Past the Tutorial Stage

TL;DR: Organize by feature, not by file type. Keep shared plumbing in a core package that knows nothing about features, put business logic in a plain service.py, build the app inside a create_app() function, and add one small test that fails when somebody breaks those rules. Everything below is a complete layout you can copy.

Every FastAPI project I start begins as a single main.py. Fifteen lines, one route, and /docs is working before my coffee cools down. That's the best part of the framework and I'm not knocking it.

The trouble shows up around the third or fourth feature. main.py is 600 lines. Models and Pydantic schemas live in the same file. Somebody added an import and now you have a circular import. And checking that a missing user returns a 404 somehow needs a real database running on your laptop.

This post is the structure I reach for when a project is going to live longer than a weekend. We'll build a small task tracker called Taskboard (users own tasks) and I'll show every file that matters, why it's shaped the way it is, how to test it, and how to stop the structure from rotting six months from now.

A quick note on versions: this is written for FastAPI with Pydantic v2 and SQLAlchemy 2.0's async API, on Python 3.12 or newer. FastAPI moves fast, so if anything here disagrees with the version you have installed, the official docs win.

What we'll cover

1. Why the tutorial layout stops working (and what to do instead)

2. The folder layout and the one rule that keeps it healthy

3. Settings, database and sessions

4. Error handling that doesn't leak HTTP into your business logic

5. A full feature, file by file

6. A second feature, and how features talk to each other

7. Wiring it together, plus migrations

8. Testing: fast API tests, one real-wiring test, and a test for the architecture itself

9. Gotchas, a "where does this go?" cheat sheet, and when to bend the rules

Why the tutorial layout stops working

FastAPI's own "Bigger Applications" guide organizes code by kind of file: an app/ package with main.py, a dependencies.py, and a routers/ folder holding users.py and items.py. It's a great starting point, and for a small app or a microservice with a couple of concerns it can stay that way forever.

The problem is what happens when you copy that idea across the whole project. You end up with folders like routers/, models/, schemas/ and crud/, and a single feature gets smeared across all four. To change how users work, you open four folders. To delete a feature, you go hunting.

The alternative is to group by feature (people also call it by domain or by module). Everything about users lives in users/. Everything about tasks lives in tasks/. A well-known write-up of this approach is the fastapi-best-practices repo, which says its structure is inspired by Netflix's Dispatch project.

That last cell matters. Grouping by feature isn't free. Its weak spot is features reaching into each other's internals until you have a ball of mud with nicer folder names. So we'll add rules, and then, because rules nobody checks are just wishes, a test that enforces them.

The layout

Here's the whole thing. Each package also has an empty __init__.py.

taskboard/
├── pyproject.toml
├── .env.example
├── migrations/                 # created by Alembic (step 7)
├── app/
│   ├── main.py                 # create_app() and the lifespan
│   ├── api_v1.py               # collects every feature router
│   ├── health.py               # GET /health
│   ├── registry.py             # imports every model (Alembic and tests need it)
│   ├── core/                   # shared plumbing, knows nothing about features
│   │   ├── config.py
│   │   ├── database.py
│   │   ├── dependencies.py
│   │   ├── exceptions.py
│   │   ├── error_handlers.py
│   │   └── clock.py
│   ├── users/                  # one folder per feature
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   ├── service.py
│   │   ├── dependencies.py
│   │   └── exceptions.py
│   └── tasks/                  # same shape as users
└── tests/
    ├── conftest.py
    ├── test_architecture.py
    ├── test_health.py
    ├── test_wiring.py
    ├── users/test_users_api.py
    └── tasks/test_tasks_api.py

Every feature has the same six files, so you never have to guess where something lives:

And here's the one idea that keeps it all healthy: dependencies only point one way.

HTTP request
                   │
                   ▼
   router.py ─────────────► schemas.py
        │
        ▼
   service.py ────────────► models.py
        │
        ▼
   another feature's service.py      ← the only door between features


   Every feature may import from core/.
   core/ never imports from any feature.

Two things fall out of that picture. core/ is safe to depend on because it can't depend back on you. And if tasks needs something from users, it goes through users/service.py, never through users/models.py or users/router.py. We'll turn those into automated checks in step 8.

Step 1: Settings

I want typed config, read from environment variables once, with no globals scattered around.

app/core/config.py

from functools import lru_cache
from typing import Literal

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_", extra="ignore")

    app_name: str = "Taskboard API"
    environment: Literal["local", "test", "production"] = "local"
    debug: bool = False
    database_url: str = "sqlite+aiosqlite:///./taskboard.db"
    db_echo: bool = False


@lru_cache
def get_settings() -> Settings:
    return Settings()

A few small decisions in there:

• env_prefix="APP_" means the variable for database_url is APP_DATABASE_URL. A prefix keeps you from colliding with whatever else is in the environment.

• extra="ignore" stops a stray variable in your .env from crashing startup.

• get_settings() is wrapped in lru_cache so the environment is parsed once. Tests don't use it though. They build their own Settings(...) and pass it in, which is why create_app (step 6) takes settings as an argument.

Commit a .env.example with the variable names and keep the real .env out of git:

.env.example

APP_ENVIRONMENT=local
APP_DATABASE_URL=sqlite+aiosqlite:///./taskboard.db
# APP_DATABASE_URL=postgresql+asyncpg://taskboard:secret@localhost:5432/taskboard
APP_DB_ECHO=false

Step 2: Database and sessions

Three things live in database.py: the declarative Base for your models, a function that builds an engine, and a function that builds a session factory.

app/core/database.py

from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    """Parent class of every ORM model in the project."""


def build_engine(url: str, *, echo: bool = False) -> AsyncEngine:
    return create_async_engine(url, echo=echo, pool_pre_ping=True)


def build_session_factory(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(engine, expire_on_commit=False)

Notice there's no module-level engine = create_async_engine(...). Importing a module shouldn't open connections, and it should be easy to build a different engine in a test or a script. The engine gets created in the app's lifespan instead (step 6).

Two settings are worth a sentence each:

• expire_on_commit=False: by default SQLAlchemy expires every loaded attribute after a commit, so the next access quietly reloads from the database. In async code an attribute access can't quietly do I/O, so SQLAlchemy's docs recommend turning this off.

• pool_pre_ping=True: checks that a pooled connection is still alive before handing it to you. It saves you from odd errors after a database restart or a dropped idle connection.

Now the dependency that hands a session to each request:

app/core/dependencies.py

from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession


async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    session_factory = request.app.state.session_factory
    async with session_factory() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session)]

SessionDep is an Annotated alias, so any route or dependency can just write session: SessionDep. Two details:

• The session factory is read from request.app.state, which the lifespan fills in. That's what lets tests swap the whole thing out with dependency_overrides without touching a real engine.

• The async with means the session is closed when the request finishes, even if an error was raised.

Gotcha: one session per request is something you get because FastAPI caches dependency results within a request. If a route uses SessionDep directly and also depends on something else that uses SessionDep, they get the same session. In step 5 we rely on that: a dependency loads a task, and then the route updates and commits it using that very same session. If you ever set use_cache=False on a dependency like this, that guarantee goes away.

Step 3: Errors that don't leak HTTP

Here's a rule I stick to: services never raise HTTPException. A service function might be called from a route today, a CLI command tomorrow, and a background job next month. None of those know what a 404 is.

So services raise plain domain errors, and one handler turns them into HTTP responses. The base classes live in core:

app/core/exceptions.py

from http import HTTPStatus


class AppError(Exception):
    """Base class for every error we raise on purpose."""

    status_code: int = HTTPStatus.INTERNAL_SERVER_ERROR.value
    code: str = "internal_error"

    def __init__(self, message: str | None = None) -> None:
        self.message = message or type(self).__name__
        super().__init__(self.message)


class NotFoundError(AppError):
    status_code = HTTPStatus.NOT_FOUND.value
    code = "not_found"


class ConflictError(AppError):
    status_code = HTTPStatus.CONFLICT.value
    code = "conflict"

And the single place that knows how to turn them into JSON:

app/core/error_handlers.py

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

from app.core.exceptions import AppError


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": {"code": exc.code, "message": exc.message}},
        )

Every feature then defines its own specific errors by subclassing these (you'll see UserNotFoundError in a moment). The client gets a consistent shape:

{"error": {"code": "user_not_found", "message": "User 999 does not exist"}}

Note: FastAPI's own validation errors (the 422s) keep their default detail shape. You can override that too, but I'd leave it alone until a client actually needs one consistent format.

Step 4: A full feature, file by file

Let's build users. I'll go in the order I'd write the files.

Models. A tiny helper keeps timestamps timezone-aware and in one place:

app/core/clock.py

from datetime import UTC, datetime


def utcnow() -> datetime:
    return datetime.now(UTC)

app/users/models.py

from datetime import datetime

from sqlalchemy import DateTime, String
from sqlalchemy.orm import Mapped, mapped_column

from app.core.clock import utcnow
from app.core.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
    full_name: Mapped[str] = mapped_column(String(120))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

The created_at default is set in Python (default=utcnow) rather than in the database. The email column gets a unique index, and that index is what makes duplicate detection reliable later.

Schemas. Separate models for what comes in and what goes out:

app/users/schemas.py

from datetime import datetime

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserCreate(BaseModel):
    email: EmailStr
    full_name: str = Field(min_length=1, max_length=120)


class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    full_name: str
    created_at: datetime

UserCreate validates input (EmailStr needs the email-validator package, which fastapi[standard] includes). UserRead has from_attributes=True so it can be built straight from an ORM object. Keeping these two apart means you can never accidentally expose a column just because it exists on the table.

Exceptions.

app/users/exceptions.py

from app.core.exceptions import ConflictError, NotFoundError


class UserNotFoundError(NotFoundError):
    code = "user_not_found"

    def __init__(self, user_id: int) -> None:
        super().__init__(f"User {user_id} does not exist")


class EmailAlreadyRegisteredError(ConflictError):
    code = "email_already_registered"

    def __init__(self, email: str) -> None:
        super().__init__(f"Email {email} is already registered")

Service. This is where the actual logic lives:

app/users/service.py

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.users.exceptions import EmailAlreadyRegisteredError, UserNotFoundError
from app.users.models import User
from app.users.schemas import UserCreate


async def create_user(session: AsyncSession, data: UserCreate) -> User:
    email = data.email.lower()
    user = User(email=email, full_name=data.full_name)
    session.add(user)
    try:
        await session.commit()
    except IntegrityError:
        await session.rollback()
        raise EmailAlreadyRegisteredError(email) from None
    return user


async def get_user(session: AsyncSession, user_id: int) -> User:
    user = await session.get(User, user_id)
    if user is None:
        raise UserNotFoundError(user_id)
    return user


async def list_users(session: AsyncSession, *, limit: int, offset: int) -> list[User]:
    result = await session.scalars(select(User).order_by(User.id).limit(limit).offset(offset))
    return list(result.all())


async def ensure_user_exists(session: AsyncSession, user_id: int) -> None:
    found = await session.scalar(select(User.id).where(User.id == user_id))
    if found is None:
        raise UserNotFoundError(user_id)

Look at create_user. I don't check "does this email exist?" and then insert. Between the check and the insert, another request could sneak the same email in. Instead I insert and let the database's unique constraint be the referee, then translate the IntegrityError into a domain error. It's one fewer query and it's race-safe. I also lowercase the email so [email protected] and [email protected] count as the same person.

There's also ensure_user_exists, which exists purely so other features have a clean, narrow way to ask "is this a real user?" without ever touching the User model.

Dependencies. A reusable "load it or 404" helper:

app/users/dependencies.py

from typing import Annotated

from fastapi import Depends

from app.core.dependencies import SessionDep
from app.users import service
from app.users.models import User


async def get_user_or_404(user_id: int, session: SessionDep) -> User:
    return await service.get_user(session, user_id)


UserDep = Annotated[User, Depends(get_user_or_404)]

Router. And finally the part that's almost boring, which is the point:

app/users/router.py

from typing import Annotated

from fastapi import APIRouter, Query, status

from app.core.dependencies import SessionDep
from app.users import service
from app.users.dependencies import UserDep
from app.users.models import User
from app.users.schemas import UserCreate, UserRead

router = APIRouter(prefix="/users", tags=["users"])


@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(data: UserCreate, session: SessionDep) -> User:
    return await service.create_user(session, data)


@router.get("", response_model=list[UserRead])
async def list_users(
    session: SessionDep,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[User]:
    return await service.list_users(session, limit=limit, offset=offset)


@router.get("/{user_id}", response_model=UserRead)
async def get_user(user: UserDep) -> User:
    return user

Each route is a couple of lines because the thinking happens in the service. One thing to notice: the routes say response_model=UserRead but their return annotation is User. The function genuinely returns an ORM object, so I annotate it honestly (type checkers are happy), and FastAPI uses the response model to turn it into the public shape.

Step 5: A second feature, and how features talk

Tasks belong to users. The interesting part is tasks/service.py, because it needs to know a user exists before creating a task:

app/tasks/service.py

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

from app.tasks.exceptions import TaskNotFoundError
from app.tasks.models import Task
from app.tasks.schemas import TaskCreate, TaskUpdate
from app.users import service as users_service


async def create_task(session: AsyncSession, data: TaskCreate) -> Task:
    await users_service.ensure_user_exists(session, data.owner_id)
    task = Task(title=data.title, owner_id=data.owner_id)
    session.add(task)
    await session.commit()
    return task


async def get_task(session: AsyncSession, task_id: int) -> Task:
    task = await session.get(Task, task_id)
    if task is None:
        raise TaskNotFoundError(task_id)
    return task


async def list_tasks(
    session: AsyncSession,
    *,
    owner_id: int | None,
    done: bool | None,
    limit: int,
    offset: int,
) -> list[Task]:
    stmt = select(Task)
    if owner_id is not None:
        stmt = stmt.where(Task.owner_id == owner_id)
    if done is not None:
        stmt = stmt.where(Task.done == done)
    result = await session.scalars(stmt.order_by(Task.id).limit(limit).offset(offset))
    return list(result.all())


async def update_task(session: AsyncSession, task: Task, data: TaskUpdate) -> Task:
    changes = data.model_dump(exclude_unset=True, exclude_none=True)
    for field, value in changes.items():
        setattr(task, field, value)
    await session.commit()
    return task

Check the import at the top: from app.users import service as users_service. That's the only door. Tasks never imports users.models and never reaches into a user's table. If the way users are stored changes tomorrow, tasks doesn't notice.

The update logic has a subtle choice in it. model_dump(exclude_unset=True, exclude_none=True) means a PATCH body only changes the fields the client actually sent, and sending null for a field means "leave it alone" rather than "set it to NULL". Without exclude_none, {"title": null} would try to write NULL into a NOT NULL column and blow up with a 500. Here are the schemas that go with it:

app/tasks/schemas.py

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    owner_id: int


class TaskUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    done: bool | None = None


class TaskRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    done: bool
    owner_id: int
    created_at: datetime

And the router, including the PATCH route that uses a loaded task and the session together (this is the dependency-caching guarantee from step 2 at work):

app/tasks/router.py

from typing import Annotated

from fastapi import APIRouter, Query, status

from app.core.dependencies import SessionDep
from app.tasks import service
from app.tasks.dependencies import TaskDep
from app.tasks.models import Task
from app.tasks.schemas import TaskCreate, TaskRead, TaskUpdate

router = APIRouter(prefix="/tasks", tags=["tasks"])


@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
async def create_task(data: TaskCreate, session: SessionDep) -> Task:
    return await service.create_task(session, data)


@router.get("", response_model=list[TaskRead])
async def list_tasks(
    session: SessionDep,
    owner_id: int | None = None,
    done: bool | None = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[Task]:
    return await service.list_tasks(
        session, owner_id=owner_id, done=done, limit=limit, offset=offset
    )


@router.get("/{task_id}", response_model=TaskRead)
async def get_task(task: TaskDep) -> Task:
    return task


@router.patch("/{task_id}", response_model=TaskRead)
async def update_task(task: TaskDep, data: TaskUpdate, session: SessionDep) -> Task:
    return await service.update_task(session, task, data)

The task's models.py, exceptions.py and dependencies.py mirror the users versions, so I won't repeat them. The only difference worth mentioning is that Task.owner_id is a ForeignKey("users.id"), written as a string so the tasks package never has to import the users model.

Step 6: Wiring it together

Three small files tie everything up. First, one place that collects every feature router:

app/api_v1.py

from fastapi import APIRouter

from app.tasks.router import router as tasks_router
from app.users.router import router as users_router

router = APIRouter()
router.include_router(users_router)
router.include_router(tasks_router)

The /api/v1 prefix gets added in main.py, so shipping a v2 later means adding a sibling file rather than rewriting this one.

Second, the model registry:

app/registry.py

"""Import every ORM model so Base.metadata knows about all tables.

Alembic's env.py and the test fixtures import this module before they touch the metadata.
"""

from app.tasks.models import Task
from app.users.models import User

__all__ = ["Task", "User"]

This looks pointless until the day Alembic autogenerates an empty migration, or your tests create zero tables. SQLAlchemy only knows about models that have actually been imported, so something needs to import all of them. This file is that something.

And then main.py, which builds the app:

app/main.py

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api_v1 import router as api_v1_router
from app.core.config import Settings, get_settings
from app.core.database import build_engine, build_session_factory
from app.core.error_handlers import register_exception_handlers
from app.health import router as health_router


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    settings: Settings = app.state.settings
    engine = build_engine(settings.database_url, echo=settings.db_echo)
    app.state.session_factory = build_session_factory(engine)
    try:
        yield
    finally:
        await engine.dispose()


def create_app(settings: Settings | None = None) -> FastAPI:
    settings = settings or get_settings()
    app = FastAPI(title=settings.app_name, debug=settings.debug, lifespan=lifespan)
    app.state.settings = settings
    register_exception_handlers(app)
    app.include_router(health_router)
    app.include_router(api_v1_router, prefix="/api/v1")
    return app


app = create_app()

Why a create_app() function instead of a bare app = FastAPI() at module level? Because tests can call it with their own settings and get a fresh app each time. And why create the engine in lifespan and not at import time? Because importing app.main (which every test does) shouldn't try to connect to anything. The lifespan creates the engine on startup and disposes of it on shutdown.

The module-level app = create_app() at the bottom is just what the server imports.

There's also a trivial health endpoint, so a load balancer has something to poll:

app/health.py

from fastapi import APIRouter

router = APIRouter(tags=["health"])


@router.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}

Step 7: Migrations

For schema changes, use Alembic. The async template gets you going:

uv run alembic init -t async migrations

Then make two edits in the generated migrations/env.py. Right after the line config = context.config, add the database URL from your settings, and replace the target_metadata = None line:

from app import registry  # noqa: F401
from app.core.config import get_settings
from app.core.database import Base

config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata

After that the day-to-day loop is:

uv run alembic revision --autogenerate -m "create users and tasks"
uv run alembic upgrade head

Always read the generated migration before you apply it. Autogenerate is a good first draft, not an oracle.

Step 8: Tests

This is the part that makes the structure pay off. Because the service layer has no idea what HTTP is, and the session arrives through a dependency, testing is mostly a matter of swapping one function.

Here's my pyproject.toml, since the test and tooling config matters as much as the dependencies:

pyproject.toml

[project]
name = "taskboard"
version = "0.1.0"
description = "Example FastAPI project structure that scales past the tutorial stage"
requires-python = ">=3.12"
dependencies = [
    "fastapi[standard]>=0.115",
    "pydantic-settings>=2.4",
    "sqlalchemy[asyncio]>=2.0.30",
    "aiosqlite>=0.20",
    "alembic>=1.13",
]

[dependency-groups]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.24",
    "httpx>=0.27",
    "ruff>=0.6",
]

[tool.fastapi]
entrypoint = "app.main:app"

[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "ASYNC"]

A few things to point at:

• [tool.fastapi] entrypoint tells fastapi dev and fastapi run where your app lives. FastAPI's docs recommend setting it here, because other tools (like the VS Code extension and FastAPI Cloud) can't find the app otherwise.

• pythonpath = ["."] lets tests do from app... with the project root as the import root.

• asyncio_mode = "auto" means every async def test_... just runs, with no marker on each one. Recent versions of pytest-asyncio also ask you to set the fixture loop scope explicitly, hence the second line.

Aside: FastAPI's own async-testing docs use anyio's pytest.mark.anyio. That works fine too. I like pytest-asyncio's auto mode because async fixtures need no ceremony. Pick one and stay consistent.

The fixtures

tests/conftest.py

from collections.abc import AsyncIterator, Awaitable, Callable

import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine

from app import registry  # noqa: F401  (makes sure every model is on Base.metadata)
from app.core.config import Settings
from app.core.database import Base, build_session_factory
from app.core.dependencies import get_session
from app.main import create_app


@pytest.fixture
async def engine(tmp_path) -> AsyncIterator[AsyncEngine]:
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest.fixture
async def api_app(engine: AsyncEngine) -> AsyncIterator[FastAPI]:
    session_factory = build_session_factory(engine)

    async def override_get_session() -> AsyncIterator[AsyncSession]:
        async with session_factory() as session:
            yield session

    application = create_app(Settings(environment="test"))
    application.dependency_overrides[get_session] = override_get_session
    yield application
    application.dependency_overrides.clear()


@pytest.fixture
async def client(api_app: FastAPI) -> AsyncIterator[AsyncClient]:
    transport = ASGITransport(app=api_app)
    async with AsyncClient(transport=transport, base_url="http://test") as http_client:
        yield http_client


@pytest.fixture
def create_user(client: AsyncClient) -> Callable[..., Awaitable[dict]]:
    async def _create_user(
        email: str = "[email protected]", full_name: str = "Ada Lovelace"
    ) -> dict:
        response = await client.post("/api/v1/users", json={"email": email, "full_name": full_name})
        assert response.status_code == 201, response.text
        return response.json()

    return _create_user


@pytest.fixture
def create_task(client: AsyncClient) -> Callable[..., Awaitable[dict]]:
    async def _create_task(owner_id: int, title: str = "Write the post") -> dict:
        response = await client.post("/api/v1/tasks", json={"title": title, "owner_id": owner_id})
        assert response.status_code == 201, response.text
        return response.json()

    return _create_task

What's going on here:

• A fresh database per test. Each test gets its own SQLite file in pytest's tmp_path, with tables created from the models' metadata. No state leaks between tests, and no cleanup code.

• One override. dependency_overrides[get_session] points the app at that test database. Nothing else in the app is mocked, so tests exercise the real routers, services, schemas and error handlers.

• AsyncClient with ASGITransport. The app is called in-process. There's no server and no port.

• create_user and create_task fixtures. Tests that need a user shouldn't repeat the setup request ten times.

Gotcha: ASGITransport does not run your lifespan. FastAPI's docs call this out and point at the asgi-lifespan package if you need it. Because we override the session dependency, we don't need the lifespan for these tests. But that also means the real get_session and the real lifespan are never exercised, so I add exactly one test for them below.

What the API tests look like

They read like the sentence you'd say out loud. Here's the duplicate-email rule, including the case-insensitive part:

tests/users/test_users_api.py

async def test_duplicate_email_is_a_409_even_with_different_casing(
    client: AsyncClient, create_user
) -> None:
    await create_user(email="[email protected]")

    response = await client.post(
        "/api/v1/users", json={"email": "[email protected]", "full_name": "Someone Else"}
    )

    assert response.status_code == 409
    assert response.json()["error"]["code"] == "email_already_registered"

The error shape from step 3, checked exactly:

tests/users/test_users_api.py

async def test_get_unknown_user_is_a_404_with_our_error_shape(client: AsyncClient) -> None:
    response = await client.get("/api/v1/users/999")

    assert response.status_code == 404
    assert response.json() == {
        "error": {"code": "user_not_found", "message": "User 999 does not exist"}
    }

The null PATCH behavior from step 5:

tests/tasks/test_tasks_api.py

async def test_patch_with_null_title_changes_nothing(
    client: AsyncClient, create_user, create_task
) -> None:
    user = await create_user()
    task = await create_task(owner_id=user["id"], title="Still here")

    response = await client.patch(f"/api/v1/tasks/{task['id']}", json={"title": None})

    assert response.status_code == 200
    assert response.json()["title"] == "Still here"

And filtering, which builds up some data and then asks three different questions:

tests/tasks/test_tasks_api.py

async def test_list_tasks_filters_by_owner_and_done(
    client: AsyncClient, create_user, create_task
) -> None:
    ada = await create_user(email="[email protected]")
    grace = await create_user(email="[email protected]", full_name="Grace Hopper")
    ada_task = await create_task(owner_id=ada["id"], title="Ada 1")
    await create_task(owner_id=ada["id"], title="Ada 2")
    await create_task(owner_id=grace["id"], title="Grace 1")
    await client.patch(f"/api/v1/tasks/{ada_task['id']}", json={"done": True})

    ada_only = await client.get("/api/v1/tasks", params={"owner_id": ada["id"]})
    ada_done = await client.get("/api/v1/tasks", params={"owner_id": ada["id"], "done": True})
    everything = await client.get("/api/v1/tasks")

    assert [t["title"] for t in ada_only.json()] == ["Ada 1", "Ada 2"]
    assert [t["title"] for t in ada_done.json()] == ["Ada 1"]
    assert len(everything.json()) == 3

The full project has 24 tests covering creation, validation (422s), not-found and conflict errors, pagination limits, filtering, and PATCH edge cases. I'm only showing the interesting ones here; the rest follow the same pattern.

One test with no overrides

Overriding get_session is great for speed, but it means the real thing never runs. So there's one test that doesn't override anything. It creates the tables with plain synchronous SQLAlchemy, then uses Starlette's TestClient as a context manager. Used that way, it runs the real lifespan, which builds the real engine, which feeds the real get_session:

tests/test_wiring.py

"""One test that uses the real lifespan and the real get_session (no overrides)."""

from fastapi.testclient import TestClient
from sqlalchemy import create_engine

from app import registry  # noqa: F401
from app.core.config import Settings
from app.core.database import Base
from app.main import create_app


def test_lifespan_and_real_session_dependency(tmp_path) -> None:
    db_file = tmp_path / "wiring.db"
    sync_engine = create_engine(f"sqlite:///{db_file}")
    Base.metadata.create_all(sync_engine)
    sync_engine.dispose()

    application = create_app(Settings(database_url=f"sqlite+aiosqlite:///{db_file}"))
    assert not hasattr(application.state, "session_factory")

    with TestClient(application) as client:
        assert application.state.session_factory is not None

        created = client.post(
            "/api/v1/users", json={"email": "[email protected]", "full_name": "Ada Lovelace"}
        )
        assert created.status_code == 201

        fetched = client.get(f"/api/v1/users/{created.json()['id']}")
        assert fetched.status_code == 200
        assert fetched.json()["email"] == "[email protected]"

Gotcha: SQLite is convenient for tests, but it isn't Postgres. It doesn't enforce foreign keys unless you ask it to, and it stores datetimes without timezone information, so values come back naive. We're covered here because the service checks the owner exists itself. If production runs Postgres, also run this suite against Postgres in CI (a service container or testcontainers both work). It's the cheapest way to catch "works on SQLite" surprises.

Step 9: A test for the architecture itself

Remember the rules from the layout section? Here they are as code. The test parses every file in app/ with Python's ast module and checks the imports. It never imports your app, so it's fast and needs nothing installed beyond pytest.

The rules:

tests/test_architecture.py

"""Executable version of the rules in the project's README.

Pure standard library: it parses the source with `ast` and never imports the app.
"""

import ast
from pathlib import Path

APP_DIR = Path(__file__).resolve().parent.parent / "app"

# What one feature may import from another feature.
PUBLIC_MODULES = {"service", "schemas", "exceptions"}


def _features(app_dir: Path) -> set[str]:
    return {p.name for p in app_dir.iterdir() if p.is_dir() and (p / "router.py").exists()}


def _imports(path: Path, package: str) -> set[str]:
    """Every dotted module name this file imports from our own package."""
    found: set[str] = set()
    for node in ast.walk(ast.parse(path.read_text())):
        if isinstance(node, ast.Import):
            found.update(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            if node.level:
                found.add(f"{package}.<relative import>")
            elif node.module:
                found.add(node.module)
                found.update(f"{node.module}.{alias.name}" for alias in node.names)
    return {m for m in found if m == package or m.startswith(f"{package}.")}


def _locate(module: str, features: set[str]) -> tuple[str | None, str | None]:
    """'app.users.service.create_user' -> ('users', 'service'); non-feature -> (None, None)."""
    parts = module.split(".")
    if len(parts) > 1 and parts[1] in features:
        return parts[1], parts[2] if len(parts) > 2 else None
    return None, None


def find_violations(app_dir: Path) -> list[str]:
    package = app_dir.name
    features = _features(app_dir)
    problems: list[str] = []
    graph: dict[str, set[str]] = {feature: set() for feature in features}

    for path in sorted(app_dir.rglob("*.py")):
        rel = path.relative_to(app_dir).as_posix()
        owner = rel.split("/")[0] if "/" in rel else None
        for module in sorted(_imports(path, package)):
            if module.endswith("<relative import>"):
                problems.append(f"{rel}: use absolute imports")
                continue
            target, sub = _locate(module, features)
            if target is None:
                continue
            where = f"{package}.{target}" + (f".{sub}" if sub else "")
            crosses_features = owner in features and target != owner
            if crosses_features:
                graph[owner].add(target)
            if owner == "core":
                problems.append(f"{rel}: core must not import feature code ({target})")
            if sub == "router" and rel != "api_v1.py" and owner != target:
                problems.append(f"{rel}: only api_v1.py may import a router ({where})")
            elif crosses_features and sub is not None and sub not in PUBLIC_MODULES:
                problems.append(
                    f"{rel}: {owner} may only import service, schemas or exceptions"
                    f" from {target} ({where})"
                )

    def reachable(start: str) -> set[str]:
        seen: set[str] = set()
        stack = [start]
        while stack:
            for nxt in graph[stack.pop()]:
                if nxt not in seen:
                    seen.add(nxt)
                    stack.append(nxt)
        return seen

    cyclic = sorted(feature for feature in features if feature in reachable(feature))
    if cyclic:
        problems.append(f"features import each other in a cycle: {', '.join(cyclic)}")
    return list(dict.fromkeys(problems))


def test_the_real_project_follows_the_layering_rules() -> None:
    assert find_violations(APP_DIR) == []


def _write(root: Path, rel: str, source: str = "") -> None:
    path = root / rel
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(source)


def test_the_checker_really_catches_violations(tmp_path: Path) -> None:
    app = tmp_path / "app"
    for feature in ("users", "tasks"):
        for name in ("router", "service", "models"):
            _write(app, f"{feature}/{name}.py")
    _write(app, "core/leak.py", "from app.users.service import create_user\n")
    _write(app, "tasks/peeks.py", "from app.users.models import User\n")
    _write(app, "tasks/cycle.py", "from app.users import service\n")
    _write(app, "users/cycle_back.py", "from app.tasks.service import create_task\n")
    _write(app, "tasks/sneaky.py", "from app.users.router import router\n")
    _write(app, "users/relative.py", "from . import service\n")

    problems = "\n".join(find_violations(app))

    assert "core/leak.py: core must not import feature code" in problems
    assert "tasks/peeks.py" in problems and "users.models" in problems
    assert "tasks/sneaky.py: only api_v1.py may import a router" in problems
    assert "users/relative.py: use absolute imports" in problems
    assert "features import each other in a cycle: tasks, users" in problems


def test_the_checker_accepts_a_clean_tree(tmp_path: Path) -> None:
    app = tmp_path / "app"
    for feature in ("users", "tasks"):
        for name in ("router", "service", "models"):
            _write(app, f"{feature}/{name}.py")
    _write(app, "tasks/service.py", "from app.users import service as users_service\n")
    _write(app, "api_v1.py", "from app.users.router import router\n")
    _write(app, "registry.py", "from app.users.models import User\n")

    assert find_violations(app) == []

The second and third tests are my favorite part. A checker that always says "all good" is useless, so one test builds a deliberately broken project in a temp folder (a leaky core, a feature peeking at another's models, a cycle, a sneaky router import, a relative import) and asserts that every violation is reported. Another confirms a clean tree produces nothing. You're testing the tester.

When somebody on your team adds from app.users.models import User inside tasks at 5pm on a Friday, CI fails with a message that names the file and the rule.

Running it

uv sync                      # install everything, dev tools included
uv run pytest                # run the tests
cp .env.example .env
uv run alembic upgrade head  # after the migration steps above
uv run fastapi dev           # development server, with auto-reload

Then open http://127.0.0.1:8000/docs. In production you'd use fastapi run instead of fastapi dev.

Gotchas worth knowing

1. Session sharing depends on dependency caching. Covered in step 2. It's the quiet assumption behind the "load, then update" pattern.

2. expire_on_commit=False is required for async. Without it you'll hit errors reading attributes after a commit.

3. Let the database enforce uniqueness. Check-then-insert has a race. Insert and catch IntegrityError.

4. Decide what null means in a PATCH. We chose "no change". Whatever you choose, test it.

5. Import every model somewhere. That's registry.py. Otherwise autogenerate and create_all silently do nothing.

6. The test client skips the lifespan. Cover the real wiring with one explicit test.

7. Tests on SQLite can hide Postgres problems. Run CI on your production database engine.

Where does this code go?

Adding a feature, as a checklist: create the folder with its __init__.py, write models.py and schemas.py, write service.py and its exceptions.py, write router.py, register the model in registry.py, include the router in api_v1.py, generate a migration, write tests. The architecture test will tell you if you slipped.

When I'd bend the rules

Structure should serve you, not the other way around. Here's where I'd deviate:

• Authentication. Almost every feature needs "the current user". I'd make auth its own feature and add dependencies to the list of modules other features may import (it's one line in the architecture test). Putting it in core/ doesn't work, because it needs the user model and core can't import features.

• A repository layer. Plain service functions with a session are enough for a long time. Once queries get complicated or you have more than one data store, a repository class between the service and the database starts to earn its keep.

• A service that outgrows its file. Turn service.py into a service/ package and re-export the public functions from __init__.py. Callers don't change.

• Tiny projects. If you really only have one or two routes, the single-file tutorial layout is fine. Don't build a cathedral for a shed.

Wrapping up

The whole trick fits in a sentence: group by feature, keep dependencies pointing one way, keep HTTP out of your business logic, and write down your rules as a test. Everything else in this post is just the code that makes that sentence real.

If you take only one thing, take the architecture test. Folder layouts decay quietly, one convenient import at a time, and a test that fails loudly is the cheapest way I know to stop that.