aflib

aflib

Helper library for business logic inside Airflows plpython3u function bodies.

An Airflows function body is a bare Python snippet with a plpy global. It cannot import another function body, cannot see the HTTP request, and cannot shape the HTTP response — so every body ends up re-implementing the same parameter parsing, the same row reading and the same result envelope. aflib is that plumbing, written once.

Status: early, and deliberately so. Nothing enters this library until it has run on a real Airflows instance — written first as a plain function body, deployed, called, and observed. A version ships only capabilities that have each run there, which is why the numbers start so low.

Install

Airflows resolves Python packages from public PyPI by name and version only; there is no local-import, git or private-index route. Declare aflib in the instance's Python packages pinned to an exact version (0.0.12). The exported package list may show the pin as aflib==0.0.12==0.0.12: that is how Airflows stores a pinned version, not an error.

Then import it from a function body. To check which version a body really imports, return aflib.__version__ from it.

What 0.0.12 gives you

Batch triggers. Airflows can declare a statement-level trigger (each: STATEMENT) but not its transition tables, so the statement trigger cannot see which rows changed. An AFTER ROW trigger records each key; the AFTER STATEMENT trigger reads the whole set once and runs one set-based check or effect for the batch:

# AFTER ROW trigger body
from aflib import trigger
trigger.collect(plpy, TD)            # key="id"; on DELETE the key of TD["old"]
return None

# AFTER STATEMENT trigger body
from aflib import trigger
ids = trigger.take(plpy, TD)         # sorted keys of this table, event and trigger depth
if ids:
    ...                              # a RAISE here rolls back the whole statement
return None

• take consumes what it returns. A statement that changed nothing gives [] (the statement trigger still fires). A statement run from inside a trigger is its own batch, one trigger depth deeper, and never sees the outer statement's keys.
• The keys live in the session temp table aflib_trigger_keys (tbl text, event text, depth int, key bigint), created on first use ON COMMIT DROP: a rolled-back statement leaves no keys, and nothing reaches the next transaction on a pooled connection. tbl is schema.table. A plpgsql row trigger can feed the same table.
• collect refuses to run outside an AFTER ROW trigger (a later BEFORE trigger could still skip the row) and refuses a key that is not an integer; take runs only in an AFTER STATEMENT trigger.
• Cost on a real instance: about 4 ms for a 100-row INSERT and 18 ms for 1,000 rows, with the collector and two statement triggers on.

What 0.0.11 gives you

• Errors answered with the platform's own messages. errors.to_fail(exc, log=log, plpy=plpy) matches a database error against the instance's error messages (the regex and translated text an Airflows administrator maintains, which the platform otherwise applies only in the browser) and returns that message's name as the code and its text, in English unless locale="es_ES" (or another configured locale) is passed. $n groups and $label(...) placeholders are filled from the error and from the entity and field labels; a field or entity without a label shows its name. The error's own detail, which quotes row values, is used to match and never returned. Without a matching message, or without plpy, the 0.0.10 classification applies.
• Db.execute(..., expect=n) — any other number of rows touched raises db.RowCountMismatch, an internal error for the caller. The write has happened by then: inside transaction.atomic() or at the end of the call it rolls back.
• aflib.testing — the plpy double for testing function bodies off the platform. FakePlpy with seed(pattern, rows), seed_write(pattern, nrows) and seed_error(pattern, exc); spi_error(sqlstate, message, ...), business_error(id, message) and custom_error(code, message) build errors with the class, sqlstate and spidata observed on a real instance. A SQLSTATE that has not been observed is refused rather than guessed. Import it from tests only.

try:
    ...
except Exception as exc:
    return errors.to_fail(exc, log=log, plpy=plpy, locale=p_locale or "en_US")

What 0.0.10 gives you

