buildaquery 1.1.9

A Python SQL query builder with cross-dialect execution primitives for OLTP workflows and early OLAP support.

Build-a-Query

A Python-based query builder designed to represent, compile, and execute SQL queries using a dialect-agnostic Abstract Syntax Tree (AST). Supports PostgreSQL, SQLite, DuckDB, ClickHouse, MySQL, MariaDB, CockroachDB, Oracle, and SQL Server.

Supported Databases

Build-a-Query currently supports the following databases:

• PostgreSQL
• SQLite
• DuckDB
• ClickHouse
• MySQL
• MariaDB
• CockroachDB
• Oracle
• SQL Server

Backend shape:

• Embedded / local-first: SQLite, DuckDB
• Server-based OLTP: PostgreSQL, MySQL, MariaDB, CockroachDB, Oracle, SQL Server
• Server-based OLAP: ClickHouse

Features

• Dialect-Agnostic AST: Build queries using high-level Python objects.
• Full DML Support: Create SELECT, INSERT, UPDATE, and DELETE statements.
• Advanced Querying: Support for CTEs (WITH), Subqueries, Set Operations (UNION, INTERSECT, EXCEPT), and Window Functions (OVER).
• Concurrency Controls: Optional lock_clause support on SELECT for row locking (FOR UPDATE, FOR SHARE, NOWAIT, SKIP LOCKED) with dialect-aware behavior.
• Dialect-Aware Upsert: InsertStatementNode.upsert_clause supports conflict handling across dialects (ON CONFLICT, ON DUPLICATE KEY UPDATE, and MERGE-based paths).
• Write-Return Payloads: returning_clause on INSERT/UPDATE/DELETE supports RETURNING and SQL Server OUTPUT equivalents.
• Batch Writes: InsertStatementNode.rows supports multi-row insert payloads, and executors expose execute_many(...) for driver-level bulk execution.
• Rich Expression Logic: Includes CASE expressions, IN, BETWEEN, and type casting.
• Expanded DDL Support: Table-level constraints (PRIMARY KEY, UNIQUE, FOREIGN KEY, CHECK), index statements (CREATE INDEX/DROP INDEX), and ALTER TABLE action paths.
• Visitor Pattern Traversal: Extensible architecture for analysis and compilation.
• Secure Compilation: Automatic parameterization for values plus compiler-side validation for table/schema/column/alias identifiers.
• Raw SQL Guardrails: execute_raw(...) can be policy-gated with raw_sql_policy (allow, deny_untrusted, deny_all) and explicit trusted=True.
• Execution Layer: Built-in support for executing compiled queries via psycopg (PostgreSQL/CockroachDB), duckdb (DuckDB), clickhouse-driver (ClickHouse), mysql-connector-python (MySQL), mariadb (MariaDB), oracledb (Oracle), pyodbc (SQL Server), and the standard library sqlite3 (SQLite).
• Transaction APIs: First-class transaction control with begin(), commit(), rollback(), savepoint(), rollback_to_savepoint(), and release_savepoint() across executors.
• Transaction Context Helper: with executor.transaction(): ... automatically commits on success and rolls back on error.
• Row Shaping: Executors can return rows as tuples, dicts, or typed model instances via row_output.
• Normalized Error + Retry APIs: Execution retry helpers (execute_with_retry, fetch_all_with_retry, fetch_one_with_retry, execute_many_with_retry) with normalized error types for deadlocks/serialization/lock timeouts/connection timeouts.
• Actionable Error Context: Normalized execution errors include dialect, operation, SQLSTATE when available, and redacted placeholder SQL context.
• Connection Management Controls: Executor lifecycle management (close, context manager), connect timeout configuration (connect_timeout_seconds), and pool hooks (acquire_connection, release_connection).
• Observability Hooks: Structured query observations plus lifecycle logging events (query/retry/transaction/connection) via ObservabilitySettings.
• Boundary Input Validation (Optional): Minimal Pydantic models/translators for validating external config and raw execution payloads before executor usage.
• SQL Preview Helpers: compiler.to_sql(...), executor.to_sql(...), and CompiledQuery.to_sql() expose placeholder-based SQL plus separate params for safe debug/inspection.
• Named-Parameter Convenience: Executors accept dict-style params for manual CompiledQuery(...) and execute_raw(...) inputs, rewriting :name markers into each dialect's native placeholders.
• Capability Introspection: executor.capabilities() exposes explicit dialect support flags so applications can branch safely on transactions, savepoints, returning, upsert, and lock-clause features.
• Deterministic Seeding: SeedRunner and SeedStep provide a small first-party seeding utility for ordered bootstrap data flows.
• Migration Tracking: MigrationRunner applies ordered schema changes, records applied versions, and supports explicit rollback of the latest applied migration when a down action is defined.
• Migration-Lite Utility: MigrationRunner and ordered MigrationStep definitions provide first-party schema version tracking with optional rollback support.

