SQL Files¶

Keeping non-trivial queries in dedicated .sql files gives you full editor syntax highlighting, formatter support, and DBA reviewability while retaining SQLSpec's parameter binding, runtime type validation, and AST composition. SQLFileLoader and SQLSpec parse structured comment directives and inline markers without custom template engines.

Directive and Marker Cheat Sheet¶

Directive or Marker	Scope	Syntax	Purpose
`-- name:`	Statement header	`-- name: list-users`	Starts a named query block; hyphens normalize to underscores (`list_users`) on lookup.
`-- dialect:`	Statement header	`-- dialect: postgres`	Sets the SQLGlot dialect for expression slots, `.builder()`, and `.explain()`.
`-- param:`	Statement header	`-- param: status str? Optional filter`	Declares parameter name, runtime type (`?` = optional, defaults to `None`), and description.
`-- fragment:`	File section	`-- fragment: active_users_cte`	Declares a reusable static SQL fragment (cannot declare `-- dialect:`, `-- param:`, or `-- slot:`).
`/* include: */`	Statement or fragment body	`/* include: active_users_cte */`	Splices a registered fragment (supports nested includes and `<namespace>.<name>` lookup).
`-- slot:`	Statement header	`-- slot: ordering = id ASC`	Declares an optional default SQL fill for a matching `/* slot: name */` marker in the body.
`/* slot: */`	Statement or fragment body	`/* slot: ordering */`	Reserves a fill point replaced at `spec.get_sql(name, **slots)` retrieval time.

Loading SQL Files and Directory Namespaces¶

Load .sql files or directory trees with spec.load_sql_files(...) (or register inline strings via spec.add_named_sql(...)). Relative file and directory paths, including bare names such as "queries.sql", resolve against the current working directory:

queries.sqlapp.pyInline Registration

-- name: list_teams
SELECT id, name
FROM teams
ORDER BY id;

spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
spec.load_sql_files(sql_file)

with spec.provide_session(config) as session:
    session.execute("CREATE TABLE teams (id INTEGER PRIMARY KEY, name TEXT)")
    session.execute("INSERT INTO teams (name) VALUES ('Litestar'), ('SQLSpec')")
    rows = session.select(spec.get_sql("list_teams"))

spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
spec.add_named_sql(
    "find_team",
    """
    SELECT id, name
    FROM teams
    WHERE name = :name
    """,
)

with spec.provide_session(config) as session:
    session.execute("CREATE TABLE teams (id INTEGER PRIMARY KEY, name TEXT)")
    session.execute("INSERT INTO teams (name) VALUES ('Litestar'), ('SQLSpec')")
    row = session.select_one(spec.get_sql("find_team"), name="SQLSpec")

When spec.load_sql_files(Path("app/sql")) scans a directory tree, top-level files register bare query names while nested subdirectories prepend dot-separated relative directory namespaces:

app/sql/
├── users.sql            # -- name: list-users   -> "list_users"
└── billing/
    ├── invoices.sql     # -- name: list-overdue -> "billing.list_overdue"
    └── reports/
        └── mrr.sql      # -- name: monthly-mrr  -> "billing.reports.monthly_mrr"

spec.list_sql_queries() / spec.has_sql_query(name) / spec.get_sql_files(): Inspect registered query names and file paths at startup.
spec.reload_sql_files(reload=True): Re-reads only .sql files whose MD5 checksum changed and invalidates cached statement instances.
SQLFileLoader(storage_registry=...): Loads .sql bundles from S3, GCS, Azure Blob Storage, or fsspec aliases registered in StorageRegistry.

Dialect Declarations¶

Add -- dialect: <name> under -- name: so SQLGlot expression slots, SQL.builder(), and SQL.explain() target the exact engine syntax:

-- name: upsert_subscription
-- dialect: postgres
INSERT INTO subscriptions (account_id, plan)
VALUES (:account_id, :plan)
ON CONFLICT (account_id) DO UPDATE SET plan = EXCLUDED.plan;

The loader normalizes these aliases (DIALECT_ALIASES) before handing the name to SQLGlot; any other value passes through unchanged:

Input Alias	Normalized Dialect
`postgresql`, `pg`, `pgplsql`	`postgres`
`plsql`, `oracledb`	`oracle`
`tsql`	`mssql`

