🇺🇦 pgmini 🇺🇦

PostgreSQL query builder with two core principles:

• simple — predictable, no magic, python code maps 1:1 to SQL structure
• fast — all objects are immutable (built on attrs), no heavy machinery

The library builds SQL strings and parameter lists — nothing else. It doesn't manage connections, doesn't escape params, doesn't validate your schema. Use it together with asyncpg / psycopg which do those jobs well.

All public methods use PascalCase (From, Where, And, Else, With, As ...) to avoid collisions with python reserved words.

Table of contents

• Installation
• Quick start
• Core concepts
• SELECT
• Operations
• Logical operators
• Functions
• CASE / ARRAY
• INSERT
• UPDATE
• DELETE
• MERGE
• VALUES as a FROM item
• Subqueries and CTE (WITH)
• Row locking (FOR UPDATE / FOR SHARE)
• API cheat sheet

Installation

pip install pgmini

Quick start

from pgmini import Select, Table, build

User = Table('user')  # columns are dynamic: User.<anything> is a column

q = Select(User.id, User.name).From(User).Where(User.email == 'test@test.com')

build(q)
# (
#     'SELECT id, name FROM "user" WHERE email = $1',
#     ['test@test.com'],
# )

Core concepts

build()

build(query, driver='asyncpg') compiles any statement to (sql, params):

• driver='asyncpg' (default): placeholders $1, $2, ..., params is a list
• driver='psycopg': placeholders %(p1)s, %(p2)s, ..., params is a dict

build(Select(t.id).From(t).Where(t.id > 5))
# ('SELECT id FROM t WHERE id > $1', [5])

build(Select(t.id).From(t).Where(t.id > 5), driver='psycopg')
# ('SELECT id FROM t WHERE id > %(p1)s', {'p1': 5})

Table and columns

Table('name') gives dynamic columns: any attribute access returns a column object. t.STAR is *. Reserved names (user, role) are quoted automatically. Column names are prefixed with the table name automatically when the query references more than one table (multiple FROMs or JOINs).

t = Table('t')
t.any_column          # t.any_column
t.STAR                # t.*
tx = t.As('x')        # aliased table: compiles to "t AS x", columns to "x.col"

All examples below assume t = Table('t'), t2 = Table('t2'). Note: with a single table in FROM the prefix is omitted (SELECT id FROM t); a standalone expression or a multi-table query gets prefixes (t.id).

A table schema can be defined explicitly — this enables IDE completion, reusable filters and refactoring:

class RoleSchema(Table):
    id: int
    name: str
    status: str

    @property
    def status_active(self):  # can also be decorated with functools.cache
        return self.status == Literal('active')

    def name_startswith(self, value: str):
        return self.name.Like(f'{value}%')

Role = RoleSchema('role')
q = Select(Role.id).From(Role).Where(Role.status_active, Role.name_startswith('admin'))

Param / Literal / Raw

Any plain python value used inside a query becomes a parameter automatically. Use wrappers only when you need extra behavior:

Wrapper	Meaning	Example SQL
plain value	parameter (auto-wrapped)	$1
Param(x)	explicit parameter, allows .Cast() / .As()	$1::int AS x
Literal(x)	value inlined into SQL. Not escaped — SQL injection risk, use only with 100% trusted data. Supports int, float, str, bool, None, date, datetime and lists of those	'abc', 15, ARRAY[1, 2], NULL, TRUE
Raw('expr')	raw SQL string inserted as is (e.g. to reference an alias)	expr
NULL	shortcut for Literal(None)	NULL

q = (
    Select(t.STAR, Param(10).Cast('int').As('added')).From(t)
    .Where(
        t.id1 == 1,
        t.id2 != Param(2).Cast('float'),
        t.id3 > Literal(3),
        t.id4 < Literal(4).Cast('numeric'),
    )
)
# SELECT *, $1::int AS added FROM t
# WHERE id1 = $2 AND id2 != $3::float AND id3 > 3 AND id4 < 4::numeric
# params: [10, 1, 2]

Immutability

Every method returns a new object; the original is never modified. Queries and expressions can be safely shared and extended:

