chocolatine 1.1.0

Dynamic generation of SQL queries

chocolatine

A Python SQL query builder that feels like writing Python, not SQL strings.

Chocolatine lets you build complex, type-safe SQL queries programmatically — with full IDE autocompletion, runtime type checking, and zero string concatenation.

from chocolatine import Query, Col as _, Window, With

# Revenue per customer with running total — all in Python
monthly = Query.get_rows(
    "orders",
    cols=[
        _("customer_id"),
        _("created_at").date_format("%Y-%m").alias("month"),
        (_("quantity") * _("unit_price")).alias("revenue"),
    ],
    groups=["customer_id", "month"]
)

With(
    ctes=[("monthly", monthly)],
    query=Query.get_rows("monthly", cols=[
        _("customer_id"),
        _("month"),
        _("revenue"),
        Window.sum(_("revenue"), partition_by=[_("customer_id")], order_by=[_("month")]).alias("running_total"),
    ])
).build()

WITH monthly AS (
  SELECT customer_id, DATE_FORMAT(created_at, '%Y-%m') AS month, (quantity * unit_price) AS revenue
  FROM orders GROUP BY customer_id, month
)
SELECT customer_id, month, revenue,
       SUM(revenue) OVER (PARTITION BY customer_id ORDER BY month) AS running_total
FROM monthly

---

Why Chocolatine?

Most SQL builders are either too verbose (SQLAlchemy Core), too tied to an ORM (Django ORM, SQLModel), or too low-level (raw f-strings). Chocolatine sits in the middle: lightweight, expressive, and complete.

Feature	Chocolatine	Raw SQL strings	SQLAlchemy Core
Type-safe at runtime	Yes	No	Partial
IDE autocompletion	Yes	No	Yes
Near-zero dependencies	Yes	Yes	No
Window functions	Yes	Yes	Yes
CTEs	Yes	Yes	Yes
Learning curve	Low	None	High

---

Installation

pip install chocolatine

Requires Python 3.13+.

---

Quick start

from chocolatine import Query, Col as _

# SELECT
Query.get_rows("users", filters=[_("active") == 1], limit=10).build()
# SELECT * FROM users WHERE (active = 1) LIMIT 10

# INSERT
Query.insert_row("users", [_("name"), _("email")], ("Alice", "[e-mail]")).build()
# INSERT INTO users (name, email) VALUES ('Alice', '[e-mail]')

# UPDATE
Query.update_rows("users", filters=[_("id") == 1], assignations=[_("name") == "Bob"]).build()
# UPDATE users SET (name = 'Bob') WHERE (id = 1)

# DELETE
Query.delete_rows("users", filter=_("active") == 0).build()
# DELETE FROM users WHERE (active = 0)

---

Core concepts

Columns

from chocolatine import Col as _

_("price")                          # price
_("u.email")                        # u.email  (table prefix)
_("amount").alias("total")          # amount AS total
_("amount:total")                   # amount AS total  (shorthand)
_("amount").sum().alias("total")    # SUM(amount) AS total
_("name").upper()                   # UPPER(name)
_("created_at").date_format("%Y-%m")# DATE_FORMAT(created_at, '%Y-%m')

Filtering

Filters are combined with AND automatically when chained:

q = Query.get_rows("users")
q.filter(_("age") > 18)
q.filter(_("city") == "Paris")
q.filter(_("active") == 1)
# WHERE (((age > 18) AND (city = 'Paris')) AND (active = 1))

Use operators directly for more control:

from chocolatine import Col as _

_("age").between(18, 65)                        # age BETWEEN 18 AND 65
_("email").is_null()                             # email IS NULL
_("status") << ("active", "pending")            # status IN ('active', 'pending')
_("name") >> "%alice%"                          # name LIKE '%alice%'
~_("email").is_null()                            # NOT (email IS NULL)
(_("age") > 18) & (_("city") == "Paris")        # (age > 18) AND (city = 'Paris')
(_("tier") == "gold") | (_("tier") == "silver") # (tier = 'gold') OR (tier = 'silver')

Arithmetic expressions

