FastAPI in Practice: Build an Async CRUD API with SQLModel + Dependency Injection (and Test It)

FastAPI is popular because it’s fast to build with, easy to validate inputs, and friendly to testing. In this hands-on guide, you’ll build a small but production-shaped API that includes:

Async endpoints with FastAPI
Data models with SQLModel (Pydantic-style models + SQLAlchemy under the hood)
Dependency injection for database sessions
Clean routing structure
Tests that override dependencies (so you don’t hit a real database)

The example is a “Tasks” API: create tasks, list them, update status, and delete. You can reuse this structure for real apps.

1) Project setup

Create a folder and install dependencies:

python -m venv .venv # macOS/Linux source .venv/bin/activate # Windows (PowerShell) # .\.venv\Scripts\Activate.ps1 pip install fastapi uvicorn sqlmodel sqlalchemy aiosqlite httpx pytest pytest-asyncio

Suggested structure:

app/ __init__.py main.py db.py models.py routers/ __init__.py tasks.py tests/ test_tasks.py

2) Define models (database + API schemas)

With SQLModel, you can define a table model and separate “input” models for create/update. This helps prevent clients from setting fields they shouldn’t (like IDs).

# app/models.py from datetime import datetime from typing import Optional from sqlmodel import SQLModel, Field class TaskBase(SQLModel): title: str description: Optional[str] = None class Task(TaskBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) done: bool = Field(default=False, index=True) created_at: datetime = Field(default_factory=datetime.utcnow) class TaskCreate(TaskBase): pass class TaskUpdate(SQLModel): title: Optional[str] = None description: Optional[str] = None done: Optional[bool] = None

3) Database engine + session dependency

The key pattern in FastAPI apps is: “create a dependency that yields a database session.” Your endpoints receive it via Depends. Later, tests can override it.

# app/db.py from typing import AsyncGenerator from sqlmodel import SQLModel from sqlmodel.ext.asyncio.session import AsyncSession from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.orm import sessionmaker DATABASE_URL = "sqlite+aiosqlite:///./tasks.db" engine = create_async_engine( DATABASE_URL, echo=False, # set True if you want SQL logs while learning future=True ) AsyncSessionLocal = sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False ) async def init_db() -> None: async with engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) async def get_session() -> AsyncGenerator[AsyncSession, None]: async with AsyncSessionLocal() as session: yield session

4) Build a router with CRUD endpoints

We’ll keep routes in app/routers/tasks.py. Notice the flow:

Validate input with TaskCreate / TaskUpdate
Use AsyncSession to query/commit
Return a Task (or list of tasks)

# app/routers/tasks.py from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession from app.db import get_session from app.models import Task, TaskCreate, TaskUpdate router = APIRouter(prefix="/tasks", tags=["tasks"]) @router.post("", response_model=Task, status_code=status.HTTP_201_CREATED) async def create_task( payload: TaskCreate, session: AsyncSession = Depends(get_session) ) -> Task: task = Task.model_validate(payload) # turns create schema into table model session.add(task) await session.commit() await session.refresh(task) return task @router.get("", response_model=List[Task]) async def list_tasks( done: Optional[bool] = Query(default=None, description="Filter by done status"), limit: int = Query(default=50, ge=1, le=200), offset: int = Query(default=0, ge=0), session: AsyncSession = Depends(get_session), ) -> List[Task]: stmt = select(Task) if done is not None: stmt = stmt.where(Task.done == done) stmt = stmt.order_by(Task.created_at.desc()).limit(limit).offset(offset) result = await session.exec(stmt) return result.all() @router.get("/{task_id}", response_model=Task) async def get_task( task_id: int, session: AsyncSession = Depends(get_session) ) -> Task: task = await session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return task @router.patch("/{task_id}", response_model=Task) async def update_task( task_id: int, payload: TaskUpdate, session: AsyncSession = Depends(get_session) ) -> Task: task = await session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") # Only apply provided fields (partial update) updates = payload.model_dump(exclude_unset=True) for key, value in updates.items(): setattr(task, key, value) session.add(task) await session.commit() await session.refresh(task) return task @router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_task( task_id: int, session: AsyncSession = Depends(get_session) ) -> None: task = await session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") await session.delete(task) await session.commit() return None