base = Select(t.id).From(t)
q1 = base.Where(t.status == 'active')  # base is unchanged
q2 = base.Where(t.status == 'deleted')

Expression modifiers

Available on any expression (column, function, param, literal, operation, select):

Method	SQL
.Cast('int')	expr::int (wraps in brackets when needed)
.As('alias')	expr AS alias
.Distinct()	DISTINCT expr
.Asc() / .Desc()	expr ASC / expr DESC (for ORDER BY)
.NullsFirst() / .NullsLast()	expr NULLS FIRST / expr NULLS LAST

SELECT

Columns, FROM, WHERE

Select(t.id, t.name).From(t)
# SELECT id, name FROM t

Select(t.STAR).From(t)
# SELECT * FROM t

Select(t.id).From(t).Where(t.name == 'x', t.age > 25)  # *args work as AND
# SELECT id FROM t WHERE name = $1 AND age > $2

q = Select(t.id).From(t).Where(t.id > 1)
q = q.Where(t.id < 10)  # chainable: filters are appended
# SELECT id FROM t WHERE id > $1 AND id < $2

Select(t.id).From(t).AddColumns(t.name, t.age)  # append columns to existing select
# SELECT id, name, age FROM t

q.GetColumns() returns a tuple of output column names (alias, column name or None): useful to zip query results with names.

JOIN

Join (inner) / LeftJoin / RightJoin / FullJoin / CrossJoin and lateral variants JoinLateral / LeftJoinLateral / CrossJoinLateral. The ON condition can be any expression or True (compiles to ON TRUE); CrossJoin takes no condition.

sq = Select(t2.name).From(t2).Where(t2.id == t.id).Subquery('sq')
q = (
    Select(t.id).From(t)
    .Join(t2, t2.id == t.id)
    .LeftJoin(t2, And(t2.id == t.id, t2.status == 'active'))
    .JoinLateral(sq, True)
    .LeftJoinLateral(sq, sq.name != 'test')
)
# SELECT t.id FROM t
# JOIN t2 ON t2.id = t.id
# LEFT JOIN t2 ON t2.id = t.id AND t2.status = $1
# JOIN LATERAL (SELECT name FROM t2 WHERE id = t.id) AS sq ON TRUE
# LEFT JOIN LATERAL (SELECT name FROM t2 WHERE id = t.id) AS sq ON sq.name != $2
# params: ['active', 'test']

Select(t.id).From(t).RightJoin(t2, t2.id == t.id)
# SELECT t.id FROM t RIGHT JOIN t2 ON t2.id = t.id

Select(t.id).From(t).FullJoin(t2, t2.id == t.id)
# SELECT t.id FROM t FULL JOIN t2 ON t2.id = t.id

Select(t.id).From(t).CrossJoin(t2)
# SELECT t.id FROM t CROSS JOIN t2

f = F.unnest(t.tags).As('x(tag)')
Select(t.id, f.tag).From(t).CrossJoinLateral(f)
# SELECT t.id, x.tag FROM t CROSS JOIN LATERAL UNNEST(t.tags) AS x(tag)

Table-series join (FROM table, UNNEST(...))

A function can be used as a FROM item alongside tables. Give it an alias with a column list x(col) and reference its columns as attributes:

f = F.unnest(t.tags).As('x(tag)')
Select(t.id, f.tag).From(t, f)
# SELECT t.id, x.tag FROM t, UNNEST(t.tags) AS x(tag)

f = F.unnest(Param([1, 2]).Cast('int[]'), Param(['a', 'b']).Cast('text[]')).As('x(a, b)')
Select(f.a, f.b).From(f)
# SELECT a, b FROM UNNEST($1::int[], $2::text[]) AS x(a, b)

GROUP BY / HAVING

Select(F.count('*')).From(t).GroupBy(t.status).Having(F.count('*') > 10)
# SELECT COUNT(*) FROM t GROUP BY status HAVING COUNT(*) > $1

# GROUP BY by output alias — use Raw
Select(t.id.As('xyz')).From(t).GroupBy(Raw('xyz'))
# SELECT id AS xyz FROM t GROUP BY xyz

