{"schemaVersion":"1.0","type":"TechArticle","types":["Article","TechArticle"],"slug":"fastapi-project-structure-that-scales-past-the-tutorial-stage-8c9g3","url":"https://zyvop.com/fastapi-project-structure-that-scales-past-the-tutorial-stage-8c9g3","title":"FastAPI Project Structure That Scales Past the Tutorial Stage","subtitle":"A practical FastAPI architecture for growing projects, with feature-based modules, clean service layers, async database access, testing, migrations, and enforced dependency boundaries.","tldr":"A practical guide to structuring FastAPI projects beyond the tutorial stage. Learn how to organize by feature, separate business logic from HTTP, handle async database access, test real wiring, and keep architecture from degrading over time.","keywords":["backend-development","Software Architecture","Python","FastAPI","Async Python","Tutorial"],"entities":["Ankit Singh","Software Engineer","backend-development","Software Architecture","Python","FastAPI","Async Python","Tutorial","ZyVOP"],"keyTakeaways":["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."],"headings":["Why the tutorial layout stops working","The layout","Step 1: Settings","Step 2: Database and sessions","Step 3: Errors that don't leak HTTP","Step 4: A full feature, file by file","Step 5: A second feature, and how features talk","Step 6: Wiring it together","Step 7: Migrations","Step 8: Tests","The fixtures","What the API tests look like","One test with no overrides","Step 9: A test for the architecture itself","Running it","Gotchas worth knowing","Where does this code go?","When I'd bend the rules","Wrapping up"],"outboundLinks":[],"contentText":"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 Why the tutorial layout stops working (and what to do instead) The folder layout and the one rule that keeps it healthy Settings, database and sessions Error handling that doesn't leak HTTP into your business logic A full feature, file by file A second feature, and how features talk to each other Wiring it together, plus migrations Testing: fast API tests, one real-wiring test, and a test for the architecture itself 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. Grouped by file type Grouped by feature Add a \"billing\" feature Touch 4 folders Add 1 folder Delete a feature Hunt through every folder Delete the folder Reviewing a change Diff is scattered Diff lives in one place Works best for Small apps, microservices with few scopes Apps with several distinct areas Weak spot Everything ends up importing everything Features can tangle together unless you set rules 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.pyEvery feature has the same six files, so you never have to guess where something lives: File What goes in it What must not go in it router.py URL paths, status codes, query params; calls one service function SQL, business rules schemas.py Pydantic models for request and response bodies Database code models.py SQLAlchemy tables Anything that knows about HTTP service.py Business rules and queries. Takes a session, returns models, raises domain errors HTTPException, Request, Response dependencies.py Depends(...) helpers for this feature Business logic exceptions.py Errors that mean something in this domain HTTP details 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=falseStep 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: datetimeUserCreate 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 ADA@example.com and ada@example.com 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 userEach 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 taskCheck 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: datetimeAnd 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 migrationsThen 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.metadataAfter that the day-to-day loop is: uv run alembic revision --autogenerate -m \"create users and tasks\" uv run alembic upgrade headAlways 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 = \"ada@example.com\", 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_taskWhat'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=\"ada@example.com\") response = await client.post( \"/api/v1/users\", json={\"email\": \"ADA@example.com\", \"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=\"ada@example.com\") grace = await create_user(email=\"grace@example.com\", 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()) == 3The 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\": \"ada@example.com\", \"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\"] == \"ada@example.com\"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: # Rule Why 1 core/ never imports a feature Keeps shared code safe to depend on 2 A feature may import another feature's service, schemas or exceptions, nothing else Features talk through narrow doors 3 Features never import each other in a cycle No circular imports 4 Only api_v1.py imports routers Routing stays in one place 5 Absolute imports only Makes the checks above reliable 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-reloadThen open http://127.0.0.1:8000/docs. In production you'd use fastapi run instead of fastapi dev. Gotchas worth knowing Session sharing depends on dependency caching. Covered in step 2. It's the quiet assumption behind the \"load, then update\" pattern. expire_on_commit=False is required for async. Without it you'll hit errors reading attributes after a commit. Let the database enforce uniqueness. Check-then-insert has a race. Insert and catch IntegrityError. Decide what null means in a PATCH. We chose \"no change\". Whatever you choose, test it. Import every model somewhere. That's registry.py. Otherwise autogenerate and create_all silently do nothing. The test client skips the lifespan. Cover the real wiring with one explicit test. Tests on SQLite can hide Postgres problems. Run CI on your production database engine. Where does this code go? I'm writing... It goes in... A request or response body <feature>/schemas.py A SQL query or a business rule <feature>/service.py A Depends(...) that loads something by id <feature>/dependencies.py An error that means something in the domain <feature>/exceptions.py Settings, DB engine, shared dependencies core/ A new table <feature>/models.py, an import in registry.py, and a migration A whole new feature A new folder with the same six files, plus one line in api_v1.py 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.","contentHash":"sha256:897e862e5e5245ecc667e00024bc8b9ad343a7050efb4f2b7c2808c1f8836d89","authorName":"Ankit Singh","authorUrl":"https://zyvop.com/author/ankit","authorSameAs":[],"category":"Tutorial","tags":["backend-development","Software Architecture","Python","FastAPI","Async Python"],"audience":"Senior software engineers, systems architects, and technical leads working with Tutorial","tone":"Instructional, practical, code-first","readingTimeMinutes":28,"wordCount":6072,"faqs":null,"primaryTopic":"Tutorial","publishedAt":"2026-10-01T09:19:37.600Z","updatedAt":"2026-10-01T09:19:37.600Z","canonicalUrl":"https://zyvop.com/fastapi-project-structure-that-scales-past-the-tutorial-stage-8c9g3"}