todo.md yhdistää koodin ja annotaatiot: miksi pattern on valittu, mitä EI saa tehdä. 1567 tokenia (vs raaka 1340, compact 335). Benchmark lataa .md-version oletuksena, fallback erillisiin tiedostoihin.
221 lines
6.2 KiB
Markdown
221 lines
6.2 KiB
Markdown
# Todo — referenssitoteutus (FastAPI + SQLAlchemy 2.0 + SQLite)
|
|
|
|
Tämä on täydellinen esimerkki. Generoi vastaava rakenne annetulle projektille.
|
|
Käytä VAIN JSON-spekin kenttiä — älä lisää ylimääräisiä.
|
|
|
|
## models.py
|
|
|
|
SQLAlchemy 2.0: `DeclarativeBase` + `Mapped` + `mapped_column`. EI `Column()`, EI `declarative_base()`.
|
|
|
|
```python
|
|
"""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)
|
|
```
|
|
|
|
Huomaa:
|
|
- `str | None` (ei `Optional[str]`)
|
|
- `String(20)` status-kentälle (ei Enum)
|
|
- Vain spekin kentät — ei `created_at` tai muita ylimääräisiä
|
|
|
|
## schemas.py
|
|
|
|
Pydantic v2: `ConfigDict(from_attributes=True)`. EI `class Config: orm_mode = True`.
|
|
|
|
```python
|
|
"""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: POST 201, GET list, GET by id 404, PUT, DELETE 204. Käytä `model_dump()` (ei `.dict()`).
|
|
|
|
```python
|
|
"""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
|
|
|
|
Testit: erillinen test.db, `override_get_db`, `TestClient`. Uniikki suomenkielinen data per testi.
|
|
PUT-testi lähettää KAIKKI pakolliset kentät.
|
|
|
|
```python
|
|
"""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
|
|
```
|