Query Builder#

SQLSpec includes a fluent query builder for teams who prefer structured SQL construction. The builder outputs SQL objects that can be executed with the same driver APIs.

Selects#

select query#

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

db_path = tmp_path / "builder.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

with spec.provide_session(config) as session:
    session.execute("create table if not exists teams (id integer primary key, name text)")
    session.execute("insert into teams (name) values ('SQLSpec')")

    query = sql.select("id", "name").from_("teams").where("name = ?")
    result = session.execute(query, "SQLSpec")
    print(result.one())

NULL placement#

By default, the database decides where NULL rows go. Choose their place with Column.asc(nulls="first") or Column.desc(nulls="last"), or use a string item. SQLSpec adds a NULL clause or a CASE sort key, as needed for the database.

from sqlspec import sql
from sqlspec.builder import Column

query = sql.select("id").from_("users").order_by(Column("id").desc(nulls="last"))
same_order = sql.select("id").from_("users").order_by("id DESC NULLS LAST")

Inserts and Updates#

insert query#

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

db_path = tmp_path / "builder_insert.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

with spec.provide_session(config) as session:
    session.execute("create table if not exists users (id integer primary key, name text)")
    query = sql.insert("users").columns("name").values("Ada")
    result = session.execute(query)
    print(result.rows_affected)

update query#

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

db_path = tmp_path / "builder_update.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

with spec.provide_session(config) as session:
    session.execute("create table if not exists 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)
    print(result.rows_affected)

UPDATE ... FROM and CTEs#

Builder queries support UPDATE ... FROM with subqueries or Common Table Expressions (CTEs), enabling queue claim statements and batch updates across supported dialects.

from sqlspec import sql

claim_query = (
    sql.update("tasks")
    .set(status="processing")
    .from_(
        sql.select("id")
        .from_("tasks")
        .where_eq("status", "pending")
        .limit(1)
        .for_update(skip_locked=True),
        alias="sub",
    )
    .where("tasks.id = sub.id")
    .returning(sql.column("id", table="tasks"))
)

Dialect Support Matrix#

Dialect	UPDATE ... FROM Support	Notes
PostgreSQL	Yes	Native `UPDATE ... FROM` with `RETURNING` and row locking
CockroachDB	Yes	Native `UPDATE ... FROM`
SQLite	Yes	Native `UPDATE ... FROM` (SQLite 3.33.0+)
DuckDB	Yes	Native `UPDATE ... FROM`
SQL Server (MSSQL)	Yes	Native `UPDATE ... FROM`
MySQL / MariaDB	No	Raises `SQLBuilderError`; use multi-table join update or MERGE
Oracle	No	Builder conservatively raises `SQLBuilderError`; use MERGE or raw SQL on Oracle 23+
Spanner	No	Raises `SQLBuilderError`
BigQuery	Yes	Native `UPDATE ... FROM`; a `WHERE` condition is required

Upserts (ON CONFLICT)#

Use .on_conflict() to handle insert conflicts. Chain .do_nothing() to skip conflicting rows, or .do_update(**columns) to update them.

Dialects natively supporting ON CONFLICT (PostgreSQL, CockroachDB, SQLite, DuckDB, and Spanner) render standard ON CONFLICT syntax. For MySQL and MariaDB, the builder automatically transpiles .on_conflict().do_update() to ON DUPLICATE KEY UPDATE, and .do_nothing() to a no-op self-assignment (e.g., col = col). This requires a conflict column or explicit insert columns. MySQL handles conflicts on any unique key, regardless of the requested conflict target; the no-op update can still fire update triggers. References to excluded.column in update expressions become VALUES(column). Dialects without native upsert clauses (Oracle, T-SQL / SQL Server, and BigQuery) raise SQLBuilderError in both build() and to_statement() advising the use of sql.merge().

upsert with on_conflict#

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

db_path = tmp_path / "upsert.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

with spec.provide_session(config) as session:
    session.execute("create table settings (  key text primary key,  value text not null)")

    # ON CONFLICT DO NOTHING - skip if key exists
    insert_ignore = (
        sql.insert("settings").columns("key", "value").values("theme", "dark").on_conflict("key").do_nothing()
    )
    session.execute(insert_ignore)

    # ON CONFLICT DO UPDATE - upsert pattern
    upsert = (
        sql
        .insert("settings")
        .columns("key", "value")
        .values("theme", "light")
        .on_conflict("key")
        .do_update(value="light")
    )
    session.execute(upsert)

    result = session.select_one("select value from settings where key = 'theme'")
    print(result)  # {"value": "light"}

Raw Expressions and RETURNING#

Use sql.raw() to embed raw SQL fragments (like database functions) inside builder queries. Use .returning() on INSERT, UPDATE, or DELETE to get back the affected rows.

raw expressions and returning#

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

db_path = tmp_path / "raw.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

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

    # sql.raw() creates a raw SQL expression for use inside builders
    raw_count = sql.raw("COUNT(*)")
    query = sql.select("name", raw_count).from_("events").group_by("name")
    result = session.execute(query)
    print(result.all())

    # Use RETURNING clause with INSERT
    insert_returning = sql.insert("events").columns("name").values("logout").returning("id", "name")
    new_row = session.execute(insert_returning)
    print(new_row.one())  # {"id": 3, "name": "logout"}

Joins#

join query#

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

db_path = tmp_path / "builder_joins.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

with spec.provide_session(config) as session:
    session.execute("create table if not exists customers (id integer primary key, name text)")
    session.execute("create table if not exists orders (id integer primary key, customer_id int)")
    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")
    )
    result = session.execute(query)
    print(result.all())