# column ordinals: plain ints are inlined (NOT params)
Select(t.id, t.name, F.count('*')).From(t).GroupBy(1, 2)
# SELECT id, name, COUNT(*) FROM t GROUP BY 1, 2

GroupBy can be set only once; Having is chainable (works as AND).

GROUPING SETS / ROLLUP / CUBE

from pgmini import GroupingSets

Select(t.brand, t.size, F.sum(t.qty)).From(t).GroupBy(
    GroupingSets((t.brand, t.size), t.brand, ()),
)
# SELECT brand, size, SUM(qty) FROM t GROUP BY GROUPING SETS ((brand, size), (brand), ())

Select(t.brand, t.size, F.sum(t.qty)).From(t).GroupBy(F.rollup(t.brand, t.size))
# SELECT brand, size, SUM(qty) FROM t GROUP BY ROLLUP(brand, size)

Select(t.brand, F.sum(t.qty)).From(t).GroupBy(F.cube(t.brand, t.size))
# SELECT brand, SUM(qty) FROM t GROUP BY CUBE(brand, size)

ORDER BY / LIMIT / OFFSET

Select(t.STAR).From(t).OrderBy(t.id.Desc(), t.name.NullsLast())
# SELECT * FROM t ORDER BY id DESC, name NULLS LAST

Select(t.id).From(t).OrderBy(t.id).OrderBy(t.age)  # chainable: appended
# SELECT id FROM t ORDER BY id, age

# column ordinals: plain ints are inlined (NOT params — a param would sort by a constant)
Select(t.id, t.name).From(t).OrderBy(2, Literal(1).Desc())
# SELECT id, name FROM t ORDER BY 2, 1 DESC

q.OrderBy(None)   # removes ORDER BY
q.Limit(10)       # LIMIT $n  (value becomes a param; Literal(10) inlines it)
q.Limit(None)     # removes LIMIT
q.Offset(20)      # OFFSET $n
q.Offset(None)    # removes OFFSET

DISTINCT / DISTINCT ON

Select(t.id.Distinct()).From(t)
# SELECT DISTINCT id FROM t

Select(t.id, t.status).From(t).DistinctOn(t.status)
# SELECT DISTINCT ON (status) id, status FROM t

UNION / INTERSECT / EXCEPT

Union / UnionAll / Intersect / Except, chainable:

Select(t.id).From(t).Union(Select(t2.id).From(t2))
# SELECT id FROM t UNION SELECT id FROM t2

Select(Literal('x')).Union(Select(Literal('a'))).UnionAll(Select(Literal('b')))
# SELECT 'x' UNION SELECT 'a' UNION ALL SELECT 'b'

To ORDER BY / LIMIT the combined result, wrap the union into a subquery and apply them outside.

Scalar subquery as a column

A Select used as a column is wrapped in brackets automatically:

Select(t.id, Select(t2.id).From(t2).Where(t2.id == t.id).As('other')).From(t)
# SELECT id, (SELECT id FROM t2 WHERE id = t.id) AS other FROM t

Operations

Math operators work natively: +, -, *, /, >, >=, <, <=, ==, !=. == None/True/False compiles to IS; != to IS NOT. Everything else is a method:

Python	SQL
t.col == 1 / t.col != 1	col = $1 / col != $1
t.col == None	col IS $1 (use t.col == NULL for col IS NULL)
t.col.Is(None) / t.col.IsNot(False)	col IS $1 / col IS NOT $1
t.col.In([1, 2, 3])	col IN ($1, $2, $3)
t.col.In(Select(...))	col IN (SELECT ...)
t.col.NotIn(...)	col NOT IN (...)
t.col.Any([1, 2])	col = ANY($1) — single param, faster plan cache than IN
t.col.All([1, 2])	col = ALL($1)
t.col.IsDistinctFrom(x)	col IS DISTINCT FROM ... — null-safe compare
t.col.IsNotDistinctFrom(x)	col IS NOT DISTINCT FROM ...
t.col.NotLike('%x%') / t.col.NotIlike('%x%')	col NOT LIKE $1 / col NOT ILIKE $1
t.col.LikeAny(['a%', '%b'])	col LIKE ANY($1)
t.col.IlikeAny(Literal(['%a%']))	col ILIKE ANY(ARRAY['%a%'])
t.col.Between(1, 2)	col BETWEEN $1 AND $2
t.col.Like('%x%') / t.col.Ilike('%x%')	col LIKE $1 / col ILIKE $1
t.col.Op('->>', 'key')	col ->> $1 — any custom operator
t.col[1]	col[$1] — array index
t.col[2:5], t.col[:5], t.col[4:]	col[$1:$2] — array slice

