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())

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)

Upserts (ON CONFLICT)¶

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

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¶

where helpers + pagination¶

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

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

    query = (
        SQL("select id, name, status from users")
        .where_eq("status", "active")
        .select_only("id", "name")
        .paginate(page=1, page_size=10)
    )
    result = session.execute(query)
    print(result.all())