• Db.execute(sql, args=None, argtypes=None) — run a write and get back the number of rows it inserted, updated or deleted. On the platform a write without RETURNING holds no rows, so len() of its result is always 0; the count is nrows(), and that is what this returns. A write that needs values back adds RETURNING and goes through query(): db.query("INSERT ... RETURNING id", [...], [...]).scalar("id").
• Constraint violations reach the caller as a readable code. errors.to_fail now maps a unique violation to duplicate, and a CHECK violation, a too-long text or an out-of-range number to invalid_value, each with a fixed message. Not-null and foreign-key violations stay internal_error: either is as likely a bug in the body as bad input. The log line names the table, column and constraint, never the row values the database puts in its own detail.
• params.as_int_list(value, name=None, required=False) — a list of ids from a typed int[] (a Python list) or from a query string, where ?p_ids=1,2,3 arrives as the one text '1,2,3'. Comma-separated only; an empty element or a NULL is refused, not dropped.
• Business rules raised by a plpgsql trigger reach the caller as written. A trigger that does RAISE EXCEPTION 'ERROR_<id>: <message>' gives the caller code <id> and that message. Any other RAISE without the ERROR_ prefix stays internal_error.
• Every log line carries a trace, the same for every line of one call and of the functions it calls through SQL: the app.trace setting when the edge set one with SET LOCAL, otherwise the backend pid plus the transaction's start time. The pid alone is not enough, because a pooled connection serves many calls. Read once, on the first line.
• log.timed(event, **fields) — a block whose line, written when it ends, carries duration_ms; ok: false and level error if it raised. The exception is not caught.
• params.as_choice(value, allowed, ...) — an exact match against a fixed set.
• params.exactly_one_of(**named) — the name of the one non-blank argument, or BadParameter.

All of it ran on a real instance before release, exact source, called by a non-admin user.

from aflib import Db, errors, logging, ok, params

db = Db(plpy)
log = logging.logger(plpy, source="Sales.svc_mark_paid")
try:
    ids = params.as_int_list(p_ids, name="p_ids", required=True)
    n = db.execute('UPDATE "Sales"."Invoice" SET paid = true WHERE id = ANY($1)',
                   [ids], ["int[]"])
    return ok({"updated": n})
except Exception as exc:
    return errors.to_fail(exc, log=log)

What 0.0.9 gives you

