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