Parameter Type Declarations¶

Declare -- param: <name> <type>[?] <description> headers below -- name: to document parameters and validate them:

teams.sqlapp.py

-- name: get_team_by_name
-- param: name str  The team name to look up
SELECT id, name FROM teams WHERE name = :name;

-- name: list_teams
-- param: name str?  Optional team name filter
SELECT id, name FROM teams
WHERE (:name IS NULL OR name = :name)
ORDER BY id;

spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
spec.load_sql_files(sql_file)

declarations = spec.get_query_parameters("get_team_by_name")
query = spec.get_sql("get_team_by_name")

with spec.provide_session(config) as session:
    session.execute("CREATE TABLE teams (id INTEGER PRIMARY KEY, name TEXT)")
    session.execute("INSERT INTO teams (name) VALUES ('Litestar'), ('SQLSpec')")

    row = session.select_one(query, name="SQLSpec")
    optional_rows = session.select(spec.get_sql("list_teams"))

Placeholder drift check at load time: The loader raises SQLFileParseError when a declared -- param: name is missing from the query's :placeholder tokens or positional counts differ. Queries that use fragments or slots are checked at get_sql() time instead.
Required vs optional at execution: Omitting a required parameter raises SQLSpecError before a cursor is acquired. A ? type suffix (or a description ending in (optional)) defaults an omitted keyword argument to None, so (:name IS NULL OR name = :name) or COALESCE(:name, name) predicates reuse one prepared-statement plan.
Runtime type checks: Non-None values are checked against the declared type: str, int, float, bool, bytes, date, datetime, time, decimal, uuid, dict, json, jsonb, list, list[int], list[str], list[float], list[bool], or tuple. Unknown type strings are documentation only unless registered with register_param_type("money", Decimal). Pass SQLFileLoader(strict_parameter_annotations=True) to raise SQLFileParseError on malformed -- param: lines.

Fragments and Slots¶

Reuse static SQL blocks across queries with -- fragment: <name> and /* include: <name> */, and fill controlled dynamic SQL markers at retrieval time with -- slot: <name> = <default> and /* slot: <name> */:

teams.sqlapp.py

-- fragment: team_source
WITH team_source AS (
    SELECT id, name
    FROM teams
    WHERE workspace_id = :workspace_id
)

-- name: list_teams
-- slot: predicates = 1 = 1
-- slot: ordering = t.id
/* include: team_source */
SELECT t.id, t.name
FROM team_source AS t
WHERE /* slot: predicates */
ORDER BY /* slot: ordering */;

-- name: count_teams
/* include: team_source */
SELECT COUNT(*)
FROM team_source AS t
WHERE /* slot: predicates */

sql_file = Path(__file__).with_suffix(".sql")
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
spec.load_sql_files(sql_file)
try:
    with spec.provide_session(config) as session:
        session.execute("CREATE TABLE teams (id INTEGER PRIMARY KEY, name TEXT, workspace_id INTEGER)")
        session.execute_many(
            "INSERT INTO teams VALUES (:id, :name, :workspace_id)",
            [
                {"id": 1, "name": "Litestar", "workspace_id": 42},
                {"id": 2, "name": "SQLSpec", "workspace_id": 42},
                {"id": 3, "name": "Other", "workspace_id": 99},
            ],
        )
        teams = session.select(spec.get_sql("list_teams"), workspace_id=42)
        ordered = session.select(spec.get_sql("list_teams", ordering="t.name DESC"), workspace_id=42)

        predicate = SQL("t.name = :team_name", team_name="SQLSpec")
        query = spec.get_sql("list_teams", predicates=predicate, ordering="t.name DESC")
        selected = session.select(query, workspace_id=42)
        count = session.select_value(spec.get_sql("count_teams", predicates=predicate), workspace_id=42)
finally:
    config.close_pool()

Value Type	Example	Splicing Semantics	Parameter Binding
`str`	`ordering="t.name DESC"`	Spliced verbatim as trusted SQL text (use allowlists for identifiers and directions).	None (bind user data separately).
`sqlglot.exp.Expr`	`ordering=sql.col("name", table="t").desc()`	Rendered to SQL in the query's `-- dialect:`.	None.
`SQL`	`predicates=SQL("t.name = :team_name", team_name="SQLSpec")`	Splices `SQL.sql` into the marker and merges `SQL.named_parameters` onto the returned statement.	Carries named parameters (`:team_name`) safely into execution.

