SQL File Loader Demo

Load named queries from queries/users.sql and execute them against SQLite.

uv run python docs/examples/loaders/sql_files.py

Source

 1"""Load named SQL statements from disk and execute them with SQLite."""
 2
 3from pathlib import Path
 4
 5from sqlspec import SQLFileLoader, SQLSpec
 6from sqlspec.adapters.sqlite import SqliteConfig
 7
 8__all__ = ("main",)
 9
10
11QUERIES = Path(__file__).resolve().parents[1] / "queries" / "users.sql"
12
13
14def main() -> None:
15    """Create the demo schema, run a named insert, and list rows."""
16    loader = SQLFileLoader()
17    loader.load_sql(QUERIES)
18    registry = SQLSpec()
19    config = registry.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
20    with registry.provide_session(config) as session:
21        session.execute(
22            """
23            CREATE TABLE users (
24                id INTEGER PRIMARY KEY,
25                username TEXT NOT NULL,
26                email TEXT NOT NULL,
27                password_hash TEXT NOT NULL,
28                is_active BOOLEAN NOT NULL DEFAULT 1,
29                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
30                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
31                last_login_at TIMESTAMP
32            )
33            """
34        )
35        session.execute(
36            loader.get_sql("create_user"),
37            {"username": "quinn", "email": "[email protected]", "password_hash": "demo", "is_active": True},
38        )
39        rows = session.select(loader.get_sql("list_active_users"), {"limit": 5, "offset": 0})
40        totals = session.select(loader.get_sql("count_users_by_status"))
41        print({"active": rows, "totals": totals})
42
43
44if __name__ == "__main__":
45    main()