OLTP Capabilities

The project includes first-class OLTP-oriented support across AST, compiler, execution, and tests:

• Transactions: begin(), commit(), rollback(), savepoint(name), rollback_to_savepoint(name), and release_savepoint(name).
• Transaction context helper: with executor.transaction(): ... wraps explicit transaction lifecycle with automatic commit/rollback.
  • DuckDB note: savepoint APIs are runtime-version dependent; unsupported runtimes raise a clear executor RuntimeError.
• Row locking controls: lock clauses on SELECT with dialect-aware compilation (including NOWAIT and SKIP LOCKED where supported).
• Upsert paths: conflict-aware writes across dialects (ON CONFLICT, ON DUPLICATE KEY UPDATE, MERGE paths).
• Write-return payloads: returning_clause support (RETURNING and SQL Server OUTPUT equivalents).
• Batch writes: multi-row insert payloads via InsertStatementNode.rows and executor-level execute_many(...).
• Transient retry APIs: execute_with_retry(...), fetch_all_with_retry(...), fetch_one_with_retry(...), execute_many_with_retry(...) with RetryPolicy.
• Normalized transient errors: DeadlockError, SerializationError, LockTimeoutError, and ConnectionTimeoutError.
• Connection lifecycle and pool hooks: close(), context-manager control, connect_timeout_seconds, acquire_connection, and release_connection.
• Observability hooks: per-query timing (query_observer) and lifecycle logging events (event_observer) using ObservabilitySettings.
  • Built-in JSON logger helper: make_json_event_logger(logger=...) for one-line structured event logs.
  • Built-in adapters: InMemoryMetricsAdapter, InMemoryTracingAdapter, and compose_event_observers(...).
• Raw SQL guardrails: set raw_sql_policy="deny_untrusted" (or "deny_all") to restrict execute_raw(...); allow vetted calls with trusted=True.
• Capability introspection: call executor.capabilities() to branch safely on lock/upsert/returning/transaction support without hardcoding dialect names.
• OLTP integration coverage: contention/retry correctness, deadlock normalization, lost-update prevention, isolation visibility semantics, and lock behavior validation.

Dialect Notes

• MySQL does not support INTERSECT / EXCEPT or DROP TABLE ... CASCADE in this implementation (the compiler raises ValueError).
• SQLite does not support DROP TABLE ... CASCADE (the compiler raises ValueError).
• DuckDB does not support DROP TABLE ... CASCADE or trailing row-lock clauses (FOR UPDATE / FOR SHARE) in this compiler.
• ClickHouse does not support DROP TABLE ... CASCADE, row-lock clauses (FOR UPDATE / FOR SHARE), upsert clauses, generic RETURNING payloads, or explicit transaction/savepoint APIs in this executor.
• Oracle does not support IF EXISTS / IF NOT EXISTS in DROP TABLE/CREATE TABLE (the compiler raises ValueError), and EXCEPT is compiled as MINUS.
• SQL Server does not support EXCEPT ALL / INTERSECT ALL or DROP TABLE ... CASCADE in this implementation (the compiler raises ValueError).
• MariaDB supports INTERSECT / EXCEPT (including ALL), and accepts DROP TABLE ... CASCADE (treated as a no-op).
• CockroachDB supports INTERSECT / EXCEPT (including ALL) and DROP TABLE ... CASCADE.
• Upsert behavior:
  • PostgreSQL, SQLite, CockroachDB: ON CONFLICT (...) DO NOTHING / DO UPDATE.
  • MySQL, MariaDB: ON DUPLICATE KEY UPDATE (no DO NOTHING mode in this AST contract).
  • Oracle, SQL Server: MERGE-based upsert generation from InsertStatementNode + upsert_clause.
