agentic-studio/kipina-codebench/golden-examples/todo-readme.md

Todo App — FastAPI + SQLAlchemy + SQLite

A simple todo CRUD API. Uses only the fields defined in the spec — no extra fields.

Project Structure

models.py       # SQLAlchemy 2.0 models
schemas.py      # Pydantic v2 schemas
main.py         # FastAPI CRUD endpoints
test_main.py    # Pytest with TestClient

models.py

"""Tietokantamallit — SQLAlchemy 2.0, Mapped-tyypitys, SQLite."""

from datetime import date

from sqlalchemy import String, Text, Date, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class Base(DeclarativeBase):
    pass


class Todo(Base):
    """Tehtävä — otsikko, kuvaus, deadline, prioriteetti ja status."""

    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(255))
    description: Mapped[str | None] = mapped_column(Text, default=None)
    due_date: Mapped[date | None] = mapped_column(Date, default=None)
    priority: Mapped[int] = mapped_column(default=1)
    status: Mapped[str] = mapped_column(String(20), default="pending")


Base.metadata.create_all(bind=engine)

schemas.py

"""Pydantic v2 -skeemat — Create sisääntulolle, Response vastaukselle."""

from datetime import date

from pydantic import BaseModel, ConfigDict


class TodoCreate(BaseModel):
    """Uuden tehtävän luonti. Pakolliset: title."""

    title: str
    description: str | None = None
    due_date: date | None = None
    priority: int = 1
    status: str = "pending"


class TodoResponse(TodoCreate):
    """Palautettava tehtävä — sisältää id:n."""

    id: int
    model_config = ConfigDict(from_attributes=True)

main.py

"""FastAPI CRUD — yksi endpoint-setti per entiteetti."""

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

from models import SessionLocal, Todo
from schemas import TodoCreate, TodoResponse

app = FastAPI()


def get_db():
    """Tietokantasessio per pyyntö."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/todos/", response_model=TodoResponse, status_code=201)
def create_todo(item: TodoCreate, db: Session = Depends(get_db)):
    db_item = Todo(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


@app.get("/todos/", response_model=list[TodoResponse])
def list_todos(db: Session = Depends(get_db)):
    return db.query(Todo).all()


@app.get("/todos/{item_id}", response_model=TodoResponse)
def get_todo(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Todo).filter(Todo.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Todo not found")
    return item


@app.put("/todos/{item_id}", response_model=TodoResponse)
def update_todo(item_id: int, item: TodoCreate, db: Session = Depends(get_db)):
    db_item = db.query(Todo).filter(Todo.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Todo not found")
    for key, value in item.model_dump().items():
        setattr(db_item, key, value)
    db.commit()
    db.refresh(db_item)
    return db_item


@app.delete("/todos/{item_id}", status_code=204)
def delete_todo(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Todo).filter(Todo.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Todo not found")
    db.delete(db_item)
    db.commit()

test_main.py

Exactly 6 tests per entity. Database is shared — use >= 1 not == 1 in list tests. For child entities with foreign keys: create parent FIRST, then child with parent's id.

"""Pytest — TestClient, erillinen test.db, uniikki data per testi."""

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from main import app, get_db
from models import Base

test_engine = create_engine(
    "sqlite:///./test.db", connect_args={"check_same_thread": False}
)
TestSession = sessionmaker(autocommit=False, autoflush=False, bind=test_engine)
Base.metadata.create_all(bind=test_engine)


def override_get_db():
    db = TestSession()
    try:
        yield db
    finally:
        db.close()


app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)


def test_create_todo():
    response = client.post("/todos/", json={"title": "Osta maitoa", "priority": 2})
    assert response.status_code == 201
    assert response.json()["title"] == "Osta maitoa"
    assert "id" in response.json()


def test_list_todos():
    client.post("/todos/", json={"title": "Listattava tehtävä"})
    response = client.get("/todos/")
    assert response.status_code == 200
    assert len(response.json()) >= 1


def test_get_todo_by_id():
    created = client.post("/todos/", json={"title": "Haettava tehtävä"}).json()
    response = client.get(f"/todos/{created['id']}")
    assert response.status_code == 200
    assert response.json()["id"] == created["id"]


def test_get_todo_not_found():
    response = client.get("/todos/99999")
    assert response.status_code == 404


def test_update_todo():
    created = client.post("/todos/", json={"title": "Vanha otsikko"}).json()
    response = client.put(
        f"/todos/{created['id']}", json={"title": "Uusi otsikko"}
    )
    assert response.status_code == 200
    assert response.json()["title"] == "Uusi otsikko"


def test_delete_todo():
    created = client.post("/todos/", json={"title": "Poistettava"}).json()
    response = client.delete(f"/todos/{created['id']}")
    assert response.status_code == 204
    response = client.get(f"/todos/{created['id']}")
    assert response.status_code == 404