5) Wire it up in `main.py` (startup + include_router)

On startup, initialize tables (for SQLite demos, this is fine; in real production you’ll likely use migrations).

# app/main.py from fastapi import FastAPI from app.db import init_db from app.routers.tasks import router as tasks_router app = FastAPI(title="Tasks API", version="1.0.0") @app.on_event("startup") async def on_startup() -> None: await init_db() app.include_router(tasks_router)

Run the app:

uvicorn app.main:app --reload

Now try it:

Docs UI: /docs
Create a task: POST /tasks
List tasks: GET /tasks (try ?done=true)

6) Testing: override the DB dependency

This is where FastAPI really shines. Because your endpoints depend on get_session, tests can swap it for a test session. We’ll use a temporary SQLite database for tests.

# tests/test_tasks.py import pytest from fastapi.testclient import TestClient from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.orm import sessionmaker from sqlmodel import SQLModel from sqlmodel.ext.asyncio.session import AsyncSession from app.main import app from app.db import get_session from app.models import Task TEST_DB_URL = "sqlite+aiosqlite:///./test_tasks.db" test_engine = create_async_engine(TEST_DB_URL, echo=False, future=True) TestSessionLocal = sessionmaker( bind=test_engine, class_=AsyncSession, expire_on_commit=False ) async def override_get_session(): async with TestSessionLocal() as session: yield session @pytest.fixture(scope="module", autouse=True) def setup_db(): # Create tables once for the module async def _create(): async with test_engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) import asyncio asyncio.get_event_loop().run_until_complete(_create()) app.dependency_overrides[get_session] = override_get_session yield app.dependency_overrides.clear() def test_create_and_list_tasks(): client = TestClient(app) r1 = client.post("/tasks", json={"title": "Ship v1", "description": "Add CRUD + tests"}) assert r1.status_code == 201 data = r1.json() assert data["title"] == "Ship v1" assert data["done"] is False assert "id" in data r2 = client.get("/tasks") assert r2.status_code == 200 items = r2.json() assert len(items) >= 1 def test_patch_task_done(): client = TestClient(app) created = client.post("/tasks", json={"title": "Write docs"}).json() task_id = created["id"] patched = client.patch(f"/tasks/{task_id}", json={"done": True}) assert patched.status_code == 200 assert patched.json()["done"] is True def test_delete_task(): client = TestClient(app) created = client.post("/tasks", json={"title": "Temporary"}).json() task_id = created["id"] deleted = client.delete(f"/tasks/{task_id}") assert deleted.status_code == 204 missing = client.get(f"/tasks/{task_id}") assert missing.status_code == 404

Run tests:

pytest -q

7) Practical tips you’ll reuse in real FastAPI projects

Keep “table models” separate from “input models”. It prevents clients from sending fields like id or created_at.
Return response_model=.... It documents the API, validates outputs, and avoids accidentally leaking fields.
Dependency injection is your testing superpower. If something is hard to test, it’s often because it isn’t injected (DB, cache, external API client).
Use query params for filtering/pagination. Even in small apps, limit/offset patterns keep endpoints scalable.
Don’t auto-migrate in production. For real deployments, use a migration tool (e.g., Alembic) rather than create_all at startup.

Wrap-up

You now have a compact FastAPI codebase that looks like a real project: router modules, async DB sessions, validated schemas, and tests that don’t touch production resources. From here, you can add auth, background jobs, or a proper migration workflow—without needing to rewrite the foundation.

FastAPI in Practice: Build an Async CRUD API with SQLModel + Dependency Injection (and Test It)