FastAPI Development in Practice: Clean Project Structure, Dependency Injection, and a Real “Service Layer”

FastAPI makes it easy to ship endpoints fast—but many projects turn into a single main.py file with tangled logic, inconsistent error handling, and copy-pasted database code. This guide shows a practical structure you can reuse on real teams: clear folders, dependency injection (DI) for database sessions and auth, and a simple “service layer” so your routes stay thin and testable.

We’ll build a tiny “Tasks” API:

Create a task
List tasks (with pagination)
Mark a task as done

The examples use SQLAlchemy 2.0 async with asyncpg (PostgreSQL). You can adapt to SQLite for local dev, but async Postgres is a common production setup.

1) Dependencies and project layout

Install dependencies:

pip install fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg pydantic-settings python-dotenv pip install pytest pytest-asyncio httpx

Recommended layout:

app/ main.py core/ config.py db.py tasks/ models.py schemas.py repository.py service.py router.py tests/ test_tasks.py .env

This separation keeps responsibilities clear:

router.py: request/response + HTTP concerns
service.py: business logic + decisions
repository.py: database access
core/*: shared infrastructure (settings, DB session)

2) Configuration with pydantic-settings

Create app/core/config.py:

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_ignore_empty=True) DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres" settings = Settings()

Create a .env file:

DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/postgres

3) Async database session dependency

Create app/core/db.py:

from typing import AsyncIterator from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from sqlalchemy.orm import DeclarativeBase from app.core.config import settings engine = create_async_engine(settings.DATABASE_URL, echo=False, pool_pre_ping=True) SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) class Base(DeclarativeBase): pass async def get_db() -> AsyncIterator[AsyncSession]: async with SessionLocal() as session: yield session

Key idea: get_db() is a dependency. FastAPI injects it per-request and closes it safely.

4) Model and schemas

Create app/tasks/models.py:

from sqlalchemy import Boolean, String from sqlalchemy.orm import Mapped, mapped_column from app.core.db import Base class Task(Base): __tablename__ = "tasks" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) title: Mapped[str] = mapped_column(String(200), nullable=False) done: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)

Create app/tasks/schemas.py:

from pydantic import BaseModel, Field class TaskCreate(BaseModel): title: str = Field(min_length=1, max_length=200) class TaskRead(BaseModel): id: int title: str done: bool model_config = {"from_attributes": True}

5) Repository: DB code lives here

Create app/tasks/repository.py:

from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.tasks.models import Task class TaskRepository: def __init__(self, db: AsyncSession): self.db = db async def create(self, title: str) -> Task: task = Task(title=title) self.db.add(task) await self.db.commit() await self.db.refresh(task) return task async def list(self, limit: int, offset: int) -> list[Task]: stmt = select(Task).order_by(Task.id.desc()).limit(limit).offset(offset) res = await self.db.execute(stmt) return list(res.scalars().all()) async def get(self, task_id: int) -> Task | None: return await self.db.get(Task, task_id) async def mark_done(self, task: Task) -> Task: task.done = True await self.db.commit() await self.db.refresh(task) return task

This is intentionally boring—good. Keeping DB calls centralized makes the rest of your code easier to change and test.

6) Service layer: business rules and errors

Create app/tasks/service.py:

from fastapi import HTTPException, status from app.tasks.repository import TaskRepository from app.tasks.models import Task class TaskService: def __init__(self, repo: TaskRepository): self.repo = repo async def create_task(self, title: str) -> Task: # Business rule example: trim and reject empty titles title = title.strip() if not title: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Title cannot be empty.", ) return await self.repo.create(title=title) async def list_tasks(self, limit: int, offset: int) -> list[Task]: limit = min(max(limit, 1), 100) # clamp 1..100 offset = max(offset, 0) return await self.repo.list(limit=limit, offset=offset) async def complete_task(self, task_id: int) -> Task: task = await self.repo.get(task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") if task.done: return task return await self.repo.mark_done(task)

Notice what we gained:

Routes stay thin (no database code in endpoints)
Rules (clamping, validation, 404s) live in one place
Service logic can be unit-tested with a fake repository

7) Router: small endpoints, clean dependencies

Create app/tasks/router.py:

from fastapi import APIRouter, Depends, Query from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db from app.tasks.repository import TaskRepository from app.tasks.service import TaskService from app.tasks.schemas import TaskCreate, TaskRead router = APIRouter(prefix="/tasks", tags=["tasks"]) def get_task_service(db: AsyncSession = Depends(get_db)) -> TaskService: repo = TaskRepository(db) return TaskService(repo) @router.post("", response_model=TaskRead, status_code=201) async def create_task( payload: TaskCreate, service: TaskService = Depends(get_task_service), ): task = await service.create_task(payload.title) return task @router.get("", response_model=list[TaskRead]) async def list_tasks( limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), service: TaskService = Depends(get_task_service), ): return await service.list_tasks(limit=limit, offset=offset) @router.post("/{task_id}/complete", response_model=TaskRead) async def complete_task( task_id: int, service: TaskService = Depends(get_task_service), ): return await service.complete_task(task_id)

Two DI layers are happening:

get_db creates a per-request AsyncSession
get_task_service wires repo + service using that session

8) App entrypoint + table creation

Create app/main.py:

from fastapi import FastAPI from app.core.db import Base, engine from app.tasks.router import router as tasks_router app = FastAPI(title="Tasks API") @app.on_event("startup") async def on_startup() -> None: async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) app.include_router(tasks_router)

Run it:

uvicorn app.main:app --reload

Try requests:

curl -X POST http://127.0.0.1:8000/tasks \ -H "Content-Type: application/json" \ -d '{"title":"Buy milk"}'
curl "http://127.0.0.1:8000/tasks?limit=10&offset=0
"
curl -X POST http://127.0.0.1:8000/tasks/1/complete

9) Testing the API without spinning up a server

For junior/mid devs, the biggest test win is: use httpx against the FastAPI app object. Here’s a simple test that mocks the service dependency so you can validate HTTP behavior without touching a real database.

Create tests/test_tasks.py:

import pytest from httpx import AsyncClient from fastapi import Depends from app.main import app from app.tasks.router import get_task_service from app.tasks.schemas import TaskRead class FakeService: def __init__(self): self.items = [TaskRead(id=1, title="Test", done=False)] async def list_tasks(self, limit: int, offset: int): return self.items[offset: offset + limit] @pytest.mark.asyncio async def test_list_tasks_returns_items(): fake = FakeService() app.dependency_overrides[get_task_service] = lambda: fake async with AsyncClient(app=app, base_url="http://test") as ac: res = await ac.get("/tasks?limit=20&offset=0") assert res.status_code == 200 data = res.json() assert data[0]["title"] == "Test" app.dependency_overrides.clear()

Run tests:

pytest -q

This pattern is powerful: override dependencies to test routers quickly, then separately test the service and repository with integration tests when you’re ready.

10) Common pitfalls and small upgrades

Don’t leak sessions into global state. Always pass AsyncSession via dependencies.
Keep “policy” in the service layer. Clamping limits, default sorting, and error mapping belong there.
Return schemas, not ORM models. We used response_model and from_attributes so FastAPI can serialize safely.
Add migrations later. create_all is fine for a demo; production should use Alembic migrations.
Prefer one responsibility per module. When files get big, split further (e.g., separate dependencies.py).

Wrap-up

The goal isn’t “enterprise architecture”—it’s keeping FastAPI apps maintainable as soon as they grow past a few endpoints. With a small structure (router/repo/service) and DI done right, you get cleaner code, easier tests, and fewer “where does this logic live?” moments.

If you want to extend this next, add:

API key or JWT auth as a dependency (e.g., get_current_user)
Alembic migrations
Structured logging + request IDs

FastAPI Development in Practice: Clean Project Structure, Dependency Injection, and a Real “Service Layer”