(_("qty") * _("unit_price")).alias("total")          # (qty * unit_price) AS total
(_("revenue") - _("cost")).alias("margin")           # (revenue - cost) AS margin
(_("revenue") / _("sessions") * 100).alias("rate")  # ((revenue / sessions) * 100) AS rate
_("stock") + 1                                        # (stock + 1)

# Use in filters
Query.get_rows("orders", filters=[_("qty") * _("unit_price") > 500])
# WHERE ((qty * unit_price) > 500)

Joins

from chocolatine import JoinType

Query.get_rows(
    "orders o",
    cols=[_("o.id"), _("c.name"), _("c.email")],
    joins=[
        ("customers c", _("c.id") == _("o.customer_id")),
        ("order_items oi", _("oi.order_id") == _("o.id"), JoinType.Left),
    ]
)

# CROSS JOIN and NATURAL JOIN (no condition needed)
q.join("log", join_type=JoinType.Cross)
q.join("categories", join_type=JoinType.Natural)

Aggregations

from chocolatine import count, sum, Col as _

Query.get_rows(
    "orders",
    cols=[
        _("customer_id"),
        count().alias("nb_orders"),
        _("amount").sum().alias("total"),
        _("amount").average().alias("avg"),
    ],
    groups=["customer_id"],
    filters=[_("amount").sum() > 100]   # routed to HAVING automatically
)
# SELECT customer_id, COUNT(*) AS nb_orders, SUM(amount) AS total, AVG(amount) AS avg
# FROM orders GROUP BY customer_id HAVING (SUM(amount) > 100)

String functions

_("name").upper()                       # UPPER(name)
_("name").lower()                       # LOWER(name)
_("name").trim()                        # TRIM(name)
_("name").length()                      # LENGTH(name)
_("name").replace("foo", "bar")         # REPLACE(name, 'foo', 'bar')
_("bio").substring(1, 255)              # SUBSTRING(bio, 1, 255)
_("code").lpad(10, "0")                 # LPAD(code, 10, '0')
_("name").left(3)                       # LEFT(name, 3)
_("score").round(2)                     # ROUND(score, 2)
_("created_at").date_format("%Y-%m")    # DATE_FORMAT(created_at, '%Y-%m')
_("password").md5()                     # MD5(password)

# Concatenation
_("first_name") & " " & _("last_name")  # CONCAT(first_name, ' ', last_name)

Date functions

from chocolatine import now, curdate, IntervalUnit, Col as _

_("created_at").date(_("created_at"))           # DATE(created_at)
_("created_at").datediff(_("updated_at"))       # DATEDIFF(created_at, updated_at)
_("created_at").date_add(7, IntervalUnit.Day)   # DATE_ADD(created_at, INTERVAL 7 DAY)
_("created_at").date_sub(1, IntervalUnit.Month) # DATE_SUB(created_at, INTERVAL 1 MONTH)

now()     # NOW()
curdate() # CURDATE()

# Use in conditions
_("updated_at") == now()     # (updated_at = NOW())
_("created_at") > curdate()  # (created_at > CURDATE())

Conditional expressions

from chocolatine import When, ColWhen, Coalesce, NullIf, IfNull, SqlIf, Col as _

# CASE WHEN
When(
    (_("score") >= 90, _("score") >= 70, _("score") >= 50),
    ("A", "B", "C"),
    else_returned_val="F",
    compact=True
)
# CASE WHEN (score >= 90) THEN 'A' WHEN (score >= 70) THEN 'B' ... ELSE 'F' END

# CASE col WHEN value
ColWhen("status", ("active", "pending"), ("Active", "Pending"), else_returned_val="Other")
# CASE status WHEN 'active' THEN 'Active' WHEN 'pending' THEN 'Pending' ELSE 'Other' END

# COALESCE
Coalesce(_("phone"), _("mobile"), "N/A").alias("contact")
# COALESCE(phone, mobile, 'N/A') AS contact

# NULLIF — returns NULL if both args are equal
NullIf(_("score"), 0).alias("score")
# NULLIF(score, 0) AS score

# IFNULL — returns fallback if first arg is NULL
IfNull(_("phone"), "N/A").alias("contact")
# IFNULL(phone, 'N/A') AS contact