• Write-return behavior:
  • PostgreSQL, SQLite, CockroachDB: RETURNING ... on INSERT/UPDATE/DELETE.
  • MariaDB: RETURNING ... on INSERT/DELETE (not UPDATE in this compiler).
  • SQL Server: OUTPUT INSERTED... / OUTPUT DELETED... compiled from returning_clause.
  • MySQL: generic RETURNING payloads are not supported in this compiler.
• Oracle: RETURNING ... INTO requires out-bind plumbing and is not yet supported.
• Batch insert behavior:
  • All dialect compilers support multi-row insert payloads through InsertStatementNode.rows.
  • Oracle compiles multi-row inserts via INSERT ALL ... SELECT 1 FROM dual.
  • SQL Server and Oracle MERGE upsert paths currently require single-row values (not rows).
• DDL constraint/index behavior:
  • CreateStatementNode.constraints supports table-level PRIMARY KEY, UNIQUE, FOREIGN KEY, and CHECK.
  • CreateIndexStatementNode / DropIndexStatementNode are available with dialect-specific syntax/guards.
  • AlterTableStatementNode supports add/drop column and add/drop constraint actions, with dialect-specific limitations (for example, SQLite ALTER TABLE is intentionally restricted in this compiler).
• Transaction and lock behavior:
  • Lock clause behavior is dialect-specific; integration tests validate supported semantics, including NOWAIT and SKIP LOCKED paths where available.
• Normalized retry behavior:
  • Retry helpers target transient classes (deadlock/serialization/lock timeout/connection timeout); non-transient failures are surfaced directly.

Installation

For Users

Install Build-a-Query via pip:

pip install buildaquery

Install with optional database driver extras as needed:

pip install "buildaquery[postgres]"
pip install "buildaquery[cockroach]"
pip install "buildaquery[mysql]"
pip install "buildaquery[mariadb]"
pip install "buildaquery[oracle]"
pip install "buildaquery[mssql]"
pip install "buildaquery[duckdb]"
pip install "buildaquery[clickhouse]"
pip install "buildaquery[validation]"
pip install "buildaquery[all-databases]"

Requirements:

• Python 3.12+
• PostgreSQL database: A running PostgreSQL instance (version 12+ recommended). You can set this up locally, via Docker, or use a cloud service.
  • Example with Docker: docker run --name postgres -e POSTGRES_PASSWORD=yourpassword -d -p 5432:5432 postgres:15
• psycopg (install via buildaquery[postgres] or buildaquery[cockroach]) - the PostgreSQL/CockroachDB adapter for Python.
• MySQL database: A running MySQL instance (version 8.0+ recommended).
  • Example with Docker: docker run --name mysql -e MYSQL_ROOT_PASSWORD=yourpassword -e MYSQL_DATABASE=buildaquery -d -p 3306:3306 mysql:8.0
• mysql-connector-python (install via buildaquery[mysql]) - the MySQL adapter for Python.
• MariaDB database: A running MariaDB instance (MariaDB 10.3+ recommended).
  • Example with Docker (MariaDB): docker run --name mariadb -e MARIADB_ROOT_PASSWORD=yourpassword -e MARIADB_DATABASE=buildaquery -d -p 3306:3306 mariadb:11.4
