/sqlspec

name: sqlspec
description: "Use when working with sqlspec, SQLSpec, SQLFileLoader, drivers, query builders, named SQL, filters, pagination, Arrow, framework extensions, ADK stores, or observers. Not for ORM repositories — use advanced-alchemy."

SQLSpec Skill

SQLSpec is a **type-safe SQL query mapper for Python** -- NOT an ORM. It provides flexible connectivity with consistent interfaces across its database adapter packages. Write raw SQL, use the builder API, or load SQL from files. Statements pass through a sqlglot-powered AST pipeline for validation, parameter handling, and dialect conversion.

Match-Your-Framework — read first

sqlspec ships first-party extensions for five web frameworks. If your project uses one of these, **jump directly to the matching integration guide and skip the others**:

• **Litestar** — register configs on `SQLSpec`, then pass that registry to `SQLSpecPlugin`. The plugin adds DI, the `litestar db` CLI, and request observability. See [`references/extensions.md`](references/extensions.md).
• **FastAPI** → [`references/fastapi-integration.md`](references/fastapi-integration.md) — `Depends(plugin.provide_session())` DI, `Annotated[...]` handlers, filter providers.
• **Flask** → [`references/flask-integration.md`](references/flask-integration.md) — `plugin.init_app(app)`, pull-based `plugin.get_session()`, async-via-portal.
• **Starlette** → [`references/starlette-integration.md`](references/starlette-integration.md) — `request.state`-based session access, lifespan wrapping, middleware variants.
• **Sanic** — first-party ASGI-style extension for Sanic applications; match Sanic's app/request lifecycle instead of copying Litestar DI examples.

Shared topics that apply to every framework live in [`references/commit-modes.md`](references/commit-modes.md) (autocommit / manual middleware) and [`references/multi-database.md`](references/multi-database.md) (multi-config registry). Read the framework guide first, then those for depth.

The rest of this SKILL.md covers framework-agnostic topics: adapter setup, query builder, driver methods, filters, observability, migrations, the ADK extension, and data-dictionary introspection.

Code Style Rules

• **`from __future__ import annotations` rule** — SQLSpec adapter config modules and driver definitions avoid `from __future__ import annotations` because configs are introspected at runtime. Consumer application modules (handlers, services, tests that *use* a configured driver) MAY and typically SHOULD use it — canonical Litestar apps use it in 100+ files.

Quick Reference

Adapter Pattern

from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    connection_config={
        "dsn": "postgresql://user:pass@localhost:5432/mydb",
        "min_size": 2,
        "max_size": 10,
    },
)
db_manager = SQLSpec()
db_manager.add_config(config)

async with db_manager.provide_session(config) as db:
    users = await db.select(
        "SELECT * FROM users WHERE active = $1",
        True,
        schema_type=User,
    )

Query Builder Essentials

from sqlspec import sql

stmt = (
    sql.select("id", "name", "email")
    .from_("users")
    .where_eq("status", "active")
    .where("created_at > :since", since=cutoff_date)
    .order_by("created_at", desc=True)
    .limit(50)
    .to_statement()
)

insert_stmt = (
    sql.insert("users").columns("name", "email").values(name="Alice", email="alice@example.com").to_statement()
)

merge_stmt = (
    sql.merge("inventory", dialect="postgres")
    .using("updates")
    .on("inventory.product_id = updates.product_id")
    .when_matched_then_update(qty="updates.qty")
    .when_not_matched_then_insert(product_id="updates.product_id", qty="updates.qty")
    .to_statement()
)

Driver Method Summary

| Method | Returns | Use Case | | --- | --- | --- | | `select()` / `fetch()` | List of rows | Filtered queries, listing | | `select_value()` | Single scalar | `COUNT(*)`, `MAX()`, existence checks | | `select_value_or_none()` | Scalar or `None` | Optional scalar lookup | | `select_one()` | One row (strict) | Get-by-ID, raises `NotFoundError` | | `select_one_or_none()` | One row or `None` | Optional lookup | | `select_with_total()` | Rows plus total | Pagination | | `select_stream()` / `fetch_stream()` | Context-managed row stream | Bounded row iteration where adapter supports native streaming | | `select_to_arrow()` / `fetch_to_arrow()` | `ArrowResult` | Bulk data export, analytics | | `select_to_storage()` | `StorageBridgeJob` | Export query results to local or cloud storage | | `execute()` | `SQLResult` | INSERT/UPDATE/DELETE metadata | | `execute_many()` | `SQLResult` | Batch operation metadata | | `execute_script()` | `SQLResult` | Multi-statement SQL script execution | | `execute_stack()` | `tuple[StackResult, ...]` | Ordered statement-stack execution | | `load_from_arrow()` | `StorageBridgeJob` | Adapter-supported Arrow ingest | | `load_from_storage()` | `StorageBridgeJob` | Adapter-supported staged-file ingest | | `load_from_records()` | `StorageBridgeJob` | Records normalized through the Arrow ingest path | | `transaction()` | Context manager | Atomic transaction or savepoint scope on a driver |

Arrow Integration Basics

arrow_result = await db.select_to_arrow(
    "SELECT * FROM large_dataset WHERE region = $1",
    region,
    return_format="reader",
    batch_size=10_000,
)

await db.load_from_arrow("users", arrow_result)
await db.load_from_records("users", [{"id": 1, "name": "Ada"}])

<workflow>

Workflow

Step 1: Choose Adapter and Pattern

| Need | Adapter | Key Feature | | --- | --- | --- | | PostgreSQL async | `asyncpg`, `psycopg` | Async, NUMERIC/PYFORMAT params, auto-probed `pgvector` / `pg_textsearch` / `paradedb` | | PostgreSQL sync | `psycopg` | Sync+async, PYFORMAT params | | SQLite | `sqlite`, `aiosqlite` | QMARK params, local dev | | DuckDB analytics