# IF — MySQL IF()
SqlIf(_("score") >= 50, "pass", "fail").alias("result")
# IF((score >= 50), 'pass', 'fail') AS result

Type conversion

from chocolatine import Cast, SqlType, Col as _

Cast(_("price"), SqlType.Integer).alias("int_price")
# CAST(price AS INT) AS int_price

Cast(_("code"), SqlType.String).alias("str_code")
# CAST(code AS VARCHAR(255)) AS str_code

EXISTS / NOT EXISTS

from chocolatine import Exists, Query, Col as _

subquery = Query.get_rows("orders", filters=[_("customer_id") == _("c.id")])

Query.get_rows("customers c", filters=[Exists(subquery)])
# SELECT * FROM customers c WHERE EXISTS (SELECT * FROM orders WHERE (customer_id = c.id))

Query.get_rows("customers c", filters=[~Exists(subquery)])
# SELECT * FROM customers c WHERE NOT EXISTS (...)

Subqueries (derived tables)

from chocolatine import Query, Subquery, Col as _

inner = Query.get_rows("orders", cols=[_("customer_id"), _("amount")])

Query.get_rows(Subquery(inner, "o"), cols=[_("o.customer_id")])
# SELECT o.customer_id FROM (SELECT customer_id, amount FROM orders) AS o

# Subquery in JOIN
q = Query.get_rows("customers c")
q.join(Subquery(inner, "o"), _("o.customer_id") == _("c.id"))

ANY / ALL

from chocolatine import Query, Col as _
from chocolatine.expr.any_all import AnySubquery, AllSubquery

prices = Query.get_rows("products", cols=[_("price")])

_("price") > AnySubquery(prices)   # (price > ANY (SELECT price FROM products))
_("price") < AllSubquery(prices)   # (price < ALL (SELECT price FROM products))

Window functions

from chocolatine import Window, Col as _

Window.row_number(partition_by=[_("dept")], order_by=[_("salary").desc()]).alias("rn")
# ROW_NUMBER() OVER (PARTITION BY dept ORDER BY salary DESC) AS rn

Window.sum(_("amount"), partition_by=[_("dept")]).alias("dept_total")
# SUM(amount) OVER (PARTITION BY dept) AS dept_total

Window.lag(_("salary"), offset=1, order_by=[_("hire_date")])
# LAG(salary, 1) OVER (ORDER BY hire_date)

Window.ntile(4, order_by=[_("score").desc()])
# NTILE(4) OVER (ORDER BY score DESC)

Available: row_number, rank, dense_rank, ntile, cume_dist, percent_rank, sum, avg, count, min, max, lag, lead, first_value, last_value.

CTEs (WITH)

from chocolatine import With, Query, Col as _

active_users = Query.get_rows("users", filters=[_("active") == 1])
recent_orders = Query.get_rows("orders", filters=[_("created_at") > "2024-01-01"])

With(
    ctes=[
        ("active_users", active_users),
        ("recent_orders", recent_orders),
    ],
    query=Query.get_rows("active_users",
        joins=[("recent_orders", _("recent_orders.user_id") == _("active_users.id"))]
    )
).build()
# WITH active_users AS (...), recent_orders AS (...) SELECT * FROM active_users JOIN ...

# Recursive CTE
With(
    ctes=[("org", base_query.union_all(recursive_query))],
    query=Query.get_rows("org"),
    recursive=True
)
# WITH RECURSIVE org AS (...) SELECT * FROM org

Set operations

# UNION / UNION ALL
Query.get_rows("customers").union(Query.get_rows("prospects")).build()
# SELECT * FROM customers UNION SELECT * FROM prospects

Query.get_rows("active").union_all(Query.get_rows("archived")).build()
# SELECT * FROM active UNION ALL SELECT * FROM archived

# INTERSECT / EXCEPT
Query.get_rows("a").intersect(Query.get_rows("b")).build()
# SELECT * FROM a INTERSECT SELECT * FROM b

Query.get_rows("a").except_(Query.get_rows("b")).build()
# SELECT * FROM a EXCEPT SELECT * FROM b