t.data.Op('#>', ['k1', 'k2'])              # t.data #> $1
t.dt.Op('at time zone', 'UTC')             # t.dt at time zone $1
t.col[3:F.array_length(t.col, 1)]          # t.col[$1:ARRAY_LENGTH(t.col, $2)]
(t.id == 10).As('is_ten')                  # (t.id = $1) AS is_ten

Operations compose: any operation result is itself an expression and supports .Cast(), .As(), comparison, chaining etc.

Logical operators

from pgmini import And, Or, Not, Exists

Select(t.id).From(t).Where(
    t.id.Between(10, 20),
    Or(t.name > 'name', And(t.status == 'active', Not(t.id == 15))),
)
# SELECT id FROM t
# WHERE id BETWEEN $1 AND $2 AND (name > $3 OR (status = $4 AND NOT (id = $5)))

Select(t.STAR).From(t).Where(Not(Exists(
    Select(1).From(t2).Where(t2.id == t.id)
)))
# SELECT * FROM t WHERE NOT (EXISTS (SELECT $1 FROM t2 WHERE id = t.id))

Functions

F (alias Func) builds any function dynamically — F.<name>(*args) compiles to NAME(args). There is no allowlist: any function name works.

F.now()                      # NOW()
F.count('*')                 # COUNT(*)
F.count(t.id.Distinct())     # COUNT(DISTINCT t.id)
F.date_trunc(Literal('day'), t.created)  # DATE_TRUNC('day', t.created)

Window functions: OVER

F.row_number().Over()                                   # ROW_NUMBER() OVER ()
F.count(t.id).Over(partition_by=t.name, order_by=t.age.Desc())
# COUNT(t.id) OVER (PARTITION BY t.name ORDER BY t.age DESC)

F.sum(t.x).Over(order_by=t.id, frame='ROWS BETWEEN 1 PRECEDING AND CURRENT ROW')
# SUM(t.x) OVER (ORDER BY t.id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)

partition_by / order_by take a single expression or an iterable of them. frame is a raw frame clause string; it must start with ROWS, RANGE or GROUPS.

Aggregates: FILTER, ORDER BY, WITHIN GROUP

F.count('*').Where(t.id > 10)                # COUNT(*) FILTER (WHERE t.id > $1)
F.array_agg(t.id).OrderBy(t.id.Desc())       # ARRAY_AGG(t.id ORDER BY t.id DESC)

F.percentile_disc(t.fld).WithinGroup(t.fld2)
# PERCENTILE_DISC(t.fld) WITHIN GROUP (ORDER BY t.fld2)

F.percentile_cont(Literal(0.5)).WithinGroup(t.fld2.Desc()).As('p50')
# PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY t.fld2 DESC) AS p50

Function as a FROM item

f = F.unnest(Literal([1, 2, 3])).As('idx')
Select(f.STAR).From(f)
# SELECT * FROM UNNEST(ARRAY[1, 2, 3]) AS idx

See also table-series join.

CASE / ARRAY

from pgmini import Case, Array, Tuple

Select(Case((t.id == 1, 'first'), (t.id == 2, 'second'), Else='third').As('val')).From(t)
# SELECT CASE WHEN id = $1 THEN $2 WHEN id = $3 THEN $4 ELSE $5 END AS val FROM t

Select(Array([t.id, 5, 7])).From(t)   # SELECT ARRAY[id, $1, $2] FROM t
Tuple([t.id, 5])                      # (t.id, $1)

