Skip to content

Query Builder

When queries depend on conditional joins, dynamic column lists, or cross-dialect upsert syntax, SQLSpec's fluent sql builder constructs parameterized SQLGlot AST statements that compile cleanly to your active adapter's SQL dialect.

Building SELECT, INSERT, and UPDATE Queries

Import sql from sqlspec to compose SELECT, INSERT, UPDATE, and DELETE queries with automatic parameter binding:

docs/examples/builder/select_query.py
"""Build and execute a parameterized SELECT query with the fluent SQL builder."""

from typing import Any

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

__all__ = ("main",)


def main() -> dict[str, Any]:
    """Query a single team record from an in-memory SQLite database."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    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 ('SQLSpec')")

        query = sql.select("id", "name").from_("teams").where("name = ?")
        row = session.execute(query, "SQLSpec").one()
        assert row == {"id": 1, "name": "SQLSpec"}
        return row


if __name__ == "__main__":
    print(main())
docs/examples/builder/insert_query.py
"""Build and execute a parameterized INSERT statement with the fluent SQL builder."""

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

__all__ = ("main",)


def main() -> int:
    """Insert a row into an in-memory SQLite table and return rows_affected."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with spec.provide_session(config) as session:
        session.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
        query = sql.insert("users").columns("name").values("Ada")
        result = session.execute(query)
        assert result.rows_affected == 1
        return result.rows_affected


if __name__ == "__main__":
    print(main())
docs/examples/builder/update_query.py
"""Build and execute an UPDATE statement with the fluent SQL builder."""

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

__all__ = ("main",)


def main() -> int:
    """Update a row in an in-memory SQLite table and return rows_affected."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with spec.provide_session(config) as session:
        session.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
        session.execute("INSERT INTO users (name) VALUES ('Old')")
        query = sql.update("users").set("name", "New").where("id = 1")
        result = session.execute(query)
        assert result.rows_affected == 1
        return result.rows_affected


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

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

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

with spec.provide_session(config) as session:
    session.execute("CREATE TABLE sessions (id INTEGER PRIMARY KEY, expired INTEGER)")
    session.execute_many("INSERT INTO sessions (expired) VALUES (?)", [(0,), (1,), (1,)])

    delete_expired = sql.delete().from_("sessions").where_eq("expired", 1)
    result = session.execute(delete_expired)
    assert result.rows_affected == 2

Common predicate helpers on sql.select(...), sql.update(...), and sql.delete(...) include:

  • .where(...), .where_eq(col, val), .where_neq(col, val), .where_like(col, pattern)
  • .where_in(col, values), .where_not_in(col, values)
  • .where_is_null(col), .where_is_not_null(col) (note the _is_ verb)
  • .distinct(), .group_by(...), .having(...), .order_by(...), .limit(n), .offset(m)

Joins and Common Table Expressions

Chain .join(...), .left_join(...), .right_join(...), or .cross_join(...) to combine tables with explicit join predicates, and attach .with_(alias, subquery) for readable Common Table Expressions (CTEs):

docs/examples/builder/complex_joins.py
"""Execute a joined SELECT query built with the fluent SQL query builder."""

from typing import Any

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

__all__ = ("main",)


def main() -> list[dict[str, Any]]:
    """Run an inner join between orders and customers in SQLite."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with spec.provide_session(config) as session:
        session.execute("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT)")
        session.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER)")
        session.execute("INSERT INTO customers (name) VALUES ('Ada')")
        session.execute("INSERT INTO orders (customer_id) VALUES (1)")

        query = (
            sql
            .select("orders.id", "customers.name")
            .from_("orders")
            .join("customers", "orders.customer_id = customers.id")
        )
        rows = session.execute(query).all()
        assert rows == [{"id": 1, "name": "Ada"}]
        return rows


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

high_value = sql.select("customer_id", "amount").from_("orders").where("amount >= :min_amount", min_amount=100)
summary = (
    sql
    .select("customer_id", sql.raw("COUNT(*) AS order_count"))
    .with_("high_value", high_value)
    .from_("high_value")
    .group_by("customer_id")
)

