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. One version corresponds to one proven capability, 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 it in models/pythonPackages/pythonPackages.airflows:

pythonPackage aflib==0.0.1

Then import it from a function body.

What 0.0.1 gives you

from aflib import BadParameter, Db, fail, ok, params

db = Db(plpy)

try:
    receipt_id = params.as_int(p_id_receipt, name="p_id_receipt", required=True)
except BadParameter as exc:
    return fail("invalid_param", str(exc))

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)

• 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.

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 — it cannot be invoked at all. This is not a casting inconvenience; the endpoint is simply unreachable. Convert inside the body instead.
• 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.

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.