• config.get(plpy, name, required=False, secret=False) — a platform variable, read with the name bound as a query parameter. The platform's own get_env_variable pastes the name into the SQL text, so a name containing ' breaks or changes the query. A missing variable is None; required=True raises config.MissingConfig, which errors.to_fail reports as a plain internal error (the variable's name goes to the log, not to the caller). secret=True returns a config.Secret: its str, repr and format say Secret('***'), json refuses it, and reveal() is the only way to the value.
• http.request(method, url, *, timeout, headers=None, json_body=None, data=None) — urllib with a mandatory timeout: nothing else can stop a body blocked on a socket, and the socket has none by default. A 4xx/5xx comes back as a Response (status, ok, body, text, json()); raise_for_status() and every network failure raise http.HttpError, whose text names the method, host and status but never the path or query, so a token in the URL cannot reach a log. Only http:// and https:// URLs. Not built on the requests the platform bundles: its exceptions copy the full URL.
• identity.unscoped(plpy, role="modelsadmin") — run a block as another role and SET ROLE back to the caller in finally, also when the block raises. Never RESET ROLE: the session user is the connection pool's superuser login. Take authorization decisions before the block.
• trigger.changed ignores the recursion flag under both spellings, update_by_trigger and updated_by_trigger.

All three ran on a real instance on 2026-10-20 before release, exact source, called by a non-admin user.

from aflib import config, http, identity

base = config.get(plpy, "IAM_URL", required=True)
password = config.get(plpy, "IAM_ADMIN_PASSWORD", required=True, secret=True)
r = http.request("POST", base + "/token", timeout=10, json_body={"password": password.reveal()})
r.raise_for_status()

with identity.unscoped(plpy):
    plpy.execute("...")      # as modelsadmin; back to the caller afterwards

config queries as the calling user, so the variables must be readable by that user's role.

What 0.0.8 fixes

Batch.run accepts a predicate whose column name contains the letters _t. id_tuition IS NOT NULL, start_time < now() and is_test = false are predicates. 0.0.7 refused them, because the guard searched for _t as a substring. _t.col and "_t".col are still refused: that alias exists only on the outer UPDATE.

What 0.0.7 gives you

Everything in 0.0.6, plus three capabilities that ran on a real instance on 2026-10-01 before they were extracted:

• Batch(plpy).run(...) — one committed chunk. The pending predicate is placed in the outer WHERE and in the subquery, and the result carries done, remaining and seconds.
• workflow.start / status / stuck — insert a PENDING workflow instance and its variables, read the newest instance for a business key, and list instances that are still PENDING or IN_PROGRESS after an interval.
• Outbox(plpy, outbox=..., delivery=..., subscription=...).emit / pump — one event plus a pending delivery per active subscriber, then a chunk of deliveries. Each subscriber runs inside plpy.subtransaction(). Failures retry with attempts, next_attempt_at and last_error, and dead-letter after 3 attempts.

from aflib import workflow
from aflib.events import Outbox

workflow.start(plpy, "WFP_cadena", "case-58", {
    "SomeSchema.some_function.p_id": ("INTEGER", 42),
})

box = Outbox(plpy, outbox="app.outbox", delivery="app.delivery", subscription="app.sub")
box.emit("order.placed", {"id": 42}, 42)
box.pump(deliver, limit=50)

deliver(subscriber, outbox_id, attempt) does the work and raises to fail the attempt. Table names are identifiers you wrote, not request data.

What 0.0.6 gives you

0.0.1–0.0.3 (read/result, errors/logging, transaction.atomic) plus what the 2026-09 platform probes showed, plus two verified fixes:

• params.as_bool accepts a genuine Python bool as well as the true/false/1/0/t/f text vocabulary. A boolean customParameter on the non-HTTP path binds as True/False; converting that through as_bool no longer raises.
• logging treats source as reserved, the same way it already treats level and event. Passing source= as a field raises rather than silently re-labelling the emitting function.
• authz.can(plpy, "Schema.Entity", "UPDATE") and authz.snapshot(plpy) — table GRANTs via has_table_privilege. Not RLS. Snapshot before SET ROLE.
• trigger.changed(TD) / event / when — a plpython trigger can write TD["new"] and return "MODIFY". Compare OLD/NEW; ignore search.
• functions.call(plpy, "Schema.fn", args, argtypes) — compose through SQL with Postgres bind types. Text binds to a typed signature do not resolve.
• identity.current_user(plpy) — SELECT current_user. Not a User id.

Core API (0.0.1–0.0.3, still in 0.0.12)

from aflib import Db, errors, fail, logging, ok, params

db = Db(plpy)
log = logging.logger(plpy, source="Economic.receipt_cancel")

try:
    receipt_id = params.as_int(p_id_receipt, name="p_id_receipt", required=True)

    row = db.query(
        'SELECT id, amount FROM "Economic"."Receipt" WHERE id = $1',
        [receipt_id], ["integer"],
    ).one()

    if row is None:
        return fail("not_found", "No receipt with that id")

    return ok(row)
except Exception as exc:
    return errors.to_fail(exc, log=log)

The try is not optional, and except Exception is not laziness. See Never let an exception escape below — both are load-bearing.

• Db(plpy) — plpy is injected, never imported, because the function body is the only composition root available. db.query(...) returns a Rows you narrow with .one() (None for no row, and it refuses more than one rather than silently taking the first), .rows() for all of them, or .scalar(column) for a single named value.
• params.as_int / as_bool / as_decimal / as_date / as_json — every HTTP parameter arrives as text and must be declared text, so every body converts. These raise BadParameter, which is the caller's fault and should become a 4xx-shaped failure rather than an uncaught exception.
• ok(data) / fail(code, message) — a body's only output is its return value, which the platform wraps as [{"result": …}]. There is no status code and no header, so success and failure both have to be expressed inside the value.
• codec.dumps / loads — JSON that raises on anything it cannot serialize instead of stringifying it. numeric columns arrive as decimal.Decimal and are encoded as strings, never floats, so exactness survives the round trip.
• errors.to_fail(exc, log=log) — converts any exception into a failure envelope, keeping the caller's message and the log's detail as two separate strings. A BadParameter is described to the caller in full, because it is about their own input; anything else becomes a generic internal_error whose detail goes only to the log. It returns the envelope even if logging itself fails, because a broken log sink must not turn a handled failure back into an escaping one.
• logging.logger(plpy, source=…) — one JSON object per line, level and event always present, written through plpy.log. It never raises, even when handed a value that cannot be encoded: the line is most needed in the error path, so a single exotic field must not be able to silence it.
• transaction.atomic(plpy) — a savepoint that also rolls back a returned failure. See A returned failure does not roll back — this is a narrow helper for one specific trap, not a unit of work.

Platform constraints worth knowing before you build on it

These are observed on a live instance, not inferred from documentation. They are the reason the library has the shape it does.

• Every HTTP-facing parameter must be declared text. Declare integer and the platform binds varchar at call time, no candidate signature matches, and the endpoint fails with function does not exist. A boolean parameter binds true/false, but ?p_flag=maybe is HTTP 500 Cannot cast to boolean before the body runs. Convert with params inside the body.
• Function bodies compose only through SQL — SELECT "Schema"."fn"(…). There is no shared module and no package-local import, so a helper another body needs is a database function, not a Python one.
• A body cannot see the request or shape the response. Its globals are exactly GD, SD, args, its declared parameters and plpy: no request object, no undeclared query parameters, no body, no headers. It cannot set a status code, a header or a redirect.
• jsonb is never decoded — it always arrives as str, even through plpy.prepare. numeric arrives as decimal.Decimal and bytea as bytes; neither is JSON-serializable by default, and numeric is every money column.
• Never cache request-scoped state in GD or SD. Consecutive calls land on different pooled backends, and the connection role varies per caller.

Never let an exception escape

An unhandled exception's message is returned to the caller, and it includes internal detail that the caller should not see. A function body cannot influence that response: it is built outside the database from the driver's error message, and there is no setting that trims it. The only way to control what a caller receives is to not raise — catch everything and return a failure envelope instead. That is what errors.to_fail is for, and it is why every example here wraps its work in a try.

except Exception is required rather than merely convenient. plpy.Error does not inherit from plpy.SPIError, so the narrower except plpy.SPIError catches neither plpy.error() nor an ordinary Python exception — only their common ancestor covers all three.

Two related facts worth knowing before you design around logging:

• Of plpython's five informational sinks, only log and warning are written. debug, info and notice return successfully and are discarded, because log_min_messages defaults to WARNING and those three rank below it while log ranks above. A level that vanishes is worse than one that does not exist, which is why aflib.logging puts severity in the payload and writes everything through plpy.log.

• The platform logs every call's parameter values in plaintext before the body runs. So never pass a secret as a function parameter — it is recorded no matter how careful the body is — and do not log parameter values yourself, since that cannot reduce the exposure and only makes it look deliberate.

• A callee's error message reaches its caller unredacted. Function bodies compose through SQL, and when a callee raises, its message crosses to the caller intact while its exception type does not. So a business outcome has to travel as a returned value — ok/fail — and never as an exception, and nothing sensitive may ever go into an exception message.

A returned failure does not roll back

A function body is one transaction it does not control. plpy.commit() fails with invalid transaction termination, because a function invoked inside a query cannot terminate the transaction it is running in. What a body can do is subdivide that transaction with savepoints via plpy.subtransaction(), and those work exactly as you would hope: they roll back on any exception, Python or SQL alike, the original exception continues unchanged, nesting rolls back only the inner block, and a trigger's writes roll back together with the row that fired them.

Two further rules complete the picture: an uncaught exception discards the whole call, and a normal return commits the whole call.

That last one is the trap, because it applies to a returned failure too:

with plpy.subtransaction():
    db.query("INSERT …")
    return fail("not_allowed", "…")     # the INSERT is COMMITTED

plpy.subtransaction() reacts to exceptions, and a business failure is not an exception — it cannot be, since an exception's type does not survive crossing a function boundary. So the most natural code silently half-applies the change. transaction.atomic exists for exactly this:

from aflib import errors, transaction

try:
    with transaction.atomic(plpy) as tx:
        db.query("INSERT …")
        if not allowed:
            tx.reject("not_allowed", "You may not do that")   # rolls back
        tx.succeed({"id": 7})
    return tx.outcome
except Exception as exc:
    return errors.to_fail(exc, log=log)

reject() ends the block and rolls back; outcome is read afterwards. A scope that ends without succeed() or reject() raises rather than returning an empty success for work that may have half happened.

It is deliberately not a unit of work and cannot commit anything — the name describes the guarantee it provides (the block is all-or-nothing), not machinery it does not have.

Tests

uv run pytest

The unit tests run against a fake plpy, which verifies control flow and nothing transactional. Anything involving a savepoint is proven on the platform, not here.

License

MIT — see LICENSE.