Slot Validation and SQLSlotError Rules

Missing or unknown slots: A /* slot: name */ marker without a -- slot: name = default header or get_sql() keyword argument, or an unknown slot keyword, raises SQLSlotError.
Positional binds and filters: Passing a SQL slot value with positional parameters or attached StatementFilter objects raises SQLSlotError (attach filters to the completed SQL returned by get_sql()).
Parameter collisions: Overlapping parameter names between a SQL slot and the outer query (or between two slots with unequal values) raise SQLSlotError.
Fragment headers and cycles: Fragments cannot declare -- dialect:, -- param:, or -- slot: headers (SQLFileParseError), and cyclic /* include: */ chains raise SQLFileParseError.

Statement Modifiers and the Builder Bridge¶

spec.get_sql(...) returns an immutable SQL instance that supports runtime predicate chaining (.where_eq(), .where_in(), .where_like(), .order_by(), .paginate()), column projection narrowing (.select_only(*cols, prune_columns=True)), execution plan inspection (.explain(analyze=True, format="json")), and conversion into a fluent SQLGlot AST builder via .builder():

Dynamic ModifiersSchema Projection & Builder Bridge

spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
spec.add_named_sql("list_users", "SELECT id, name, status FROM users")

with spec.provide_session(config) as session:
    session.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, status TEXT)")
    session.execute(
        "INSERT INTO users (name, status) VALUES ('Alice', 'active'), ('Bob', 'inactive'), ('Charlie', 'active')"
    )

    base_query = spec.get_sql("list_users")
    active_users = session.select(base_query.where_eq("status", "active").order_by("name"))
    bob = session.select(base_query.where_in("name", ["Bob"]).select_only("id", "name"))
    alice_only = session.select(base_query.where_eq("status", "active").where_like("name", "A%"))

from pathlib import Path
import msgspec

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig


class Product(msgspec.Struct):
    id: int
    sku: str
    name: str
    price_cents: int

spec = SQLSpec()
config = spec.add_config(
    SqliteConfig(connection_config={"database": ":memory:"})
)
spec.load_sql_files(Path("sql"))
with spec.provide_session(config) as session:
    product = session.select_one(
        spec.get_sql("get-product"), id=1, schema_type=Product
    )
    catalog = session.select(
        spec.get_sql("list-products")
        .where("price_cents >= :min_price")
        .select_only("id", "sku", "name", "price_cents", prune_columns=True),
        min_price=1000,
        sku=None,
        schema_type=Product,
    )
    plan = session.select(spec.get_sql("list-products").explain(), sku=None)
    ranked_query = (
        spec.get_sql("list-products")
        .builder()
        .join("inventory", on="inventory.product_id = products.id")
        .build()
    )

Query Composition Decision Matrix¶

Mechanism	Phase	Mutates SQL Structure?	Parameter-Safe?	Best For
`-- param:` (`:name`, `str?`)	Execution bind	No (fixed SQL plan)	Yes (driver bind)	Scalar filters and optional `(:p IS NULL OR col = :p)` predicates that maximize prepared-statement cache hits.
`-- fragment:` + `/* include: */`	Load / `get_sql()`	Static splice	N/A (template text)	Sharing CTEs, `FROM` / `JOIN` blocks, or column lists across queries in `.sql` files.
`-- slot:` + `/* slot: */`	`get_sql(**slots)`	Controlled marker fill	Yes when filled with `SQL(..., key=val)`	Swapping trusted `ORDER BY` expressions, table/view names, or optional multi-clause predicates at retrieval time.
`SQL.where*()` / `StatementFilter`	Pre-execution AST	Appends `WHERE` / `ORDER BY` / `LIMIT`	Yes (auto-named binds)	Attaching HTTP query filters (`SearchFilter`, `InCollectionFilter`, `CursorFilter`) or `.select_only()` onto a base query.
`sql.select()` / `SQL.builder()`	Runtime AST builder	Full AST construction	Yes (auto-named binds)	Dynamic multi-join queries, conditional `GROUP BY` or window functions, cross-dialect vector distance, and programmatic DDL/DML.

GeezerVison