INSERT

Insert(table, columns) — columns are required (strings or column objects).

from pgmini import Insert

q = (
    Insert(t, columns=(t.name, t.status))
    .Values(
        (Param('some text').Cast('varchar(10)'), 'active'),
        ('other text', Literal('deleted')),
    )
    .Returning(t.STAR)
)
# INSERT INTO t (name, status)
# VALUES ($1::varchar(10), $2), ($3, 'deleted')
# RETURNING t.*

DEFAULT inserts the column default:

from pgmini import DEFAULT

Insert(t, columns=('id', 'name')).Values((1, DEFAULT))
# INSERT INTO t (id, name) VALUES ($1, DEFAULT)

INSERT ... SELECT

q = (
    Insert(t, columns=(t.name, t.status))
    .Select(Select(t.name, t.status).From(t).Where(t.id < 100).Limit(10))
)

# efficient bulk insert of python lists via UNNEST:
values = [(str(i), 'active') for i in range(1_000)]
q = Insert(t, ('name', 'status')).Select(Select(
    F.unnest(Param([name for name, _ in values]).Cast('text[]')),
    F.unnest(Param([status for _, status in values]).Cast('enum_status[]')),
))
# INSERT INTO t (name, status) SELECT UNNEST($1::text[]), UNNEST($2::enum_status[])

ON CONFLICT

OnConflict(*, constraint=None, index_elements=None, index_where=None, do_update=None, do_update_where=None, do_nothing=False). Exactly one of do_update / do_nothing is required. Excluded('col') / Excluded(t.col) references the excluded row in do_update. do_update_where makes the update conditional.

from pgmini import Excluded

Insert(t, (t.id,)).OnConflict(do_nothing=True)
# INSERT INTO t (id) ON CONFLICT DO NOTHING

Insert(t, (t.id,)).OnConflict(constraint='cc_uniq', do_nothing=True)
# INSERT INTO t (id) ON CONFLICT ON CONSTRAINT cc_uniq DO NOTHING

Insert(t, (t.id,)).OnConflict(
    index_elements=(t.id,), index_where=t.id > 0, do_nothing=True,
)
# INSERT INTO t (id) ON CONFLICT (id) WHERE id > $1 DO NOTHING

Insert(t, (t.id,)).OnConflict(do_update={
    'col1': 12,
    t.col2: t.col2 + 5,
    t.col3: Excluded('col8').Cast('int') * 88,
})
# INSERT INTO t (id) ON CONFLICT DO UPDATE
# SET col1 = $1, col2 = t.col2 + $2, col3 = excluded.col8::int * $3

Insert(t, (t.id,)).Values((1,)).OnConflict(
    index_elements=(t.id,),
    do_update={t.cnt: Excluded(t.cnt)},
    do_update_where=t.cnt < 100,
)
# INSERT INTO t (id) VALUES ($1)
# ON CONFLICT (id) DO UPDATE SET cnt = excluded.cnt WHERE t.cnt < $2

UPDATE

from pgmini import Update

Update(t).Set({t.name: 'second'}).Where(t.name == 'first').Returning(t.id)
# UPDATE t SET name = $1 WHERE t.name = $2 RETURNING t.id

Update(t).Set({t.status: t2.status}).From(t2).Where(t2.id == t.id)
# UPDATE t SET status = t2.status FROM t2 WHERE t2.id = t.id

Set takes a dict (keys: column objects or strings). Where is chainable (AND).

RETURNING old / new (PostgreSQL 18+)

Old(col) / New(col) reference the row values before / after the change. Works in RETURNING of INSERT / UPDATE / DELETE; accepts a column object, a column name string or .STAR; composes with expressions.

from pgmini import New, Old

Update(t).Set({t.price: 100}).Where(t.id == 1).Returning(
    Old(t.price),
    New(t.price),
    (New(t.price) - Old(t.price)).As('diff'),
)
# UPDATE t SET price = $1 WHERE t.id = $2
# RETURNING old.price, new.price, new.price - old.price AS diff

