Quickstart¶
Run your first parameterized query with zero external setup using Python's built-in SQLite driver:
"""Execute a parameterized SQLite query and fetch a typed dictionary row."""
from typing import Any
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
__all__ = ("main",)
def main() -> dict[str, Any]:
"""Run the first-query quickstart example."""
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
with spec.provide_session(config) as session:
session.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, points INTEGER)")
session.execute("INSERT INTO users (name, points) VALUES (:name, :points)", name="Ada", points=42)
count = session.select_value("SELECT COUNT(*) FROM users", value_type=int)
row = session.select_one("SELECT id, name, points FROM users WHERE name = :name", name="Ada")
assert count == 1
assert row == {"id": 1, "name": "Ada", "points": 42}
return row
if __name__ == "__main__":
print(main())
Execute the script with uv:
Expected output:
Recap¶
Walk through the highlighted lines in first_query.py:
- Line 13 (
SQLSpec()): Creates the central registry that manages database configurations, connection pools, and named SQL files. - Line 14 (
spec.add_config(...)): Registers aSqliteConfigpointing at an:memory:SQLite database and returns a typed handle. - Line 16 (
with spec.provide_session(config) as session): Acquires a pooled connection and yields aSqliteDriversession that cleans up automatically on block exit. - Line 18 (
session.execute(..., name="Ada", points=42)): Compiles:nameand:pointsplaceholders for SQLite and binds keyword arguments safely without string interpolation. - Lines 19–20 (
select_valueandselect_one):select_valueextracts a single scalar column (int), whileselect_onereturns the single matching row as adict[str, Any](raisingNotFoundErrorif zero rows match).
Positional vs Named Parameters¶
SQLSpec accepts both named keyword parameters (:name) and positional placeholders (? or $1) on any adapter, translating them to the underlying driver's wire syntax automatically:
"""Open a synchronous SQLite session and execute a parameterized query."""
from typing import Any
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
__all__ = ("main",)
def main() -> list[dict[str, Any]]:
"""Insert and fetch a note in an in-memory SQLite database."""
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
with spec.provide_session(config) as session:
session.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT)")
session.execute("INSERT INTO notes (body) VALUES (?)", "Hello, SQLSpec")
rows = session.select("SELECT id, body FROM notes")
assert rows == [{"id": 1, "body": "Hello, SQLSpec"}]
return rows
if __name__ == "__main__":
print(main())
Use positional parameters for quick single-column statements and named parameters whenever a query binds multiple fields or composes StatementFilter objects.
Asynchronous Sessions¶
Switching from synchronous sqlite3 to asyncio uses the exact same session methods with await and async with. Install sqlspec[aiosqlite] and swap SqliteConfig for AiosqliteConfig:
"""Open an asynchronous SQLite session with AiosqliteConfig and run a query."""
import asyncio
from typing import Any
from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig
__all__ = ("main",)
async def main() -> list[dict[str, Any]]:
"""Insert and read a note using an asynchronous SQLite session."""
spec = SQLSpec()
config = spec.add_config(AiosqliteConfig(connection_config={"database": ":memory:"}))
try:
async with spec.provide_session(config) as session:
await session.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT)")
await session.execute("INSERT INTO notes (body) VALUES (?)", "Hello, Async SQLSpec")
rows = await session.select("SELECT id, body FROM notes")
assert rows == [{"id": 1, "body": "Hello, Async SQLSpec"}]
return rows
finally:
await config.close_pool()
if __name__ == "__main__":
print(asyncio.run(main()))
Always call await config.close_pool() during application shutdown (or let a framework plugin manage lifespan cleanup automatically).
Typed Row Mapping¶
By default, select and select_one return dict[str, Any] rows. Pass schema_type= to decode rows directly into msgspec.Struct, TypedDict, pydantic.BaseModel, dataclass, or attrs instances (uv add "sqlspec[msgspec]"):
"""Decode query rows directly into typed msgspec.Struct instances."""
import msgspec
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
__all__ = ("User", "main")
class User(msgspec.Struct, frozen=True, gc=False):
"""Typed user row returned from SQLite."""
id: int
name: str
points: int
def main() -> User:
"""Insert a user record and decode it into a frozen msgspec.Struct."""
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
with spec.provide_session(config) as session:
session.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, points INTEGER)")
session.execute("INSERT INTO users (name, points) VALUES (:name, :points)", name="Ada", points=42)
user = session.select_one("SELECT id, name, points FROM users WHERE name = :name", name="Ada", schema_type=User)
assert isinstance(user, User)
assert user.name == "Ada"
assert user.points == 42
return user
if __name__ == "__main__":
loaded = main()
print(f"Loaded {loaded.name} with {loaded.points} points")
Defining msgspec.Struct(..., frozen=True, gc=False) gives you immutable attribute access, static type checking, and lower allocation overhead than plain dictionaries.
Moving from :memory: to a Local File¶
Replace ":memory:" with a file path to persist SQLite state across process restarts while keeping the rest of your query code identical:
from pathlib import Path
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(Path("app.db"))}))
When you are ready to point the same queries at PostgreSQL, Oracle, DuckDB, Spanner, or MySQL, swap SqliteConfig for the corresponding adapter configuration without rewriting your session.select(...) or session.execute(...) calls.
Next Steps¶
| Goal | Recommended Guide |
|---|---|
Register multiple pools, bind_key names, or custom statement settings |
Configuration |
Choose between select, select_one_or_none, execute_many, StatementStack, and select_stream |
Querying |
Move multi-line SQL out of Python strings into -- name: .sql files |
SQL Files |
Manage schema revisions with the sqlspec CLI |
Migrations |
| Connect to PostgreSQL, Oracle, DuckDB, Spanner, BigQuery, MySQL, SQL Server, or Db2 | Databases |
| Wire request-scoped sessions into Litestar, FastAPI, Google ADK, or Dishka | Integrations |