# ALL variants: intersect_all(), except_all()

Raw SQL escape hatch

from chocolatine import RawExpr, Col as _

RawExpr("NOW()").alias("ts")                   # NOW() AS ts
_("updated_at") == RawExpr("NOW()")            # (updated_at = NOW())
RawExpr("CURRENT_USER()").alias("user")        # CURRENT_USER() AS user

DDL

from chocolatine import Query, Col as _, SqlType

# CREATE TABLE
Query.create_table("users", cols=[
    _("email",      type=SqlType.String,  not_null=True, unique=True),
    _("name",       type=SqlType.String,  not_null=True),
    _("score",      type=SqlType.Integer, default=0),
    _("created_at", type=SqlType.DateTime),
], auto_id=True)
# CREATE TABLE users (id MEDIUMINT NOT NULL AUTO_INCREMENT, email VARCHAR(255) NOT NULL UNIQUE, ...)

# ALTER TABLE
Query.add_col("users", Col("country", type=SqlType.String), after="name")
Query.rename_col("users", "email", "email_address")
Query.drop_col("users", "legacy_field")

# DROP / TRUNCATE
Query.drop_table("users")
Query.truncate("users")

INSERT ON DUPLICATE KEY UPDATE

Query.insert_row("products", [_("sku"), _("name"), _("price")], ("ABC", "Widget", 9.99)) \
    .on_duplicate_key_update(_("price") == 9.99, _("updated_at") == RawExpr("NOW()"))
# INSERT INTO products (sku, name, price) VALUES ('ABC', 'Widget', 9.99)
# ON DUPLICATE KEY UPDATE (price = 9.99), (updated_at = NOW())

Compact vs extended output

q = Query.get_rows("users", compact=False, filters=[_("active") == 1])
print(q.build())
# SELECT *
# FROM users
# WHERE (active = 1)

q.compact = True   # propagates to all sub-expressions
print(q.build())
# SELECT * FROM users WHERE (active = 1)

---

Full feature list

Query types: SELECT, INSERT, UPDATE, DELETE, CREATE TABLE, ALTER TABLE (add/rename/change/drop column), DROP TABLE, TRUNCATE

Filtering: WHERE, HAVING (auto-routed), IS NULL, IS NOT NULL, BETWEEN, IN, LIKE, EXISTS, NOT EXISTS, AND, OR, NOT

Expressions: arithmetic (+, -, *, /, %), string functions, date functions, COALESCE, NULLIF, IFNULL, IF(), CAST, CASE/WHEN, subqueries, raw SQL

Aggregations: COUNT, COUNT(DISTINCT), SUM, AVG, MIN, MAX + GROUP BY + HAVING

Window functions: ROW_NUMBER, RANK, DENSE_RANK, NTILE, CUME_DIST, PERCENT_RANK, SUM, AVG, COUNT, MIN, MAX, LAG, LEAD, FIRST_VALUE, LAST_VALUE

Set operations: UNION, UNION ALL, INTERSECT, INTERSECT ALL, EXCEPT, EXCEPT ALL

Subqueries: EXISTS, NOT EXISTS, derived tables (FROM subquery), ANY, ALL

CTEs: WITH, WITH RECURSIVE (multiple CTEs supported)

DDL constraints: NOT NULL, UNIQUE, DEFAULT, AUTO_INCREMENT primary key

Other: JOINs (INNER/LEFT/RIGHT/FULL/CROSS/NATURAL), ORDER BY, LIMIT, OFFSET, DISTINCT, aliases, table prefixes, compact/extended rendering

---

SQL dialect

Chocolatine currently targets MySQL. Most queries are compatible with MariaDB. PostgreSQL and SQLite support is not guaranteed for dialect-specific features (e.g., MEDIUMINT, DATE_FORMAT, IF()).

---

Development

# Run tests
pytest

# Run tests with coverage
pytest --cov=chocolatine --cov-report=term-missing

383 tests, 99% coverage.

---

Related

• choc-expr — the lightweight expression templating engine powering Chocolatine

---

Contributors

• pe-brian — author
• ryry-shi — testing