Overview

-- Capstone async FastAPI service schema (applied once at startup, co-16/co-18).
CREATE TABLE IF NOT EXISTS tasks (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  title TEXT NOT NULL,
  description TEXT NOT NULL DEFAULT '',
  status TEXT NOT NULL DEFAULT 'todo',
  created_at TEXT NOT NULL DEFAULT (datetime ('now'))
);

"""Capstone async FastAPI service -- the ONLY module that talks to the database (co-16, co-15)."""
 
from pathlib import Path
 
import aiosqlite  # => the async driver -- query waits yield to the loop (co-16)
 
from .models import Task, TaskCreate, TaskPage, TaskUpdate
 
SCHEMA_PATH = Path(__file__).parent / "schema.sql"  # => the schema applied at startup (co-16)
 
 
async def get_connection(db_path: str) -> aiosqlite.Connection:  # => one async connection per request (co-15)
    conn = await aiosqlite.connect(db_path)  # => async acquire
    conn.row_factory = aiosqlite.Row  # => rows addressable by column name
    return conn
 
 
async def init_db(db_path: str) -> None:  # => apply schema.sql once at startup (co-16, co-18)
    conn = await get_connection(db_path)
    await conn.executescript(SCHEMA_PATH.read_text(encoding="utf-8"))  # => idempotent schema
    await conn.commit()
    await conn.close()
 
 
def _row_to_task(row: aiosqlite.Row) -> Task:  # => the ONE place a raw row becomes a typed Task (co-14)
    return Task(
        id=int(row["id"]),
        title=str(row["title"]),
        description=str(row["description"]),
        status=str(row["status"]),  # type: ignore[arg-type]  # => Pydantic validates against TaskStatus
        created_at=str(row["created_at"]),
    )
 
 
async def create_task(conn: aiosqlite.Connection, data: TaskCreate) -> Task:  # => parameterized INSERT (co-16)
    cursor = await conn.execute(  # => ? placeholders, never f-strings (co-16)
        "INSERT INTO tasks (title, description) VALUES (?, ?)",
        (data.title, data.description),
    )
    await conn.commit()
    row = await (await conn.execute("SELECT * FROM tasks WHERE id = ?", (cursor.lastrowid,))).fetchone()
    assert row is not None  # => guaranteed by the INSERT above -- narrows Row | None for strict pyright
    return _row_to_task(row)  # => re-read to include DB-generated defaults
 
 
async def get_task(conn: aiosqlite.Connection, task_id: int) -> Task | None:  # => parameterized SELECT (co-16)
    row = await (await conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,))).fetchone()
    return _row_to_task(row) if row is not None else None  # => None signals "not found" (co-17)
 
 
async def update_task(  # => parameterized UPDATE (co-16)
    conn: aiosqlite.Connection, task_id: int, data: TaskUpdate
) -> Task | None:
    cursor = await conn.execute(
        "UPDATE tasks SET title = ?, description = ?, status = ? WHERE id = ?",
        (data.title, data.description, data.status, task_id),
    )
    if cursor.rowcount == 0:  # => no row matched
        return None
    await conn.commit()
    row = await (await conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,))).fetchone()
    assert row is not None  # => rowcount > 0 above guarantees this
    return _row_to_task(row)
 
 
async def delete_task(conn: aiosqlite.Connection, task_id: int) -> bool:  # => parameterized DELETE (co-16)
    cursor = await conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
    await conn.commit()
    return cursor.rowcount > 0  # => True only if a row genuinely existed
 
 
