sqlpiston 0.1.1

A low-level SQL library that builds parameterized SQL queries through Python operator overloading

SQLPiston

Write once, query everywhere — build SQL with Python operators.

中文版

SQLPiston is a low-level SQL library that builds parameterized SQL queries through Python operator overloading. AST nodes carry zero SQL knowledge; dialect-specific compilers translate the same AST into the right SQL for each database.

Install

git clone https://github.com/MuliMuri/sqlpiston.git
cd sqlpiston
pip install -e .

Python 3.9+ on Linux / macOS / Windows.

Basic Usage

from sqlpiston import Select, Field, Insert
from sqlpiston import DBEngine, DBType, Session
from dataclasses import dataclass

eng = DBEngine(DBType.SQLite)
eng.init_engine(":memory:")
session = Session(eng)

stmt = (
    Select()
    .select("id", "name", "age")
    .from_table("users")
    .where((Field("age") >= 18) & (Field("status") == "active"))
    .order_by("id", "ASC")
    .limit(10)
)
result = session.execute(stmt)

@dataclass
class User:
    id: int
    name: str
    age: int

users = result.map(User)

session.execute(
    Insert().into("users").values({"name": "Alice", "age": 25})
)

session.commit()
session.close()

More examples: JOIN, subquery, CTE, UPSERT, DDL

JOIN

Select() \
    .select("users.name", "orders.total") \
    .from_table("users") \
    .join("orders",
          Field("user_id", "orders") == Field("id", "users")) \
    .where(Field("total", "orders") > 100)

Subquery (IN / EXISTS / Scalar)

# IN subquery
admin_ids = Select().select("id").from_table("admins")
stmt = Select().select("name").from_table("users") \
    .where(Field("id").is_in(admin_ids))

# EXISTS
sub = Select().select("1").from_table("orders") \
    .where(Field("user_id", "orders") == Field("id", "users"))
stmt = Select().select("name").from_table("users").where(sub.exists())

# Scalar
avg = Select().select(SQLFunction("avg", "salary")).from_table("employees")
stmt = Select().select("name").from_table("staff").where(Field("salary") > avg)

CTE

cte = Select().select("*").from_table("sales") \
    .where(Field("amount") > 100).cte("big_sales")
stmt = Select().with_(cte).select("*").from_table("big_sales")

UPSERT — same AST, different SQL per dialect

Upsert() \
    .into("users") \
    .values({"id": 1, "name": "X"}) \
    .on_conflict("id") \
    .do_update({"name": "X"})

Dialect	Generated SQL
MySQL	INSERT INTO ... ON DUPLICATE KEY UPDATE
SQLite	INSERT INTO ... ON CONFLICT DO UPDATE

DDL

CreateTable().table("users").if_not_exists() \
    .column("id", "INTEGER", primary_key=True, nullable=False) \
    .column("name", "VARCHAR(255)")

AlterTable().table("users").add_column("age", "INTEGER", default=0)

DropTable().table("users").if_exists()

Highlights

• Operator overloading — Field("age") >= 18 builds AST, not a boolean
• Dialect-aware — same AST compiles to MySQL (%s, backticks) or SQLite (?, double-quotes)
• Full standard SQL — SELECT, INSERT, UPDATE, DELETE, UPSERT, CTE, UNION, subqueries, DDL
• Parameterized by design — values never interpolated into SQL strings
• Weak ORM — result-to-dataclass mapping, no identity map or change tracking

Documentation

cd docs
pip install -r requirements.txt
sphinx-build -b html source build/html

License

MIT