Query Modifiers#

Row-level locking clauses such as .for_update() and .for_share() are validated against dialect capabilities at build time. On dialects without these locking clauses (T-SQL, SQLite, DuckDB, and BigQuery), building a locked query raises SQLBuilderError. Oracle also rejects .for_share(); MariaDB renders it as LOCK IN SHARE MODE and rejects of= targets for all locking clauses. PostgreSQL key lock variants are rejected on other dialect families. Spanner supports plain FOR UPDATE in both SQL modes, but rejects shared locks, SKIP LOCKED, NOWAIT, and OF modifiers. Its PostgreSQL mode requires conflict updates to assign every inserted column from the matching excluded column and does not accept conflict predicates or named constraints. Similarly, skip_locked=True is validated against the dialect's supports_skip_locked capability. The builder also normalizes common dialect aliases during build (e.g., mssql to tsql, mariadb to mysql, and cockroachdb to postgres).

ordering, pagination, and row-level locking#

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

db_path = tmp_path / "modifiers.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

with spec.provide_session(config) as session:
    session.execute("create table if not exists users (id integer primary key, name text, status text)")
    session.execute(
        "insert into users (name, status) values ('Ada', 'active'), ('Bob', 'inactive'), ('Charlie', 'active')"
    )

    query = (
        sql.select("id", "name").from_("users").where_eq("status", "active").order_by("name").limit(10).offset(0)
    )
    result = session.execute(query)
    print(result.all())

    pg_query = (
        sql
        .select("id", "name", dialect="postgres")
        .from_("users")
        .where_eq("status", "active")
        .for_update(skip_locked=True)
    )
    pg_sql = pg_query.to_sql()

Deletes#

Construct DELETE statements with target tables, conditions, and optional RETURNING clauses:

from sqlspec import sql

# DELETE FROM users WHERE status = 'inactive'
delete_query = sql.delete("users").where_eq("status", "inactive")

# DELETE with RETURNING on supported dialects
delete_returning = sql.delete("tasks").where_eq("completed", True).returning("id", "title")

Dynamic Updates with Model Dumps#

The update builder's .set_from() method accepts dataclasses, msgspec Structs, Pydantic models, or dictionaries, automatically mapping fields to column assignments:

from dataclasses import dataclass
from sqlspec import sql

@dataclass
class UserProfile:
    name: str
    email: str

profile = UserProfile(name="Ada Lovelace", email="[email protected]")

# UPDATE users SET name = :name, email = :email WHERE id = :id
query = sql.update("users").set_from(profile).where_eq("id", 1)

Merge Statements#

For dialects that do not natively support ON CONFLICT (such as Oracle, T-SQL, and BigQuery), or for complex conditional matching, use sql.merge():

from sqlspec import sql

# MERGE INTO target USING source ON target.id = source.id
merge_query = (
    sql.merge("target_table", dialect="postgres")
    .using("source_table", "s")
    .on("target_table.id = s.id")
    .when_matched_then_update({"status": "s.status"})
)

Set Operations#

Combine queries using .union(), .intersect(), or .except_():

from sqlspec import sql

query_a = sql.select("id", "name").from_("active_users")
query_b = sql.select("id", "name").from_("archived_users")

# UNION ALL via all_=True
all_users = query_a.union(query_b, all_=True)

DDL Construction#

Create tables, indexes, schemas, and views programmatically:

from sqlspec import sql

create_table = (
    sql.create_table("users")
    .column("id", "integer", primary_key=True)
    .column("username", "text", nullable=False)
    .column("email", "text", unique=True)
)

create_idx = sql.create_index("idx_users_email", "users", "email")