async def list_tasks(  # => pagination + filtering in ONE parameterized query (co-11, co-16)
    conn: aiosqlite.Connection, limit: int, offset: int, status: str | None
) -> TaskPage:
    where = " WHERE status = ?" if status is not None else ""  # => an optional filter clause
    params: list[object] = [status] if status is not None else []
    total_row = await (await conn.execute(f"SELECT COUNT(*) AS n FROM tasks{where}", params)).fetchone()
    assert total_row is not None  # => COUNT(*) always returns one row
    total = int(total_row["n"])  # => the FILTERED total (co-11)
    rows = await (await conn.execute(f"SELECT * FROM tasks{where} ORDER BY id LIMIT ? OFFSET ?", [*params, limit, offset])).fetchall()
    items = [_row_to_task(row) for row in rows]
    next_offset = offset + limit
    return TaskPage(items=items, total=total, next=next_offset if next_offset < total else None)  # => None at end
 
 
async def ping(conn: aiosqlite.Connection) -> bool:  # => cheapest real query -- used by /ready (co-16)
    await conn.execute("SELECT 1")
    return True

{"status":"ok"}
{"status":"ready"}

"""Capstone async FastAPI service -- typed Pydantic v2 models (co-12, co-13, co-14)."""
 
from typing import Literal
 
from pydantic import BaseModel, Field
 
# => a closed status set -- anything else is a 422 (co-13)
TaskStatus = Literal["todo", "in_progress", "done"]
 
 
class TaskCreate(BaseModel):  # => the POST /tasks body shape (co-12)
    title: str = Field(min_length=1, max_length=200)  # => constrained -- empty titles rejected (co-13)
    description: str = Field(default="", max_length=2000)
 
 
class TaskUpdate(BaseModel):  # => the PUT /tasks/{id} body shape (co-12)
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(default="", max_length=2000)
    status: TaskStatus = "todo"  # => a closed set -- an unknown status is a 422 (co-13)
 
 
class Task(BaseModel):  # => the response shape for every task-returning endpoint (co-14)
    id: int
    title: str
    description: str
    status: TaskStatus
    created_at: str
 
 
class TaskPage(BaseModel):  # => the paginated list envelope (co-14, co-11)
    items: list[Task]
    total: int
    next: int | None  # => None once the last page is reached

{"id":1,"title":"write report","description":"Q3","status":"todo","created_at":"2026-07-29 12:00:00"}
{"detail":[{"type":"string_too_short","loc":["body","title"],"msg":"String should have at least 1 character"}]}

"""Capstone async FastAPI service -- routing, DI, errors, streaming, config, logging (co-01 to co-24).
 
Run with: uvicorn app.main:app --port 8000  (from learning/capstone/code/).
"""
 
import asyncio  # => asyncio drives the streaming generator (co-22)
import logging  # => structured logging (co-24)
import os
import time
from collections.abc import AsyncIterator, Awaitable, Callable
 
import aiosqlite  # => the async driver -- the DB session type (co-16)
from fastapi import Depends, FastAPI, HTTPException, Query, Request  # => DI + routing + errors (co-10, co-15, co-17)
from fastapi.responses import JSONResponse, Response, StreamingResponse  # => streaming + error envelope (co-17, co-22)
from pydantic_settings import BaseSettings, SettingsConfigDict  # => env config (co-24)
 
from . import repository as repo
from .models import Task, TaskCreate, TaskPage, TaskUpdate
 
logging.basicConfig(level=logging.INFO, format="%(message)s")  # => one line per record (co-24)
logger = logging.getLogger("capstone")  # => a named logger
 
 
class Settings(BaseSettings):  # => typed config from env (co-24)
    model_config = SettingsConfigDict(env_prefix="CAPSTONE_")  # => CAPSTONE_ prefix
    db_path: str = os.path.join(os.path.dirname(__file__), "capstone.db")  # => overridable so tests use a temp DB
    env: str = "dev"
 
 
settings = Settings()  # => resolved once at startup (co-24)
 
 
async def init_at_startup() -> None:  # => apply schema once before serving (co-18, co-16)
    await repo.init_db(settings.db_path)
 
 
app = FastAPI(title="Capstone Async FastAPI Service", on_startup=[init_at_startup])  # => co-18 startup hook
 
 
@app.exception_handler(HTTPException)  # => one consistent {"error": {...}} envelope app-wide (co-17)
async def handle_http_exception(request: Request, exc: HTTPException) -> JSONResponse:
    _ = request  # => available for logging, unused in the mapping
    body = exc.detail if isinstance(exc.detail, dict) else {"error": {"code": "error", "message": str(exc.detail)}}
    return JSONResponse(status_code=exc.status_code, content=body)
 
 
