SQLSpec Skill

SQLSpec is a type-safe SQL query mapper for Python -- NOT an ORM. It provides flexible connectivity with consistent interfaces across 15+ database adapters. Write raw SQL, use the builder API, or load SQL from files. All statements pass through a sqlglot-powered AST pipeline for validation and dialect conversion.

Match-Your-Framework — read first

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

Litestar —
```
SQLSpecPlugin
```
with full DI, CLI, observability. The rest of this SKILL.md covers Litestar by default; also 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.

sqlspec has no first-party Sanic integration — other frameworks are not supported out-of-the-box.

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

python

from sqlspec.adapters.asyncpg import AsyncpgConfig, AsyncpgDriver

# Configure the adapter with connection details
config = AsyncpgConfig(
    connection_config={
        "dsn": "postgresql://user:pass@localhost:5432/mydb",
        "min_size": 2,
        "max_size": 10,
    },
)

# Use the driver as a context manager for connection lifecycle
async with config.create_driver() as db:
    users = await db.select_many(
        "SELECT * FROM users WHERE active = $1",
        [True],
        schema_type=User,
    )

Query Builder Essentials

python

from sqlspec import sql

# SELECT with filters
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_into("users")
    .columns("name", "email")
    .values(name="Alice", email="alice@example.com")
    .to_statement()
)

# MERGE / upsert
stmt = (
    sql.merge_("inventory")
    .using("updates", on="inventory.product_id = updates.product_id")
    .when_matched().do_update(qty="updates.qty")
    .when_not_matched().do_insert(product_id="updates.product_id", qty="updates.qty")
    .to_statement()
)

Driver Method Summary

Method	Returns	Use Case
`select_value()`	Single scalar	`COUNT(*)` , `MAX()` , existence checks
`select_one()`	One row (strict)	Get-by-ID, raises `NotFoundError`
`select_one_or_none()`	One row or `None`	Optional lookup
`select_many()`	List of rows	Filtered queries, listing
`select_to_arrow()`	`pyarrow.Table`	Bulk data export, analytics
`execute()`	Row count	INSERT/UPDATE/DELETE
`execute_many()`	Row count	Batch operations

Arrow Integration Basics

python

# Zero-copy on DuckDB, ADBC adapters; conversion on others
arrow_table = await db.select_to_arrow(
    "SELECT * FROM large_dataset WHERE region = $1", [region]
)

# Bulk load from Arrow
await db.copy_from_arrow(arrow_table, target_table="users")

Workflow

Step 1: Choose Adapter and Pattern

Need	Adapter	Key Feature
PostgreSQL async	`asyncpg` , `psycopg`	Async, NUMERIC/PYFORMAT params
PostgreSQL sync	`psycopg`	Sync+async, PYFORMAT params
SQLite	`sqlite` , `aiosqlite`	QMARK params, local dev
DuckDB analytics	`duckdb`	Arrow-native, zero-copy
MySQL async	`asyncmy`	PYFORMAT params
Oracle	`oracledb`	NAMED_COLON params, sync+async
BigQuery / Spanner	`bigquery` , `spanner`	NAMED_AT params
Raw SQL strings	Driver methods	`select_many()` , `execute()`
Dynamic queries	Query builder	`sql.select()...to_statement()`
SQL from files	`SQLFileLoader`	Metadata directives, caching

Step 2: Implement

Configure the adapter with connection details and pool settings
Use
```
create_driver()
```
context manager for connection lifecycle
Choose the appropriate driver method for your query shape
Use
```
schema_type
```
parameter for typed results (Pydantic or msgspec models)

Apply filters with

LimitOffsetFilter

OrderByFilter

SearchFilter

Step 3: Validate

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

</workflow> <guardrails>

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 config.create_driver() 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, enables caching
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
Adapter config / driver modules avoid
from __future__ import annotations
. Consumer app modules MAY use it.

</guardrails> <validation>

Validation Checkpoint

Before delivering SQLSpec code, verify:

Adapter config uses the correct import path (
```
sqlspec.adapters.<name>
```
)
Connection lifecycle uses
```
create_driver()
```
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

</validation> <example>

Example

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

python

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


# --- Typed model ---

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


# --- Adapter setup ---

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


# --- Query execution ---