Delete(t).Returning(Old(t.STAR))
# DELETE FROM t RETURNING old.*

Insert(t, (t.id,)).Values((1,)).OnConflict(
    index_elements=(t.id,), do_update={t.cnt: t.cnt + 1},
).Returning(Old(t.cnt), New(t.cnt))
# INSERT INTO t (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET cnt = t.cnt + $2
# RETURNING old.cnt, new.cnt

DELETE

from pgmini import Delete

Delete(t).Where(t.id == 25).Returning(t.id)
# DELETE FROM t WHERE t.id = $1 RETURNING t.id

DELETE ... USING (join-delete)

Delete(t).Using(t2).Where(t2.id == t.id, t2.status == 'deleted')
# DELETE FROM t USING t2 WHERE t2.id = t.id AND t2.status = $1

sq = Select(t2.id).From(t2).Subquery('sq')
Delete(t).Using(sq).Where(sq.id == t.id).Returning(t.id)
# DELETE FROM t USING (SELECT id FROM t2) AS sq WHERE sq.id = t.id RETURNING t.id

MERGE

PostgreSQL 15+. Merge(target).Using(source, on) plus WHEN clauses in order; each accepts an optional condition (compiles to WHEN ... AND condition THEN). Available WHEN methods: WhenMatchedUpdate(dict), WhenMatchedDelete(), WhenMatchedDoNothing(), WhenNotMatchedInsert(columns, values), WhenNotMatchedDoNothing(). Returning (PostgreSQL 17+) supports F.merge_action().

from pgmini import Merge

src = Table('src')
q = (
    Merge(t).Using(src, t.id == src.id)
    .WhenMatchedUpdate({t.name: src.name})
    .WhenNotMatchedInsert(('id', 'name'), (src.id, src.name))
)
# MERGE INTO t USING src ON t.id = src.id
# WHEN MATCHED THEN UPDATE SET name = src.name
# WHEN NOT MATCHED THEN INSERT (id, name) VALUES (src.id, src.name)

# source can be a subquery, Values or a CTE; conditions and RETURNING:
q = (
    Merge(t).Using(src, t.id == src.id)
    .WhenMatchedDelete(condition=src.deleted == Literal(True))
    .WhenMatchedUpdate({t.name: src.name})
    .WhenNotMatchedDoNothing()
    .Returning(t.id, F.merge_action())
)
# MERGE INTO t USING src ON t.id = src.id
# WHEN MATCHED AND src.deleted IS TRUE THEN DELETE
# WHEN MATCHED THEN UPDATE SET name = src.name
# WHEN NOT MATCHED THEN DO NOTHING
# RETURNING t.id, MERGE_ACTION()

VALUES as a FROM item

Values(*rows).As('v(col1, col2)') — a table literal usable in FROM, JOIN or as a MERGE source. The alias with the column list is required to reference columns.

from pgmini import Values

v = Values((1, 'a'), (2, 'b')).As('v(id, name)')
Select(v.id, v.name).From(v)
# SELECT id, name FROM (VALUES ($1, $2), ($3, $4)) AS v(id, name)

Select(t.id).From(t).Join(v, v.id == t.id)
# SELECT t.id FROM t JOIN (VALUES ($1)) AS v(id) ON v.id = t.id

Subqueries and CTE (WITH)

Any Select / Insert / Update / Delete has a .Subquery(alias, materialized=False) method. The result object exposes its columns as attributes.

sq = Select(t.id).From(t).Where(t.id < 100).Subquery('sq')

# as a subquery in FROM:
Select(sq.id).From(sq).Where(sq.id > 50)
# SELECT id FROM (SELECT id FROM t WHERE id < $1) AS sq WHERE id > $2

# as a CTE:
With(sq).Select(sq.id).From(sq).Where(sq.id > 50)
# WITH sq AS (SELECT id FROM t WHERE id < $1) SELECT id FROM sq WHERE id > $2

With(*subqueries) accepts several CTEs and starts any statement type: .Select(...), .Insert(table, columns), .Update(table), .Delete(table).

