FastAPI + SQLite Application¶

 1"""FastAPI application backed by SQLSpec and SQLite."""
 2
 3import asyncio
 4from collections.abc import Iterator
 5from typing import Annotated, Any
 6
 7from fastapi import Depends, FastAPI
 8
 9from docs.examples.shared.configs import sqlite_registry
10from docs.examples.shared.data import ARTICLES, CREATE_ARTICLES
11from sqlspec.adapters.sqlite import SqliteDriver
12from sqlspec.core import SQL
13
14__all__ = ("get_session", "list_articles", "main", "on_startup", "seed_database")
15
16
17registry, config = sqlite_registry()
18app = FastAPI()
19
20
21def seed_database() -> None:
22    """Create the demo schema and seed rows."""
23    with config.provide_session() as session:
24        session.execute(CREATE_ARTICLES)
25        for row in ARTICLES:
26            session.execute(
27                SQL(
28                    """
29                    INSERT OR REPLACE INTO articles (id, title, body)
30                    VALUES (:id, :title, :body)
31                    """
32                ),
33                row,
34            )
35
36
37@app.on_event("startup")
38async def on_startup() -> None:
39    """Seed data when the ASGI app boots."""
40    await asyncio.to_thread(seed_database)
41
42
43def get_session() -> "Iterator[SqliteDriver]":
44    """Yield a synchronous SQLite driver for request handlers."""
45    with config.provide_session() as session:
46        yield session
47
48
49@app.get("/articles")
50def list_articles(db_session: Annotated["SqliteDriver", Depends(get_session)]) -> "list[dict[str, Any]]":
51    """Return all demo articles sorted by ID."""
52    result = db_session.execute("SELECT id, title, body FROM articles ORDER BY id")
53    return result.all()
54
55
56def main() -> None:
57    """Seed the database without running the ASGI server."""
58    seed_database()
59
60
61if __name__ == "__main__":
62    main()