@app.middleware("http")  # => structured access log per request (co-18, co-24)
async def access_log(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
    start = time.perf_counter()  # => baseline
    response: Response = await call_next(request)  # => run the handler
    logger.info(  # => structured: fixed keys, machine-parseable (co-24)
        "method=%s path=%s status=%d ms=%.3f",
        request.method, request.url.path, response.status_code, (time.perf_counter() - start) * 1000,
    )
    return response
 
 
async def get_db() -> AsyncIterator[aiosqlite.Connection]:  # => one async session per request (co-15, co-16)
    conn = await repo.get_connection(settings.db_path)  # => async acquire (co-16)
    try:
        yield conn  # => hand the live session to the handler
    finally:
        await conn.close()  # => close on exit, even if the handler raised (co-15)
 
 
@app.get("/health")  # => LIVENESS -- always 200, no DB dependency (co-10)
async def health() -> dict[str, str]:
    return {"status": "ok"}
 
 
@app.get("/ready")  # => READINESS -- genuinely pings the async DB (co-16)
async def ready(conn: aiosqlite.Connection = Depends(get_db)) -> dict[str, str]:
    await repo.ping(conn)  # => async probe -- yields to the loop (co-16)
    return {"status": "ready"}
 
 
@app.post("/tasks", response_model=Task, status_code=201)  # => create (co-17)
async def create_task_route(body: TaskCreate, conn: aiosqlite.Connection = Depends(get_db)) -> Task:
    return await repo.create_task(conn, body)  # => async repository call (co-16)
 
 
@app.get("/tasks/{task_id}", response_model=Task)  # => read one, 404 on missing (co-17)
async def read_task_route(task_id: int, conn: aiosqlite.Connection = Depends(get_db)) -> Task:
    task = await repo.get_task(conn, task_id)  # => async SELECT
    if task is None:
        raise HTTPException(status_code=404, detail={"error": {"code": "not_found", "message": "no such task"}})  # => co-17
    return task
 
 
@app.put("/tasks/{task_id}", response_model=Task)  # => replace (co-17)
async def update_task_route(task_id: int, body: TaskUpdate, conn: aiosqlite.Connection = Depends(get_db)) -> Task:
    updated = await repo.update_task(conn, task_id, body)
    if updated is None:
        raise HTTPException(status_code=404, detail={"error": {"code": "not_found", "message": "no such task"}})
    return updated
 
 
@app.delete("/tasks/{task_id}", status_code=204)  # => delete (co-17)
async def delete_task_route(task_id: int, conn: aiosqlite.Connection = Depends(get_db)) -> None:
    if not await repo.delete_task(conn, task_id):
        raise HTTPException(status_code=404, detail={"error": {"code": "not_found", "message": "no such task"}})
 
 
@app.get("/tasks", response_model=TaskPage)  # => pagination + filtering (co-11, co-16)
async def list_tasks_route(
    limit: int = Query(default=10, ge=1, le=50),  # => bounded limit (co-11, co-13)
    offset: int = Query(default=0, ge=0),
    status: str | None = Query(default=None),
    conn: aiosqlite.Connection = Depends(get_db),
) -> TaskPage:
    return await repo.list_tasks(conn, limit, offset, status)
 
 
async def event_stream() -> AsyncIterator[bytes]:  # => a streaming generator (co-22)
    for i in range(3):  # => three events
        await asyncio.sleep(0.01)  # => pace the stream (co-02)
        yield f"data: event {i}\n\n".encode("utf-8")  # => one SSE-shaped chunk per yield (co-22)
 
 
@app.get("/events")  # => the streaming endpoint (co-22)
async def events() -> StreamingResponse:
    return StreamingResponse(event_stream(), media_type="text/event-stream")  # => incremental SSE

data: event 0
 
data: event 1
 
data: event 2
 

"""Capstone async FastAPI service -- async acceptance suite (co-21).
 
Drives the app in-process via httpx ASGITransport, each test against a FRESH temp DB. Run: pytest -v.
"""
 
from pathlib import Path
 
import httpx  # => the async client (co-21)
import pytest
import pytest_asyncio  # => async fixtures need the asyncio decorator (co-21)
from httpx import ASGITransport
 
from app import main  # => the app under test (co-21)
from app.main import app  # => the ASGI application
from app.repository import init_db  # => explicit schema init (co-16)
 
 
@pytest_asyncio.fixture()
async def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> httpx.AsyncClient:  # => a client per test
    db_file = tmp_path / "capstone.db"  # => a FRESH DB file per test
    main.settings.db_path = str(db_file)  # => point the shared settings at the temp file (co-24)
    await init_db(str(db_file))  # => ASGITransport skips lifespan; init schema explicitly (co-16)
    transport = ASGITransport(app=app)  # => in-process transport against the temp DB (co-21)
    return httpx.AsyncClient(transport=transport, base_url="http://test")  # => a pooled client
 
 
@pytest.mark.asyncio
async def test_health_is_200(client: httpx.AsyncClient) -> None:  # => liveness, no DB
    async with client:
        response = await client.get("/health")
        assert response.status_code == 200
        assert response.json() == {"status": "ok"}
 
 
@pytest.mark.asyncio
async def test_crud_round_trip(client: httpx.AsyncClient) -> None:  # => create, read, update, delete (co-16, co-17)
    async with client:
        created = await client.post("/tasks", json={"title": "write report", "description": "Q3"})
        assert created.status_code == 201
        task_id = created.json()["id"]
 
        read = await client.get(f"/tasks/{task_id}")  # => reads back the created row (co-16)
        assert read.status_code == 200
        assert read.json()["status"] == "todo"
 
        updated = await client.put(  # => update -> done
            f"/tasks/{task_id}", json={"title": "write report", "description": "Q3", "status": "done"}
        )
        assert updated.status_code == 200
        assert updated.json()["status"] == "done"
 
        deleted = await client.delete(f"/tasks/{task_id}")  # => delete (co-17)
        assert deleted.status_code == 204
 
        gone = await client.get(f"/tasks/{task_id}")  # => now missing -> 404 (co-17)
        assert gone.status_code == 404
 
 
@pytest.mark.asyncio
async def test_invalid_body_returns_422(client: httpx.AsyncClient) -> None:  # => validation at the boundary (co-13)
    async with client:
        response = await client.post("/tasks", json={"title": ""})  # => empty title violates min_length
        assert response.status_code == 422
 
 
@pytest.mark.asyncio
async def test_pagination_and_filter(client: httpx.AsyncClient) -> None:  # => pagination + filtering (co-11)
    async with client:
        for i in range(5):  # => seed 5 tasks
            await client.post("/tasks", json={"title": f"task {i}"})
        page = await client.get("/tasks", params={"limit": 2, "offset": 0})  # => first page of 2
        body = page.json()
        assert len(body["items"]) == 2
        assert body["total"] == 5
        assert body["next"] == 2  # => a next page exists
 
 
@pytest.mark.asyncio
async def test_streaming_endpoint(client: httpx.AsyncClient) -> None:  # => the streaming endpoint (co-22)
    async with client:
        response = await client.get("/events")
        assert response.status_code == 200
        assert "event" in response.text  # => the streamed body arrived incrementally (co-22)

{
  "typeCheckingMode": "strict",
  "pythonVersion": "3.13",
  "include": ["app"]
}

test_app.py::test_health_is_200 PASSED
test_app.py::test_crud_round_trip PASSED
test_app.py::test_invalid_body_returns_422 PASSED
test_app.py::test_pagination_and_filter PASSED
test_app.py::test_streaming_endpoint PASSED