Metadata-Version: 2.5
Name: aflib
Version: 0.0.11
Summary: Helper library for business logic inside Airflows plpython function bodies
License: MIT
License-File: LICENSE
Requires-Python: >=3.11
Description-Content-Type: text/markdown

# aflib

Helper library for business logic inside [Airflows](https://flows.ninja) `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.11`). The exported package list may show the pin as
`aflib==0.0.11==0.0.11`: 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.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.

```python
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.

```python
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.

```python
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.

```python
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.11)

```python
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](#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](#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:

```python
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:

```python
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`.
