Sqlspec

SQLSpec Skill

SQLSpec is a type-safe SQL query mapper for Python -- NOT an ORM. It provides flexible connectivity with consistent interfaces across 19 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.
• FastAPI → references/fastapi-integration.md — Depends(plugin.provide_session()) DI, Annotated[...] handlers, filter providers.
• Flask → references/flask-integration.md — plugin.init_app(app), pull-based plugin.get_session(), async-via-portal.
• Starlette → 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 (autocommit / manual middleware) and 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

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

Step 1: Choose Adapter and Pattern

Step 2: Implement

1. Configure the adapter with connection details and pool settings
2. Register the config with SQLSpec.add_config() and use SQLSpec.provide_session(config) for connection lifecycle
3. Choose the appropriate driver method for your query shape
4. Use schema_type parameter for typed results (Pydantic or msgspec models)
5. Apply filters with LimitOffsetFilter, OrderByFilter, SearchFilter
6. Use select_stream(..., native_only=True) when bounded-memory streaming is mandatory
7. Check adapter ingest capabilities, then use load_from_records() or load_from_arrow() for high-volume ingest

Step 3: Validate

Run through the validation checkpoint below before considering the work complete.

Guardrails

• Always use typed adapters: import the specific adapter config, not generic base classes
• Always use schema_type for query results -- get typed objects, not raw dicts
• Always use context managers for driver lifecycle -- async with db_manager.provide_session(config) as db:
• Prefer the query builder for complex dynamic queries -- avoids string concatenation, handles dialect conversion
• Prefer SQLFileLoader for static queries -- keeps SQL out of Python and reuses the global file-cache namespace
• Use -- param: declarations for named SQL files that cross service boundaries -- load-time and execute-time validation catches name drift and required parameter omissions
• Use native_only=True for streaming or Arrow paths only when fallback is unacceptable -- unsupported adapters otherwise use eager row conversion
• Pass regular query bind values as positional arguments -- await db.select("... WHERE id = $1", user_id, schema_type=User), not await db.select(..., [user_id], ...)
• Never concatenate SQL strings -- use parameterized queries or the query builder
• Never hold connections outside context managers -- connection leaks exhaust the pool
• Match parameter style to adapter: $1 for asyncpg, %s for psycopg, ? for sqlite, :name for oracledb
• Cloud adapter controls -- BigQuery job controls live in driver_features; Spanner request controls live in driver_features or provide_session() kwargs
• Adapter config / driver modules avoid from __future__ import annotations. Consumer app modules MAY use it.

Validation Checkpoint

Before delivering SQLSpec code, verify:

• Adapter config uses the correct import path (sqlspec.adapters.<name>)
• Connection lifecycle uses SQLSpec.provide_session(config) context manager
• Parameter style matches the adapter (see adapter registry table)
• Query results use schema_type for type-safe mapping
• Complex dynamic queries use the builder API, not string concatenation
• Filters use SQLSpec filter objects (LimitOffsetFilter, etc.) not manual LIMIT/OFFSET
• Streaming code uses context managers and sets native_only=True when eager fallback would be a bug
• Bulk ingest code checks the adapter matrix before using load_from_arrow(), load_from_storage(), or load_from_records()
• ADK stores are selected from supported adapter adk packages; BigQuery is not an ADK backend

Example

Task: "Set up an asyncpg adapter, define a typed model, and execute a parameterized query with pagination."

from dataclasses import dataclass
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.core.filters import LimitOffsetFilter, OrderByFilter


@dataclass
class User:
    id: int
    name: str
    email: str
    active: bool


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 def list_active_users(page: int = 1, page_size: int = 25) -> list[User]:
    filters = [
        OrderByFilter(field_name="name", sort_order="asc"),
        LimitOffsetFilter(limit=page_size, offset=(page - 1) * page_size),
    ]

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


async def get_user_count() -> int:
    async with db_manager.provide_session(config) as db:
        count = await db.select_value("SELECT COUNT(*) FROM users WHERE active = $1", True)
        return count

References Index