• mariadb (install via buildaquery[mariadb]) - the MariaDB adapter for Python.
• CockroachDB database: A running CockroachDB instance.
  • Example with Docker (CockroachDB): docker run --name cockroach -p 26257:26257 -p 8080:8080 cockroachdb/cockroach:v24.3.1 start-single-node --insecure
• Oracle database: A running Oracle instance (Oracle XE is suitable for development).
  • Example with Docker (Oracle XE): docker run --name oracle-xe -e ORACLE_PASSWORD=yourpassword -e APP_USER=buildaquery -e APP_USER_PASSWORD=yourpassword -d -p 1521:1521 gvenzl/oracle-xe:21-slim
• oracledb (install via buildaquery[oracle]) - the Oracle adapter for Python.
• SQL Server database: A running SQL Server instance (Express is suitable for development).
  • Example with Docker (SQL Server Express): docker run --name sqlserver -e ACCEPT_EULA=Y -e MSSQL_SA_PASSWORD=yourpassword -e MSSQL_PID=Express -d -p 1433:1433 mcr.microsoft.com/mssql/server:2022-latest
• pyodbc (install via buildaquery[mssql]) - the SQL Server adapter for Python.
• python-dotenv (automatically installed as a dependency) - for loading environment variables from a .env file.
• SQLite: Uses Python's standard library sqlite3 module.
  • SQLite Version: SQLite 3.x via Python's sqlite3 module (the exact SQLite version depends on your Python build; check sqlite3.sqlite_version at runtime).
• DuckDB: Embedded OLAP database via the duckdb Python package.
  • Install driver via buildaquery[duckdb].
• ClickHouse: Columnar OLAP database via clickhouse-driver.
  • Install driver via buildaquery[clickhouse].

Environment Variables

To connect to your PostgreSQL database, set the following environment variables (or use a .env file with python-dotenv):

• DB_HOST: PostgreSQL host (e.g., localhost)
• DB_PORT: PostgreSQL port (e.g., 5432)
• DB_NAME: Database name (e.g., mydatabase)
• DB_USER: Database username (e.g., postgres)
• DB_PASSWORD: Database password (e.g., yourpassword)

Example .env file:

DB_HOST=localhost
DB_PORT=5432
DB_NAME=buildaquery
DB_USER=postgres
DB_PASSWORD=yourpassword