Upserts (ON CONFLICT and sql.merge)

Use .on_conflict(...).do_nothing() or .on_conflict(...).do_update(...) for conflict-tolerant inserts on SQLite, PostgreSQL, CockroachDB, and DuckDB, or sql.merge(...) for ANSI MERGE INTO workflows on Oracle, SQL Server, BigQuery, and PostgreSQL 15+:

docs/examples/builder/upsert.py
"""Perform ON CONFLICT DO NOTHING and ON CONFLICT DO UPDATE upserts with the SQL builder."""

from typing import Any

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

__all__ = ("main",)


def main() -> dict[str, Any]:
    """Insert a default setting and upsert a replacement value on conflict."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with spec.provide_session(config) as session:
        session.execute("CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT NOT NULL)")

        insert_ignore = (
            sql.insert("settings").columns("key", "value").values("theme", "dark").on_conflict("key").do_nothing()
        )
        session.execute(insert_ignore)

        upsert = (
            sql
            .insert("settings")
            .columns("key", "value")
            .values("theme", "light")
            .on_conflict("key")
            .do_update(value="light")
        )
        session.execute(upsert)

        row = session.select_one("SELECT value FROM settings WHERE key = 'theme'")
        assert row["value"] == "light"
        return row


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

Raw SQL Fragments and RETURNING

Embed dialect-specific aggregate or window expressions inside builder projections with sql.raw(...), and chain .returning(...) on INSERT, UPDATE, or DELETE builders to fetch mutated columns in a single round-trip:

docs/examples/builder/raw_expressions.py
"""Combine raw SQL expressions and RETURNING clauses inside fluent builder queries."""

from typing import Any

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

__all__ = ("main",)


def main() -> dict[str, Any]:
    """Aggregate event counts with sql.raw() and insert a row using RETURNING."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with spec.provide_session(config) as session:
        session.execute(
            "CREATE TABLE events (id INTEGER PRIMARY KEY, name TEXT, created_at TEXT DEFAULT (datetime('now')))"
        )
        session.execute("INSERT INTO events (name) VALUES ('signup'), ('login')")

        raw_count = sql.raw("COUNT(*) AS total")
        grouped = session.execute(sql.select("name", raw_count).from_("events").group_by("name")).all()
        assert len(grouped) == 2

        insert_returning = sql.insert("events").columns("name").values("logout").returning("id", "name")
        inserted = session.execute(insert_returning).one()
        assert inserted["name"] == "logout"
        return inserted


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

Only pass trusted, static SQL expressions to sql.raw(...); bind user-supplied values via builder parameters.

Dialect Transpilation and Optimization Notes

  • Automatic Dialect Targeting: When executed through session.execute(query), the builder compiles against the session's configured dialect (postgres, sqlite, oracle, bigquery, spanner, spangres, duckdb, db2, tsql, mysql) automatically.
  • Preserving Verbatim Identifier Casing: By default, StatementConfig(enable_optimization=True) runs lightweight SQLGlot qualification and simplification rules. On engines like BigQuery where optimizer qualification lowercases column aliases, pass StatementConfig(dialect="bigquery", enable_optimization=False) if exact input casing must be preserved verbatim.
  • sqlglot[c] Subclassing Compatibility: When sqlspec[mypyc] (sqlglot[c]) is installed, SQLGlot's core AST classes are compiled C extensions; compose queries with sql.* and sql.raw(...) rather than subclassing sqlglot.exp nodes at runtime.

Recap

  • Reach for sql.select, sql.insert, sql.update, sql.delete, and sql.merge when predicates, projections, or conflict clauses vary dynamically at runtime.
  • Use .where_in(...) and .where_is_null(...) for safe collection and null checks without manual placeholder counting.
  • Combine sql.raw(...) with .returning(...) when mixing custom SQL expressions into structured DML builders.