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

Then import it from a function body.

What 0.0.2 gives you

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.

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.

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.

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.