For MySQL, you can use a connection URL directly in code (e.g., mysql://user:password@host:3306/dbname) or set your own environment variables and construct the URL similarly.

For Oracle, use a connection URL in the format oracle://user:password@host:port/service_name (for example: oracle://buildaquery:password@127.0.0.1:1521/XEPDB1).

For SQL Server, use a connection URL in the format mssql://user:password@host:port/dbname?driver=... (for example: mssql://sa:password@127.0.0.1:1433/buildaquery?driver=ODBC+Driver+18+for+SQL+Server&encrypt=no&trust_server_certificate=yes).

For MariaDB, use a connection URL in the format mariadb://user:password@host:port/dbname (for example: mariadb://root:password@127.0.0.1:3306/buildaquery).

For CockroachDB, use a connection URL in the format postgresql://user@host:port/dbname?sslmode=disable (for example: postgresql://root@127.0.0.1:26257/buildaquery?sslmode=disable).

For DuckDB, use a local database file path (for example: static/test-duckdb/db.duckdb) or :memory:.

For ClickHouse, use a connection URL in the format clickhouse://user:password@host:port/database (for example: clickhouse://buildaquery:password@127.0.0.1:9001/buildaquery_test for this repo's Docker test service).

For Developers

Clone the repository and set up the development environment:

git clone https://github.com/yourusername/buildaquery.git
cd buildaquery

Install dependencies using Poetry:

poetry install

Activate the virtual environment:

poetry shell

Debugging SQL

Use to_sql() when you want to inspect generated SQL without executing it. The returned SQL always keeps dialect placeholders intact; values stay in params, which avoids turning the debug path into an interpolation or logging risk.

from buildaquery.abstract_syntax_tree.models import ColumnNode, SelectStatementNode, TableNode
from buildaquery.compiler import PostgresCompiler
from buildaquery.execution.sqlite import SqliteExecutor

query = SelectStatementNode(
    select_list=[ColumnNode(name="id")],
    from_table=TableNode(name="users"),
)

compiled = PostgresCompiler().to_sql(query)
print(compiled.to_sql())
print(compiled.params)

executor = SqliteExecutor(connection_info="static/test-sqlite/db.sqlite")
preview = executor.to_sql(query)
print(preview.to_sql())
print(preview.params)

Named Parameters For Manual SQL

When you build CompiledQuery(...) yourself or call execute_raw(...), you can pass dict-style params using :name markers. Executors rewrite those markers into the target dialect's placeholder format before execution:

• PostgreSQL / MySQL / CockroachDB: %s
• SQLite / MariaDB / SQL Server / DuckDB / ClickHouse: ?
• Oracle: :1, :2, ...

Values remain parameterized. They are not interpolated into SQL text.

from buildaquery.compiler.compiled_query import CompiledQuery
from buildaquery.execution.sqlite import SqliteExecutor

executor = SqliteExecutor(connection_info="static/test-sqlite/db.sqlite")

query = CompiledQuery(
    sql="SELECT id, email FROM users WHERE email = :email AND tenant_id = :tenant_id",
    params={"email": "[e-mail]", "tenant_id": 42},
)

preview = executor.to_sql(query)
print(preview.to_sql())   # SELECT id, email FROM users WHERE email = ? AND tenant_id = ?
print(preview.params)     # ['[e-mail]', 42]

Capability Introspection

Use executor.capabilities() when application behavior depends on dialect support for transactions, savepoints, lock clauses, upsert, or write-return payloads.

The contract is explicit and conservative. It is intended for safe branching, not optimistic probing.

from buildaquery.execution.postgres import PostgresExecutor

executor = PostgresExecutor(connection=object())
capabilities = executor.capabilities()

if capabilities.select_for_update and capabilities.lock_skip_locked:
    print("Can use FOR UPDATE SKIP LOCKED")

if capabilities.insert_returning:
    print("Can fetch rows directly from INSERT")

print(capabilities.to_dict())

Deterministic Seeding

Use SeedRunner when you want a small, first-party way to load deterministic bootstrap data through the existing executor APIs.

• Steps run in order.
• AST statements, CompiledQuery(...), and executor callbacks are supported.
• Transaction wrapping is enabled by default when the executor supports transactions.
• Raw SQL callback steps still obey execute_raw(...) policy enforcement.

from buildaquery.abstract_syntax_tree.models import ColumnNode, InsertStatementNode, LiteralNode, TableNode
from buildaquery.compiler.compiled_query import CompiledQuery
from buildaquery.seeding import SeedRunner, SeedStep

steps = [
    SeedStep(
        name="insert-admin",
        action=InsertStatementNode(
            table=TableNode(name="users"),
            columns=[ColumnNode(name="id"), ColumnNode(name="email")],
            values=[LiteralNode(1), LiteralNode("[e-mail]")],
        ),
    ),
    SeedStep(
        name="insert-auditor",
        action=CompiledQuery(
            sql="INSERT INTO users (id, email) VALUES (?, ?)",
            params=[2, "[e-mail]"],
        ),
    ),
]

summary = SeedRunner(transactional=True).run(executor, steps)
print(summary.completed_steps)

Migration-Lite Utility

Use MigrationRunner when you want a small, first-party way to apply ordered schema changes and track applied versions inside the database.

• Pending migrations run in ascending version order.
• Each migration is wrapped in its own transaction by default when the executor supports transactions.
• rollback_last(...) only runs when the latest applied migration has an explicit down action.
• Internal tracking-table writes use vetted SQL and callable migration actions still obey executor execute_raw(...) policy enforcement.

from buildaquery.compiler.compiled_query import CompiledQuery
from buildaquery.migrations import MigrationRunner, MigrationStep

runner = MigrationRunner()

migrations = [
    MigrationStep(
        version=1,
        name="create-users",
        up=CompiledQuery(
            sql="CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT NOT NULL)"
        ),
        down=CompiledQuery(sql="DROP TABLE users"),
    ),
]

apply_summary = runner.apply(executor, migrations)
print(apply_summary.applied_versions)

rollback_summary = runner.rollback_last(executor, migrations)
print(rollback_summary.rolled_back)

Quick Start

For the fastest first run, start with SQLite in the section below (### SQLite Quick Start (Recommended)), which requires no external database server.

PostgreSQL Full CRUD Quick Start

Here's a full CRUD example for PostgreSQL using environment variables for connection setup (see Environment Variables section above).

from dotenv import load_dotenv
import os
from buildaquery.execution.postgres import PostgresExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

# Load environment variables
load_dotenv()

# Build connection string from environment variables
db_host = os.getenv('DB_HOST')
db_port = os.getenv('DB_PORT')
db_name = os.getenv('DB_NAME')
db_user = os.getenv('DB_USER')
db_password = os.getenv('DB_PASSWORD')

connection_string = f"postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}"

# Set up executor with your PostgreSQL connection
executor = PostgresExecutor(connection_info=connection_string)

# Define table
users_table = TableNode(name="users")

# Create table
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="SERIAL", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="TEXT", not_null=True),
        ColumnDefinitionNode(name="age", data_type="INTEGER")
    ]
)
executor.execute(create_stmt)

