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)

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