"""after_response — work a use case wants done AFTER its transaction committed and its answer went out.
from erp_platform import after_response
def cancel(caller, sales_order_id):
with engine.begin() as conn:
... # the cancellation, committed when the block ends
after_response(repromise_freed_stock, sales_order_id, conn=conn)
return answer
The rules, on both doors:
* It runs only when the use case RETURNED. A refusal, a crash, a lost version claim (409) — nothing runs.
* `conn=` binds it to that connection's transaction as well: a ROLLBACK drops it, and it waits for a COMMIT.
Without `conn=`, register it after the work is committed (after the `with engine.begin()` block).
* A transient-failure retry (erp_platform.retry) re-runs the use case: what an attempt that failed registered
is dropped with it.
* HTTP: it runs after the response is sent (Starlette's background task; FastAPI's TestClient waits for it, so
a test sees its effect when the call returns). MCP: it starts on a thread as soon as the tool's answer is
decided, and never delays it. `join_after_response()` waits for those (tests, a clean shutdown).
* A failure is logged (`after_response_failed`, the error's TYPE only) and never touches the answer. So what it
does must be safe to lose: leave a durable row the next scheduled run settles, as sales-order's re-promise does.
Outside a door (a script, a scheduled job) there is no response to wait for: an unbound job, or one whose
transaction already ended, starts on a thread at once. Inside an open transaction outside a door, wrap the
unit of work in `with response_scope():` so the job runs when it ends cleanly.
"""
import contextvars
import json
import logging
import threading
from contextlib import contextmanager
from typing import Any, Callable, Optional
_jobs: contextvars.ContextVar = contextvars.ContextVar("erp_after_response", default=None)
_threads: list = []
_lock = threading.Lock()
log = logging.getLogger("erp_platform.after_response")
class _Job:
__slots__ = ("fn", "args", "kw", "state")
def __init__(self, fn: Callable, args: tuple, kw: dict):
self.fn, self.args, self.kw, self.state = fn, args, kw, "ready"
def _bind(job: _Job, conn: Any) -> None:
from sqlalchemy import event
from erp_platform.repository import _sa_connection
sa = _sa_connection(conn)
if sa is None or not sa.in_transaction():
return # no transaction open on it: nothing left to wait for
job.state = "pending"
def committed(_c):
if job.state == "pending":
job.state = "ready"
def rolled_back(_c):
if job.state == "pending":
job.state = "dropped"
event.listen(sa, "commit", committed)
event.listen(sa, "rollback", rolled_back)
def after_response(fn: Callable, *args: Any, conn: Any = None, **kw: Any) -> None:
"""Run `fn(*args, **kw)` after this use case's transaction committed and its answer was sent."""
job = _Job(fn, args, kw)
if conn is not None:
_bind(job, conn)
jobs = _jobs.get()
if jobs is not None:
jobs.append(job)
return
if job.state == "pending":
raise RuntimeError("after_response(conn=...) inside an open transaction outside a door: wrap the unit of "
"work in `with erp_platform.after.response_scope():` so it runs when that ends")
start([job])
@contextmanager
def response_scope():
"""Collect the after_response jobs registered inside; start them when the block ends without an error."""
jobs: list = []
token = _jobs.set(jobs)
try:
yield jobs
except BaseException:
jobs.clear()
raise
finally:
_jobs.reset(token)
start(jobs)
def collecting(jobs: list):
"""For a door: make `jobs` the list after_response appends to, in the current context."""
return _jobs.set(jobs)
def reset(token) -> None:
_jobs.reset(token)
def ready(jobs: list) -> list:
"""The jobs that may run: committed (or never bound). A job still waiting for a COMMIT is dropped."""
return [j for j in jobs if j.state == "ready"]
def run(jobs: list, use_case: str = "") -> None:
"""Run them in turn. A failure is logged — type only, the message can quote values — and the next one runs."""
for j in jobs:
try:
j.fn(*j.args, **j.kw)
except Exception as e: # noqa: BLE001 — the answer stands
from erp_platform import requestlog
log.warning(json.dumps({"event": "after_response_failed", "request_id": requestlog.request_id(),
"use_case": use_case, "job": getattr(j.fn, "__name__", "job"),
"error": type(e).__name__}))
def start(jobs: list, use_case: str = "") -> Optional[threading.Thread]:
"""Run the ready jobs on a daemon thread (in a copy of this context, so logs name the request)."""
todo = ready(jobs)
if not todo:
return None
ctx = contextvars.copy_context()
from erp_platform import repository
ctx.run(repository.detach) # 0.9.13: a job is not the request: it never reads through its identity map
t = threading.Thread(target=ctx.run, args=(run, todo, use_case), daemon=True, name="erp-after-response")
with _lock:
_threads.append(t)
t.start()
return t
def join_after_response(timeout: Optional[float] = None) -> None:
"""Wait for every after_response thread started so far (tests, and a clean shutdown)."""
while True:
with _lock:
if not _threads:
return
t = _threads.pop()
t.join(timeout)