# Insert data
insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

# Query data
select_stmt = SelectStatementNode(
    select_list=[StarNode()],  # SELECT *
    from_table=users_table
)
results = executor.execute(select_stmt)
print(results)  # [(1, 'Alice', 30)]

# Drop table
drop_stmt = DropStatementNode(table=users_table, if_exists=True)
executor.execute(drop_stmt)

SQLite Quick Start (Recommended)

from buildaquery.execution.sqlite import SqliteExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

executor = SqliteExecutor(connection_info="static/test-sqlite/db.sqlite")

users_table = TableNode(name="users")
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="INTEGER", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="TEXT", not_null=True),
        ColumnDefinitionNode(name="age", data_type="INTEGER")
    ]
)
executor.execute(create_stmt)

insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[StarNode()],
    from_table=users_table
)
print(executor.execute(select_stmt))

drop_stmt = DropStatementNode(table=users_table, if_exists=True)
executor.execute(drop_stmt)

DuckDB Quick Start

from buildaquery.execution.duckdb import DuckDbExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

executor = DuckDbExecutor(connection_info="static/test-duckdb/sample.duckdb")

users_table = TableNode(name="users")
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="INTEGER", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="VARCHAR", not_null=True),
        ColumnDefinitionNode(name="age", data_type="INTEGER")
    ]
)
executor.execute(create_stmt)

insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[StarNode()],
    from_table=users_table
)
print(executor.execute(select_stmt))

drop_stmt = DropStatementNode(table=users_table, if_exists=True)
executor.execute(drop_stmt)

ClickHouse Quick Start

from buildaquery.execution.clickhouse import ClickHouseExecutor
from buildaquery.abstract_syntax_tree.models import (
    ColumnNode,
    InsertStatementNode,
    LiteralNode,
    SelectStatementNode,
    TableNode,
)

executor = ClickHouseExecutor(connection_info="clickhouse://default@127.0.0.1:9000/default")
events_table = TableNode(name="events")

# ClickHouse table creation usually requires an engine clause, so use raw SQL here.
executor.execute_raw("CREATE TABLE IF NOT EXISTS events (id UInt32, value String) ENGINE = Memory")

insert_stmt = InsertStatementNode(
    table=events_table,
    columns=[ColumnNode(name="id"), ColumnNode(name="value")],
    values=[LiteralNode(value=1), LiteralNode(value="hello")],
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[ColumnNode(name="id"), ColumnNode(name="value")],
    from_table=events_table,
)
print(executor.execute(select_stmt))

executor.execute_raw("DROP TABLE IF EXISTS events")
executor.close()

MySQL Quick Start