Choosing between sqlspec and advanced-alchemy: advanced-alchemy gives you an opinionated ORM service layer with UUIDAuditBase, lifecycle hooks, repository / service / Alembic integration, and OffsetPagination[T] out of the box — pick it when you want a complete CRUD surface with attribute-style row access and you're happy inside the SQLAlchemy ecosystem. sqlspec gives you direct SQL control, 19 adapter packages (asyncpg, oracledb, DuckDB, BigQuery, SQLite, and more), Arrow result paths for analytics, and a builder API when you need it — pick it when you want explicit SQL, heterogeneous database backends, or Arrow integration. Both skills integrate with Litestar via first-party plugins; see ../advanced-alchemy/SKILL.md for the ORM path.

For detailed instructions, patterns, and API guides, refer to the following documents:

Standards & Style

• Code Quality & Mypyc -- Type annotation rules, import standards, test structure.

Core Utilities

• SQLglot Best Practices -- v30+ guardrails, AST manipulation, copy=False pattern.

Architecture & Performance

• Architecture & Caching -- Core data flow, global cache configuration, namespaces, and driver-local statement caches.
• Performance & Cloud Controls -- Bounded async bridge, cache/fetch tuning, BigQuery job controls, Spanner session controls.
• Data Dictionary -- Dialect feature flags, runtime introspection (get_tables, get_columns, get_indexes), ADBC native metadata/statistics.

Query Building & Execution

• Query Builder API -- sql factory: select, insert, update, delete, merge.
• Driver Method Reference -- select(), select_one(), select_stream(), select_to_arrow(), load methods.
• Filter & Pagination System -- LimitOffsetFilter, OrderByFilter, SearchFilter.

Data Integration

• Arrow & ADBC Integration -- select_to_arrow() formats, Arrow-native paths, conversion fallbacks.
• Native Bulk Ingest -- load_from_arrow(), load_from_storage(), load_from_records(), adapter gates.
• SQL File Loading -- SQLFileLoader with search paths, metadata directives.

Adapters & Drivers

• Adapter & Driver Registry -- Full 19-adapter registry with dialects and parameter styles.

Framework & Storage Integrations

• Framework Extensions -- Litestar plugin, FastAPI/Starlette integration.
• Storage Integration -- ADK store, Litestar session stores, event channel backends.
• Event Channels (Pub/Sub) -- AsyncEventChannel, subscribe/publish patterns.
• ADK Extension -- ADK 2 session/memory stores, scoped state, artifact service contracts.

Migrations & Schema

• Native Migration Runner -- standalone sqlspec CLI, timestamp versioning, ddl_migrations tracker, extension migrations, and Litestar litestar db integration.

Observability

• Observability & Tracing -- Telemetry semantics, correlation extraction.

Advanced Patterns

• Design Patterns -- Service layer, batch operations, upsert, AST tenant filters.
• Service Patterns -- SQLSpecAsyncService base, named SQL templates via db_manager.get_sql, direct driver API (select_value / select_one / execute), variadic filter composition, create_filter_dependencies() wiring.
• Dishka Integration -- FromDishka as Inject alias, multi-provider pattern (REQUEST-scoped domain services, REQUEST-scoped driver, APP-scoped singletons), handler injection.
• Vector Search — Oracle VECTOR_DISTANCE cosine similarity, Vertex AI embedding generation, SHA256-keyed embedding cache, intent classification via exemplar similarity, pgvector cross-reference.

Key Resources

• SQLglot Docs: https://sqlglot.com/sqlglot.html
• SQLglot GitHub: https://github.com/tobymao/sqlglot
• Mypyc Docs: https://mypyc.readthedocs.io/
• PyArrow Docs: https://arrow.apache.org/docs/python/

Official References

• https://sqlspec.dev/
• https://sqlspec.dev/changelog.html
• https://github.com/litestar-org/sqlspec

Shared Styleguide Baseline

• Use shared styleguides for generic language/framework rules to reduce duplication in this skill.
• General Principles
• Python
• Litestar
• Keep this skill focused on tool-specific workflows, edge cases, and integration details.