🇺🇦 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.

Installation

pip install pgmini

Quick start

frompgminiimportSelect, Table, buildUser=Table('user') # columns are dynamic: User.<anything> is a columnq=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_columnt.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:

classRoleSchema(Table):
id: intname: strstatus: str@propertydefstatus_active(self): # can also be decorated with functools.cachereturnself.status==Literal('active')
defname_startswith(self, value: str):
returnself.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 unchangedq2=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 tSelect(t.STAR).From(t)
# SELECT * FROM tSelect(t.id).From(t).Where(t.name=='x', t.age>25) # *args work as AND# SELECT id FROM t WHERE name = $1 AND age > $2q=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 < $2Select(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.idSelect(t.id).From(t).FullJoin(t2, t2.id==t.id)
# SELECT t.id FROM t FULL JOIN t2 ON t2.id = t.idSelect(t.id).From(t).CrossJoin(t2)
# SELECT t.id FROM t CROSS JOIN t2f=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 RawSelect(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

frompgminiimportGroupingSetsSelect(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 LASTSelect(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 DESCq.OrderBy(None) # removes ORDER BYq.Limit(10) # LIMIT $n (value becomes a param; Literal(10) inlines it)q.Limit(None) # removes LIMITq.Offset(20) # OFFSET $nq.Offset(None) # removes OFFSET

DISTINCT / DISTINCT ON

Select(t.id.Distinct()).From(t)
# SELECT DISTINCT id FROM tSelect(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 t2Select(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 #> $1t.dt.Op('at time zone', 'UTC') # t.dt at time zone $1t.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

frompgminiimportAnd, Or, Not, ExistsSelect(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

CASE / ARRAY

frompgminiimportCase, Array, TupleSelect(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 tSelect(Array([t.id, 5, 7])).From(t) # SELECT ARRAY[id, $1, $2] FROM tTuple([t.id, 5]) # (t.id, $1)

INSERT

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

frompgminiimportInsertq= (
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:

frompgminiimportDEFAULTInsert(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') foriinrange(1_000)]
q=Insert(t, ('name', 'status')).Select(Select(
F.unnest(Param([nameforname, _invalues]).Cast('text[]')),
F.unnest(Param([statusfor_, statusinvalues]).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.

frompgminiimportExcludedInsert(t, (t.id,)).OnConflict(do_nothing=True)
# INSERT INTO t (id) ON CONFLICT DO NOTHINGInsert(t, (t.id,)).OnConflict(constraint='cc_uniq', do_nothing=True)
# INSERT INTO t (id) ON CONFLICT ON CONSTRAINT cc_uniq DO NOTHINGInsert(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 NOTHINGInsert(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 * $3Insert(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

frompgminiimportUpdateUpdate(t).Set({t.name: 'second'}).Where(t.name=='first').Returning(t.id)
# UPDATE t SET name = $1 WHERE t.name = $2 RETURNING t.idUpdate(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.

frompgminiimportNew, OldUpdate(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 diffDelete(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

frompgminiimportDeleteDelete(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 = $1sq=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().

frompgminiimportMergesrc=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.

frompgminiimportValuesv=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 UPDATESelect(t.id).From(t).ForUpdate(skip_locked=True)
# SELECT id FROM t FOR UPDATE SKIP LOCKEDSelect(t.id).From(t).ForUpdate(nowait=True)
# SELECT id FROM t FOR UPDATE NOWAITSelect(t.id).From(t).ForNoKeyUpdate()
# SELECT id FROM t FOR NO KEY UPDATESelect(t.id).From(t, t2).ForShare(of=t, nowait=True)
# SELECT t.id FROM t, t2 FOR SHARE OF t NOWAITt2a=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!

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
pgmini		pgmini
tests		tests
.gitignore		.gitignore
LICENSE		LICENSE
README.md		README.md
poetry.lock		poetry.lock
pyproject.toml		pyproject.toml

Latest commit

History

Folders and files

Repository files navigation