SQL Files¶

Keeping non-trivial queries in dedicated .sql files gives you full editor syntax highlighting, formatter support, DBA reviewability, and zero Python string-escaping noise while retaining SQLSpec's parameter binding, validation, and dynamic predicate composition.

Inline Named Queries¶

For small query registries, register named statements directly on SQLSpec with add_named_sql and retrieve a reusable SQL object via get_sql:

docs/examples/sql_files/named_queries.py

"""Register inline named SQL statements and execute them with bound keyword parameters."""

from typing import Any

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

__all__ = ("main",)


def main() -> dict[str, Any]:
    """Register find_team on SQLSpec and execute it by name."""
    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.execute(spec.get_sql("find_team"), name="SQLSpec").one()
        assert row == {"id": 2, "name": "SQLSpec"}
        return row


if __name__ == "__main__":
    print(main())

Loading `.sql` Files from Disk or Storage¶

Group related queries inside .sql files using -- name: <query_name> comment headers and load individual files or entire directories with spec.load_sql_files(...) (or SQLFileLoader directly):

docs/examples/sql_files/load_sql_files.py

"""Load named SQL queries from disk with SQLSpec.load_sql_files."""

import tempfile
from pathlib import Path
from typing import Any

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

__all__ = ("main",)


def main() -> list[dict[str, Any]]:
    """Write a named SQL file, load it into SQLSpec, and execute the named query."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with tempfile.TemporaryDirectory() as tmp_dir:
        sql_file = Path(tmp_dir) / "queries.sql"
        sql_file.write_text("-- name: list_teams\nSELECT id, name FROM teams ORDER BY id;\n", encoding="utf-8")
        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.execute(spec.get_sql("list_teams")).all()
        assert rows == [{"id": 1, "name": "Litestar"}, {"id": 2, "name": "SQLSpec"}]
        return rows


if __name__ == "__main__":
    print(main())

Query names inside nested directories can be organized by namespace (or hyphenated identifiers normalized automatically when accessed) and shared across any session opened from spec. SQLFileLoader also integrates with StorageRegistry so SQL bundles can be loaded from local directories or registered object storage aliases.

Anatomy of a Multi-Query `.sql` File¶

A single .sql file can group all reads and writes for a domain module under distinct -- name: headers, with optional dialect annotations and -- param: declarations:

-- name: create_invoices_table
CREATE TABLE IF NOT EXISTS invoices (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    status TEXT NOT NULL,
    total_cents INTEGER NOT NULL
);

-- name: insert_invoice
-- param: customer_id int  Account identifier
-- param: status str       Initial invoice state
-- param: total_cents int  Invoice total in minor currency units
INSERT INTO invoices (customer_id, status, total_cents)
VALUES (:customer_id, :status, :total_cents);

-- name: list_customer_invoices
-- param: customer_id int  Account identifier
-- param: status str?      Optional status filter
SELECT id, customer_id, status, total_cents
FROM invoices
WHERE customer_id = :customer_id
  AND (:status IS NULL OR status = :status)
ORDER BY id DESC;

Organizing Query Directories¶

When spec.load_sql_files(Path("app/sql")) scans a directory tree, you can keep queries organized per domain (app/sql/billing.sql, app/sql/users.sql) and inspect loaded definitions at startup:

spec.list_sql_queries(): Returns all registered query names so startup tests can assert expected SQL assets are present.
spec.get_sql("list_customer_invoices"): Returns the immutable SQL template ready for execution or filter composition.
SQLFileLoader(storage_registry=...): Loads .sql files from S3, GCS, Azure Blob Storage, or embedded package paths registered in StorageRegistry.

Declared Parameter Headers (`-- param:`)¶

Add -- param: <name> <type>[?] <description> headers immediately below -- name: to document expected arguments and enforce automatic required-parameter validation before a query ever touches the database connection:

docs/examples/sql_files/declared_params.py

"""Validate required and optional named parameters declared in SQL file headers."""

import tempfile
from pathlib import Path
from typing import Any

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

__all__ = ("main",)

TEAMS_SQL = (
    "-- name: get_team_by_name\n"
    "-- param: name str  The team name to look up\n"
    "SELECT id, name FROM teams WHERE name = :name;\n"
    "\n"
    "-- name: list_teams\n"
    "-- param: name str?  Optional team name filter\n"
    "SELECT id, name FROM teams WHERE (:name IS NULL OR name = :name) ORDER BY id;\n"
)


def main() -> tuple[dict[str, Any], list[dict[str, Any]]]:
    """Load declared parameter metadata from a SQL file and verify automatic validation."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with tempfile.TemporaryDirectory() as tmp_dir:
        sql_file = Path(tmp_dir) / "teams.sql"
        sql_file.write_text(TEAMS_SQL, encoding="utf-8")
        spec.load_sql_files(sql_file)

    declarations = spec.get_query_parameters("get_team_by_name")
    assert declarations[0].name == "name"
    assert declarations[0].type_str == "str"
    assert declarations[0].description == "The team name to look up"

    query = spec.get_sql("get_team_by_name")
    assert query.declared_parameters == declarations

    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.execute(query, name="SQLSpec").one()
        optional_rows = session.execute(spec.get_sql("list_teams")).all()

        missing_error = ""
        try:
            session.execute(spec.get_sql("get_team_by_name"))
        except SQLSpecError as exc:
            missing_error = str(exc)

    assert row["name"] == "SQLSpec"
    assert [team["name"] for team in optional_rows] == ["Litestar", "SQLSpec"]
    assert "name" in missing_error
    return row, optional_rows


if __name__ == "__main__":
    print(main())

Suffixing the type with ? (such as str? or int?) marks the parameter optional and defaults omitted keyword arguments to None (NULL), enabling clean (:name IS NULL OR name = :name) filter patterns in pure SQL. Omitting a required parameter raises SQLSpecError immediately with the missing parameter name.

Composing Dynamic `WHERE` Clauses onto Named SQL¶

Static .sql files often need optional runtime predicates for search, tenant scoping, or pagination. Because spec.get_sql(...) returns an immutable SQL statement object, calling .where(...), .where_eq(...), .where_like(...), or .paginate(...) returns a new derived SQL instance without mutating the cached base template:

docs/examples/sql_files/dynamic_where.py

"""Append dynamic WHERE predicates onto a named base SQL statement."""

from typing import Any

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

__all__ = ("main",)


def main() -> list[dict[str, Any]]:
    """Chain where, where_eq, and where_like filters onto a named SQL query."""
    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("status = 'active'"))
        bob = session.select(base_query.where_eq("name", "Bob"))
        alice_only = session.select(base_query.where_eq("status", "active").where_like("name", "A%"))

    assert len(active_users) == 2
    assert len(bob) == 1
    assert alice_only == [{"id": 1, "name": "Alice", "status": "active"}]
    return alice_only


if __name__ == "__main__":
    print(main())

You can also pass StatementFilter objects (LimitOffsetFilter, SearchFilter, OrderByFilter) alongside spec.get_sql("...") inside session.select(...)—remembering to keep all bound parameters named when filters are attached.

Recap¶

Store multi-line CTEs, window functions, and analytical queries in .sql files under -- name: blocks and load them once at startup via spec.load_sql_files(...).
Declare -- param: headers on shared SQL queries to catch missing required arguments early and expose parameter metadata via spec.get_query_parameters(...).
Chain .where_eq(...), .where_like(...), or StatementFilter arguments onto spec.get_sql(...) instead of concatenating raw WHERE strings in Python.

SQL Files¶

Inline Named Queries¶

Loading .sql Files from Disk or Storage¶

Anatomy of a Multi-Query .sql File¶