x1 = Select(t.id).From(t).Subquery('x1', materialized=True)
x2 = Select(t2.id).From(t2).Subquery('x2')
With(x1, x2).Select(x1.id, x2.id.As('id2')).From(x1, x2).Where(x1.id == x2.id)
# WITH x1 AS MATERIALIZED (SELECT id FROM t),
# x2 AS (SELECT id FROM t2)
# SELECT x1.id, x2.id AS id2 FROM x1, x2 WHERE x1.id = x2.id

# writable CTE:
sq = Update(t).Set({t.id: t.id2}).Returning(t.id).Subquery('sq')
With(sq).Select(F.count('*')).From(sq)
# WITH sq AS (UPDATE t SET id = t.id2 RETURNING t.id) SELECT COUNT(*) FROM sq

WITH RECURSIVE

With(..., recursive=True). Reference the CTE by its future name via a Table with the same name inside the recursive term:

tree = Table('tree')
sq = (
    Select(Literal(1).As('n'))
    .UnionAll(Select(tree.n + 1).From(tree).Where(tree.n < 10))
    .Subquery('tree')
)
With(sq, recursive=True).Select(sq.n).From(sq)
# WITH RECURSIVE tree AS (SELECT 1 AS n UNION ALL SELECT n + $1 FROM tree WHERE n < $2)
# SELECT n FROM tree

Row locking (FOR UPDATE / FOR SHARE)

Four methods, one per lock strength, with identical signatures (*, of=None, nowait=False, skip_locked=False):

Method	SQL	Typical use
.ForUpdate()	FOR UPDATE	strongest: lock for update/delete
.ForNoKeyUpdate()	FOR NO KEY UPDATE	update of non-key columns; doesn't block FK inserts into child tables
.ForShare()	FOR SHARE	shared read lock
.ForKeyShare()	FOR KEY SHARE	weakest; what FK checks take

• of — lock only rows of the given table(s): a table/alias or an iterable of them
• nowait=True — error immediately instead of waiting for a lock
• skip_locked=True — skip already locked rows
• nowait and skip_locked are mutually exclusive (raises ValueError)

Select(t.id).From(t).ForUpdate()
# SELECT id FROM t FOR UPDATE

Select(t.id).From(t).ForUpdate(skip_locked=True)
# SELECT id FROM t FOR UPDATE SKIP LOCKED

Select(t.id).From(t).ForUpdate(nowait=True)
# SELECT id FROM t FOR UPDATE NOWAIT

Select(t.id).From(t).ForNoKeyUpdate()
# SELECT id FROM t FOR NO KEY UPDATE

Select(t.id).From(t, t2).ForShare(of=t, nowait=True)
# SELECT t.id FROM t, t2 FOR SHARE OF t NOWAIT

t2a = t2.As('x')
Select(t.id).From(t, t2a).ForUpdate(of=(t, t2a), skip_locked=True)
# SELECT t.id FROM t, t2 AS x FOR UPDATE OF t, x SKIP LOCKED

# typical work-queue pattern with CTE:
sq = Select(t.id).From(t).Limit(1).ForUpdate(skip_locked=True).Subquery('sq')
With(sq).Update(t).Set({t.status: 'processing'}).Where(t.id == sq.id)
# WITH sq AS (SELECT id FROM t LIMIT $1 FOR UPDATE SKIP LOCKED)
# UPDATE t SET status = $2 WHERE t.id = sq.id

API cheat sheet

Compact reference of the whole public API (from pgmini import ...):

build(query, driver='asyncpg'|'psycopg') -> (sql, params)

Table(name) -> table; .As(alias); .STAR; .<attr> -> column
Select(*columns)
    .From(*tables) .Join/LeftJoin/RightJoin/FullJoin(item, on) .CrossJoin(item)
    .JoinLateral/LeftJoinLateral(item, on) .CrossJoinLateral(item)
    .Where(*exprs) .GroupBy(*exprs|ints) .Having(*exprs)
    .OrderBy(*exprs|ints|None) .Limit(v|None) .Offset(v|None)
    # plain ints in GroupBy/OrderBy are column ordinals, inlined as literals
    .Distinct via column.Distinct() / .DistinctOn(*exprs)
    .Union/UnionAll/Intersect/Except(select)
    .ForUpdate/.ForNoKeyUpdate/.ForShare/.ForKeyShare(of=None, nowait=False, skip_locked=False)
    .AddColumns(*exprs) .GetColumns() .As(alias) .Cast(type)
    .Subquery(alias, materialized=False)
