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()