A complete solution for a FastAPI CRUD service for a Task resource, including project structure, Pydantic V2 models, SQLModel ORM, database session management, CRUD endpoints, dependency management, and unit tests.
`` . ├── pyproject.toml └── src ├── __init__.py ├── main.py ├── database.py └── models.py └── tests └── test_main.py ``
pyproject.toml ```toml [tool.poetry.dependencies] python = "^3.10" fastapi = "^0.111.0" uvicorn = {extras = ["standard"], version = "^0.30.1"} sqlmodel = "^0.0.18" pydantic = "^2.7.4" pydantic-settings = "^2.3.3"
[tool.poetry.group.dev.dependencies] pytest = "^8.2.2" httpx = "^0.27.0" ```
src/models.py ```python from typing import Optional from sqlmodel import Field, SQLModel from pydantic import ConfigDict
class TaskBase(SQLModel): title: str = Field(index=True) description: Optional[str] = None completed: bool = False
class TaskCreate(TaskBase): pass
class TaskUpdate(SQLModel): title: Optional[str] = None description: Optional[str] = None completed: Optional[bool] = None
class TaskRead(TaskBase): id: int model_config = ConfigDict(from_attributes=True) # Pydantic V2 for ORM mapping
class Task(TaskBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) ```
src/database.py ```python from sqlmodel import create_engine, Session, SQLModel from typing import Generator
DATABASE_URL = "sqlite:///./database.db" engine = create_engine(DATABASE_URL, echo=False)
def create_db_and_tables(): SQLModel.metadata.create_all(engine)
def get_db() -> Generator[Session, None, None]: with Session(engine) as session: yield session ```
src/main.py ```python from fastapi import FastAPI, Depends, HTTPException from sqlmodel import Session, select from typing import List from .database import get_db, create_db_and_tables from .models import Task, TaskCreate, TaskRead, TaskUpdate
app = FastAPI()
@app.on_event("startup") def on_startup(): create_db_and_tables()
@app.post("/tasks/", response_model=TaskRead) def create_task(*, session: Session = Depends(get_db), task: TaskCreate): db_task = Task.model_validate(task) # Pydantic V2 validation session.add(db_task) session.commit() session.refresh(db_task) return db_task
@app.get("/tasks/{task_id}", response_model=TaskRead) def read_task(*, session: Session = Depends(get_db), task_id: int): task = session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") return task
@app.get("/tasks/", response_model=List[TaskRead]) def read_tasks(*, session: Session = Depends(get_db), offset: int = 0, limit: int = 100): tasks = session.exec(select(Task).offset(offset).limit(limit)).all() return tasks
# Additional PUT and DELETE endpoints would follow a similar pattern. ```
tests/test_main.py ```python from fastapi.testclient import TestClient from sqlmodel import SQLModel, create_engine, Session from src.main import app, get_db from src.models import TaskCreate, TaskRead import pytest
TEST_DATABASE_URL = "sqlite:///:memory:" engine = create_engine(TEST_DATABASE_URL)
@pytest.fixture(name="session") def session_fixture(): SQLModel.metadata.create_all(engine) with Session(engine) as session: yield session SQLModel.metadata.drop_all(engine)
@pytest.fixture(name="client") def client_fixture(session: Session): def get_session_override(): return session app.dependency_overrides[get_db] = get_session_override client = TestClient(app) yield client app.dependency_overrides.clear()
def test_create_task(client: TestClient): response = client.post("/tasks/", json={"title": "Test Task", "description": "Test description"}) assert response.status_code == 200 task = TaskRead.model_validate(response.json()) assert task.title == "Test Task" assert task.id is not None
def test_read_task(client: TestClient): create_response = client.post("/tasks/", json={"title": "Read Task"}) created_task = create_response.json() task_id = created_task["id"]
response = client.get(f"/tasks/{task_id}") assert response.status_code == 200 task = TaskRead.model_validate(response.json()) assert task.title == "Read Task"
def test_read_task_not_found(client: TestClient): response = client.get("/tasks/999") assert response.status_code == 404 assert response.json() == {"detail": "Task not found"} ```