async def list_active_users(page: int = 1, page_size: int = 25) -> list[User]:
    filters = [
        OrderByFilter(columns=[("name", "asc")]),
        LimitOffsetFilter(limit=page_size, offset=(page - 1) * page_size),
    ]

    async with config.create_driver() as db:
        users = await db.select_many(
            "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 config.create_driver() as db:
        count = await db.select_value(
            "SELECT COUNT(*) FROM users WHERE active = $1", [True]
        )
        return count

</example>

Query Builder

The

sql

factory provides a fluent builder API with full method chaining. All builders terminate with

.to_statement()

and pass through sqlglot for validation and dialect conversion.

Builder	Entry Point	Key Methods
SELECT	`sql.select(*cols)`	`.from_()` , `.where()` , `.where_eq()` , `.join()` , `.order_by()` , `.limit()` , `.offset()`
INSERT	`sql.insert_into(table)`	`.columns()` , `.values()` , `.returning()`
UPDATE	`sql.update(table)`	`.set_()` , `.where()` , `.returning()`
DELETE	`sql.delete_from(table)`	`.where()` , `.returning()`
MERGE	`sql.merge_(target)`	`.using()` , `.when_matched()` , `.when_not_matched()`
CREATE TABLE	`sql.create_table(name)`	`.column()` , `.primary_key()` , `.if_not_exists()`
DROP TABLE	`sql.drop_table(name)`	`.if_exists()` , `.cascade()`

ArrowResult

select_to_arrow()

returns an Apache Arrow

Table

for bulk and analytical workloads:

Zero-copy on DuckDB and ADBC-native adapters — no serialization overhead
Conversion path on other adapters — rows are materialized into an Arrow schema
Returned tables are compatible with Polars, Pandas, and PyArrow directly
Use
```
copy_from_arrow(table, target_table)
```
for bulk loads back into the database

Filters

SQLSpec filter objects are passed directly to driver methods alongside the SQL string. They modify the statement before execution.

Filter	Purpose	Example Use
`BeforeAfterFilter`	Date range bounds ( `before` , `after` )	Audit log queries, time-range pagination
`InCollectionFilter`	SQL `IN (...)` clause	Filter by a set of IDs or enum values
`LimitOffsetFilter`	Page-based pagination	`limit=25, offset=50`
`OrderByFilter`	Dynamic sort columns and direction	User-supplied sort fields
`SearchFilter`	Text search ( `ILIKE` / `LIKE` )	Full-text style search on string columns

Filters are composable — pass multiple to a single

select_many()

call and they are applied in order.

Framework Integrations

Framework	Integration	Key Feature
Litestar	`SQLSpecPlugin`	Dependency injection of typed driver; auto session lifecycle
FastAPI / Starlette	Middleware	Request-scoped connection; injects driver into route dependencies
Flask	Extension	`init_app()` pattern; driver available via `g` or `current_app`

SQLSpecPlugin

for Litestar registers the driver as a DI provider — inject it into route handlers via type annotation without manual context management.

Event Channels

For databases that support server-side pub/sub (e.g., PostgreSQL

LISTEN

NOTIFY

Use
```
AsyncEventChannel
```
to subscribe to named channels
Publish with
```
NOTIFY channel, payload
```
from SQL or from the
```
publish()
```
method
Handlers receive
```
EventMessage
```
objects with channel name, payload, and PID
Useful for real-time cache invalidation, cross-process coordination, and background job triggers

Key Design Principles

Single Source of Truth: The
```
SQL
```
object holds all state for a given statement
Immutability: All operations on a
```
SQL
```
object return new instances
Type Safety: Parameters carry type information through the processing pipeline
Protocol-Based Design: Uses Python protocols for runtime type checking instead of inheritance
Single-Pass Processing: Parse once, transform once, validate once

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, 15+ driver adapters (asyncpg, oracledb, DuckDB, BigQuery, SQLite, and more), Arrow-native result streams 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, NamespacedCache system, Mypyc compilation.
Data Dictionary -- Dialect feature flags, runtime introspection (
```
get_tables
```
,
```
get_columns
```
,
```
get_indexes
```
), driver-side metadata API.

Query Building & Execution

Query Builder API --
```
sql
```
factory: select, insert, update, delete, merge.

Driver Method Reference --

select_value()

select_one()

select_many()

select_to_arrow()

Filter & Pagination System --

LimitOffsetFilter

OrderByFilter

SearchFilter

Data Integration

Arrow & ADBC Integration --
```
select_to_arrow()
```
zero-copy,
```
copy_from_arrow()
```
bulk loading.
SQL File Loading --
```
SQLFileLoader
```
with search paths, metadata directives.

Adapters & Drivers

Adapter & Driver Registry -- Full 15-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 --

SQLSpecSessionService

SQLSpecMemoryService

SQLSpecArtifactService

, per-adapter ADK stores.

Migrations & Schema

Native Migration Runner --
```
sqlspec database
```
CLI, timestamp versioning,
```
ddl_migrations
```
tracker, extension migrations, 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

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.

sqlspec

NPX Install

Tags

SKILL.md Content

SQLSpec Skill

Match-Your-Framework — read first

Code Style Rules

Quick Reference

Adapter Pattern

Query Builder Essentials

Driver Method Summary

Arrow Integration Basics

Workflow

Step 1: Choose Adapter and Pattern

Step 2: Implement

Step 3: Validate

Guardrails

Validation Checkpoint

Example

Query Builder

ArrowResult

Filters

Framework Integrations

Event Channels

Key Design Principles

References Index

Standards & Style

Core Utilities

Architecture & Performance

Query Building & Execution

Data Integration

Adapters & Drivers

Framework & Storage Integrations

Migrations & Schema

Observability

Advanced Patterns

Key Resources

Official References

Shared Styleguide Baseline