from buildaquery.execution.mysql import MySqlExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

executor = MySqlExecutor(connection_info="mysql://root:password@127.0.0.1:3306/buildaquery")

users_table = TableNode(name="users")
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="INT AUTO_INCREMENT", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="VARCHAR(255)", not_null=True),
        ColumnDefinitionNode(name="age", data_type="INT")
    ]
)
executor.execute(create_stmt)

insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[StarNode()],
    from_table=users_table
)
print(executor.execute(select_stmt))

drop_stmt = DropStatementNode(table=users_table, if_exists=True)
executor.execute(drop_stmt)

Oracle Quick Start

from buildaquery.execution.oracle import OracleExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

executor = OracleExecutor(connection_info="oracle://buildaquery:password@127.0.0.1:1521/XEPDB1")

users_table = TableNode(name="users")
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="NUMBER", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="VARCHAR2(255)", not_null=True),
        ColumnDefinitionNode(name="age", data_type="NUMBER")
    ]
)
executor.execute(create_stmt)

insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="id"), ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value=1), LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[StarNode()],
    from_table=users_table
)
print(executor.execute(select_stmt))

drop_stmt = DropStatementNode(table=users_table, cascade=True)
executor.execute(drop_stmt)

SQL Server Quick Start

from buildaquery.execution.mssql import MsSqlExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

executor = MsSqlExecutor(connection_info="mssql://sa:password@127.0.0.1:1433/buildaquery?driver=ODBC+Driver+18+for+SQL+Server&encrypt=no&trust_server_certificate=yes")

users_table = TableNode(name="users")
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="INT", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="NVARCHAR(255)", not_null=True),
        ColumnDefinitionNode(name="age", data_type="INT")
    ]
)
executor.execute(create_stmt)

insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="id"), ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value=1), LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[StarNode()],
    from_table=users_table
)
print(executor.execute(select_stmt))

drop_stmt = DropStatementNode(table=users_table, if_exists=True)
executor.execute(drop_stmt)

MariaDB Quick Start

from buildaquery.execution.mariadb import MariaDbExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

executor = MariaDbExecutor(connection_info="mariadb://root:password@127.0.0.1:3306/buildaquery")

users_table = TableNode(name="users")
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="INT AUTO_INCREMENT", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="VARCHAR(255)", not_null=True),
        ColumnDefinitionNode(name="age", data_type="INT")
    ]
)
executor.execute(create_stmt)

insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[StarNode()],
    from_table=users_table
)
print(executor.execute(select_stmt))

drop_stmt = DropStatementNode(table=users_table, if_exists=True, cascade=True)
executor.execute(drop_stmt)

CockroachDB Quick Start

from buildaquery.execution.cockroachdb import CockroachExecutor
from buildaquery.abstract_syntax_tree.models import (
    CreateStatementNode, TableNode, ColumnDefinitionNode,
    InsertStatementNode, ColumnNode, LiteralNode,
    SelectStatementNode, StarNode, DropStatementNode
)

executor = CockroachExecutor(connection_info="postgresql://root@127.0.0.1:26257/buildaquery?sslmode=disable")

users_table = TableNode(name="users")
create_stmt = CreateStatementNode(
    table=users_table,
    columns=[
        ColumnDefinitionNode(name="id", data_type="INT", primary_key=True),
        ColumnDefinitionNode(name="name", data_type="STRING", not_null=True),
        ColumnDefinitionNode(name="age", data_type="INT")
    ]
)
executor.execute(create_stmt)

insert_stmt = InsertStatementNode(
    table=users_table,
    columns=[ColumnNode(name="id"), ColumnNode(name="name"), ColumnNode(name="age")],
    values=[LiteralNode(value=1), LiteralNode(value="Alice"), LiteralNode(value=30)]
)
executor.execute(insert_stmt)

select_stmt = SelectStatementNode(
    select_list=[StarNode()],
    from_table=users_table
)
print(executor.execute(select_stmt))