Insert(table, columns) .Values(*rows) .Select(select)
    .OnConflict(constraint=, index_elements=, index_where=,
                do_update=, do_update_where=, do_nothing=)
    .Returning(*exprs) .Subquery(alias)
Update(table) .Set(dict) .From(*tables) .Where(*exprs) .Returning(*exprs) .Subquery(alias)
Delete(table) .Using(*tables) .Where(*exprs) .Returning(*exprs) .Subquery(alias)
Merge(table) .Using(source, on)
    .WhenMatchedUpdate(dict, condition=None) .WhenMatchedDelete(condition=None)
    .WhenMatchedDoNothing(condition=None)
    .WhenNotMatchedInsert(columns, values, condition=None)
    .WhenNotMatchedDoNothing(condition=None)
    .Returning(*exprs) .Subquery(alias)
With(*subqueries, recursive=False)
    .Select(...) / .Insert(table, columns) / .Update(table) / .Delete(table) / .Merge(table)
Values(*rows).As('v(a, b)') -> (VALUES ...) AS v(a, b) — FROM/JOIN/MERGE source item

Param(value)    -> $1 / %(p1)s
Literal(value)  -> inlined (int/float/str/bool/None/date/datetime/list); NULL = Literal(None)
Raw('sql')      -> inserted as is; DEFAULT = Raw('DEFAULT') for INSERT values
F.<name>(*args) -> function; .Over(partition_by=, order_by=, frame=) .Where(*filter_exprs)
                   .OrderBy(*exprs) .WithinGroup(*order_exprs) .As('x(a, b)') for FROM usage
Case((cond, value), ..., Else=default)
Array([...]) / Tuple([...])
GroupingSets(set1, set2, ...) — each set: expr, iterable or (); also F.rollup(...)/F.cube(...)
And(*exprs) / Or(*exprs) / Not(expr) / Exists(select)
Excluded(col) — excluded.* reference for ON CONFLICT DO UPDATE
Old(col) / New(col) — old.* / new.* references in RETURNING (PostgreSQL 18+)

expression methods (any column/param/literal/function/operation/select):
    == != > >= < <= + - * /  [idx] [start:stop]
    .Is(x) .IsNot(x) .IsDistinctFrom(x) .IsNotDistinctFrom(x)
    .In(seq|select) .NotIn(seq|select)
    .Any(seq|expr) .All(seq|expr) .LikeAny(seq|expr) .IlikeAny(seq|expr)
    .Between(a, b) .Like(x) .Ilike(x) .NotLike(x) .NotIlike(x) .Op('operator', x)
    .Cast(type) .As(alias) .Distinct() .Asc() .Desc() .NullsFirst() .NullsLast()

Notes for query generation:

• plain python values become parameters; Literal inlines; Raw is verbatim
• column names get table prefixes automatically when >1 table is referenced
• Where, Having, OrderBy are chainable and append; GroupBy can be set once
• every method returns a new immutable object

---

Why not sqlalchemy?

• too smart (tries to do everything: from connection/session management, to sql generating and params escaping)
• too complex
• too slow
• mutable (on its core)

It is good for simple projects with simple sql queries. But when your project grows up, your team grows up, sqlalchemy always leads to errors, unnecessary complexity, extra time your team need to spend to learn it, find not obvious bugs etc.

Why not pypika?

While it is much simpler then sqlalchemy, it also requires you to learn their own "sql syntax" which is not always obvious. And by default it uses parameters as literals, so it can lead to sql injections.

---

The library is inspired by Ukraine🇺🇦 (Kyiv is my home) and its brave and free people🔱.

Slava Ukraini, Heroyam slava!