drop_stmt = DropStatementNode(table=users_table, if_exists=True, cascade=True)
executor.execute(drop_stmt)

For more examples, see the examples/ directory (including examples/sample_syntax_quickstart.py, examples/sample_starter_templates.py, examples/sample_duckdb.py, examples/sample_clickhouse.py, examples/sample_mysql.py, examples/sample_oracle.py, examples/sample_mssql.py, examples/sample_mariadb.py, examples/sample_cockroachdb.py, examples/sample_transactions.py, examples/sample_connection_management.py, examples/sample_observability.py, and examples/sample_observability_integration.py). For copy-paste starter snippets across CRUD/upsert/transaction/retry/observability flows, see examples/sample_starter_templates.py. For boundary validation patterns with external payloads, see examples/sample_validation.py. For dict-style named params on manual CompiledQuery or execute_raw(...) inputs, see examples/sample_named_params.py. For runtime branching on dialect support, see examples/sample_capabilities.py. For transaction control, see examples/sample_transactions.py. For automatic commit/rollback patterns, prefer with executor.transaction(): .... For opt-in dict/model row outputs, see examples/sample_row_shaping.py. For normalized retry/error handling, use RetryPolicy with *_with_retry(...) executor APIs. For connection management patterns, see examples/sample_connection_management.py. For observability hooks, see examples/sample_observability.py. For app-level observability wiring (no library source edits), see examples/sample_observability_integration.py. For upsert patterns, see examples/sample_upsert.py. For write-return payloads, see examples/sample_returning.py. For batch writes, see examples/sample_batch_write.py. For migrations, see examples/sample_migrations.py. For DDL constraints/indexes and ALTER TABLE, see examples/sample_ddl_constraints.py.

Development Setup

Prerequisites

• Python 3.12+
• Poetry (for dependency management)
• Docker (for running integration tests)

Setting Up the Environment

1. Clone the repository:

   git clone https://github.com/yourusername/buildaquery.git
   cd buildaquery
   

2. Install dependencies:

   poetry install
   

3. Activate the virtual environment:

   poetry shell
   

Running Tests

Unit Tests

Run unit tests for all modules:

poetry run pytest buildaquery/tests

Integration Tests

Integration tests require PostgreSQL, MySQL, MariaDB, CockroachDB, Oracle, SQL Server, and ClickHouse databases (and the respective drivers). DuckDB and SQLite integration tests use local database files. Start the server-based test databases using Docker:

docker-compose up -d

Then run integration tests:

poetry run pytest tests

SQLite integration tests use the file-based database at static/test-sqlite/db.sqlite. CockroachDB integration tests use the SQL port 26258 by default (see tests/README.md for the full URL and override env var). OLTP-focused integration coverage includes contention/retry correctness, deadlock normalization, lost-update prevention patterns, isolation visibility semantics, and row-lock behavior (NOWAIT/SKIP LOCKED). See tests/test_oltp_integration.py for the dedicated OLTP integration scenarios.

All Tests

Run all tests (unit and integration):

poetry run all-tests

Package Checks (PyPI Readiness)

Build source/wheel artifacts and validate distribution metadata:

poetry run package-check

Running Examples

Execute a sample script:

poetry run python examples/sample_syntax_quickstart.py

Project Structure

• buildaquery/abstract_syntax_tree/: Defines query nodes and AST models.
• buildaquery/traversal/: Base classes for AST traversal (Visitor/Transformer pattern).
• buildaquery/compiler/: Dialect-specific SQL generation (PostgreSQL, SQLite, DuckDB, ClickHouse, MySQL, MariaDB, CockroachDB, Oracle, SQL Server).
• buildaquery/execution/: Database connection and execution logic.
• tests/: Exhaustive unit and integration tests.
• examples/: Practical demonstrations of the library.
• scripts/: Utility scripts for testing and maintenance.

Contributing

Contributions are welcome! Please see the contributing guidelines for more information.

For package release steps, see RELEASES.md.

License

This project is licensed under the MIT License - see the LICENSE.txt file for details.