From a4e54e9008a9ddaab4a7e61564b7d923bc66fa37 Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 17:54:41 -0400 Subject: [PATCH 1/5] feat: add generic WebhookChannel with optional HMAC body signing Add WebhookChannel/WebhookConfig: posts a service-agnostic JSON failure payload via urllib (stdlib only), with optional X-Signature-SHA256 header when a shared secret is configured. Add manual_webhook_inspect.py as the webhook counterpart to manual_pipeline_inspect.py. --- README.md | 27 +++ src/processes/__init__.py | 2 + src/processes/_webhook_internals.py | 107 +++++++++++ src/processes/notification_channels.py | 43 +++++ src/processes/webhook_config.py | 29 +++ tests/manual_tests/manual_webhook_inspect.py | 184 +++++++++++++++++++ tests/test_notification_channels.py | 42 ++++- tests/test_webhook_channel.py | 156 ++++++++++++++++ 8 files changed, 589 insertions(+), 1 deletion(-) create mode 100644 src/processes/_webhook_internals.py create mode 100644 src/processes/webhook_config.py create mode 100644 tests/manual_tests/manual_webhook_inspect.py create mode 100644 tests/test_webhook_channel.py diff --git a/README.md b/README.md index 10cf367..3939ab3 100644 --- a/README.md +++ b/README.md @@ -342,6 +342,33 @@ substring instead. This is useful for deep-debugging code that runs through several layers of internal libraries or wrappers, where the default outermost-user-frame would land too high up the call stack. +### `WebhookChannel` + +```python +WebhookChannel( + webhook_config: WebhookConfig, +) +``` + +```python +WebhookConfig( + url: str, + headers: dict[str, str] = {}, # merged with default Content-Type: application/json + timeout: int = 5, + secret: str | None = None, # HMAC-SHA256 signs the body when set +) +``` + +POSTs a generic JSON payload to `url` on `logging.ERROR` and above — +`task_name`, `function`, `args`, `kwargs`, `exception`, `traceback`, +`downstream_impact`, `traced_vars`, and `traced_vars_location`. Not coupled +to any specific service (Slack, Discord, etc.); subclass and override +`_WebhookFormatter._build_payload` to reshape the payload for one. + +If `secret` is set, the request carries an `X-Signature-SHA256` header with +the hex-encoded `hmac.new(secret, body, hashlib.sha256)` digest of the JSON +body, so receivers can verify the payload wasn't tampered with. +
diff --git a/src/processes/__init__.py b/src/processes/__init__.py index c1fe5a5..535262a 100644 --- a/src/processes/__init__.py +++ b/src/processes/__init__.py @@ -14,10 +14,12 @@ ) from .notification_channels import EmailChannel as EmailChannel from .notification_channels import NotificationChannel as NotificationChannel +from .notification_channels import WebhookChannel as WebhookChannel from .process import Process as Process from .task import Task as Task from .task import TaskDependency as TaskDependency from .task import TaskResult as TaskResult +from .webhook_config import WebhookConfig as WebhookConfig try: __version__ = _v("processes") diff --git a/src/processes/_webhook_internals.py b/src/processes/_webhook_internals.py new file mode 100644 index 0000000..cd32f71 --- /dev/null +++ b/src/processes/_webhook_internals.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import hashlib +import hmac +import json +import logging +import urllib.request +from typing import Any + +from ._error_data import _ErrorContextFormatter, _ErrorData +from .webhook_config import WebhookConfig + +_SIGNATURE_HEADER = "X-Signature-SHA256" + + +class _WebhookFormatter(_ErrorContextFormatter): + """Pure renderer: builds a generic JSON payload from ``record.task_context``.""" + + def format(self, record: logging.LogRecord) -> str: + """Render a log record as a JSON payload string. + + Parameters + ---------- + record : logging.LogRecord + The record being formatted. + + Returns + ------- + str + A JSON-encoded object describing the task failure. + """ + error = self._error_data(record) + return json.dumps(self._build_payload(error)) + + def _build_payload(self, error: _ErrorData) -> dict[str, Any]: + """Build the JSON-serializable payload dict from ``_ErrorData``. + + Subclasses targeting a specific webhook service can override this + to reshape the payload, while reusing ``format`` and the rest of + the channel/handler machinery. + + Parameters + ---------- + error : _ErrorData + Typed failure context for the record being formatted. + + Returns + ------- + dict[str, Any] + JSON-serializable payload. + """ + return { + "task_name": error.task_name, + "function": error.function, + "args": repr(error.args), + "kwargs": repr(error.kwargs), + "exception": error.exception, + "traceback": error.traceback_str, + "downstream_impact": list(error.downstream_impact), + "traced_vars": error.traced_vars, + "traced_vars_location": error.traced_vars_location, + } + + +class _WebhookHandler(logging.Handler): + """Internal handler that POSTs formatted log records as JSON.""" + + def __init__(self, config: WebhookConfig) -> None: + super().__init__() + self._config = config + + def emit(self, record: logging.LogRecord) -> None: + try: + body = self.format(record).encode("utf-8") + headers = {"Content-Type": "application/json", **self._config.headers} + if self._config.secret is not None: + digest = hmac.new( + self._config.secret.encode("utf-8"), body, hashlib.sha256 + ).hexdigest() + headers[_SIGNATURE_HEADER] = digest + + request = urllib.request.Request( + self._config.url, data=body, headers=headers, method="POST" + ) + with urllib.request.urlopen(request, timeout=self._config.timeout): + pass + except Exception: + self.handleError(record) + + +def _build_task_webhook_handler(config: WebhookConfig) -> _WebhookHandler: + """Create a fully configured webhook handler. + + Parameters + ---------- + config : WebhookConfig + Webhook transport configuration for the handler. + + Returns + ------- + _WebhookHandler + A handler at ``logging.ERROR`` level with a ``_WebhookFormatter``. + """ + handler = _WebhookHandler(config) + handler.setFormatter(_WebhookFormatter()) + handler.setLevel(logging.ERROR) + return handler diff --git a/src/processes/notification_channels.py b/src/processes/notification_channels.py index c5c1c5a..8081e8a 100644 --- a/src/processes/notification_channels.py +++ b/src/processes/notification_channels.py @@ -5,7 +5,9 @@ from ._email_internals import _build_task_email_handler from ._logfile_formatting import _TaskLogfileFormatter +from ._webhook_internals import _build_task_webhook_handler from .email_config import HTMLEmailStyle, SMTPConfig +from .webhook_config import WebhookConfig class NotificationChannel(ABC): @@ -144,3 +146,44 @@ def frame_filter(self) -> str | None: The configured ``traced_vars_frame_filter``, or ``None``. """ return self.style.traced_vars_frame_filter + + +class WebhookChannel(NotificationChannel): + """Notification channel that POSTs a JSON alert to a webhook URL on task failure. + + The JSON payload is built generically from the task's failure context + (function, args/kwargs, exception, traceback, downstream impact, traced + variables), so it can be consumed directly or transformed by downstream + relays (e.g. Slack/Discord/Teams webhook adapters, custom alerting + servers). It is not coupled to any specific service. + + Attributes + ---------- + webhook_config : WebhookConfig + Webhook transport configuration for the alert. + + Parameters + ---------- + webhook_config : WebhookConfig + Webhook transport configuration for the alert. + """ + + def __init__(self, webhook_config: WebhookConfig): + self.webhook_config = webhook_config + + def build_handler(self, task_name: str) -> logging.Handler: + """Build a JSON webhook handler. + + Parameters + ---------- + task_name : str + Name of the task the handler will be attached to. Unused by + this channel, accepted for interface consistency. + + Returns + ------- + logging.Handler + A handler at ``logging.ERROR`` level that POSTs a JSON payload + describing the failure for each error log record. + """ + return _build_task_webhook_handler(self.webhook_config) diff --git a/src/processes/webhook_config.py b/src/processes/webhook_config.py new file mode 100644 index 0000000..a593e1c --- /dev/null +++ b/src/processes/webhook_config.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass +class WebhookConfig: + """Transport configuration for generic webhook error alerts. + + Attributes + ---------- + url : str + Destination URL the JSON payload is POSTed to. + headers : dict[str, str] + Additional HTTP headers sent with the request. ``Content-Type`` + defaults to ``"application/json"`` if not overridden here. + Defaults to ``{}``. + timeout : int + Request timeout in seconds. Defaults to ``5``. + secret : str | None + Shared secret used to HMAC-SHA256 sign the JSON request body. When + set, the hex digest is sent in the ``X-Signature-SHA256`` header. + ``None`` disables signing. Defaults to ``None``. + """ + + url: str + headers: dict[str, str] = field(default_factory=dict) + timeout: int = 5 + secret: str | None = None diff --git a/tests/manual_tests/manual_webhook_inspect.py b/tests/manual_tests/manual_webhook_inspect.py new file mode 100644 index 0000000..ac4cf3b --- /dev/null +++ b/tests/manual_tests/manual_webhook_inspect.py @@ -0,0 +1,184 @@ +"""Manual end-to-end inspection of ``WebhookChannel`` alerts. + +This is the webhook counterpart to ``manual_pipeline_inspect.py`` (which +exercises ``EmailChannel`` against maildev). It is **not** picked up by +pytest — see ``tests/conftest.py`` and the ``python_files = "test_*.py"`` +setting in ``pytest.ini``. Run it by hand: + + python tests/manual_tests/manual_webhook_inspect.py + +What this exercises +-------------------- +A small DAG with one failing task and one cascading-skipped downstream task. +Both ``A_load_config`` (success) and ``B_apply_config`` (failure) are wired +with a ``WebhookChannel`` pointing at a throwaway local HTTP server started +by this script. ``C_notify_done`` depends on ``B_apply_config`` and is +cascade-skipped. + +The local server prints every received request: method, headers (including +the ``X-Signature-SHA256`` header), and the JSON body. It also independently +recomputes the HMAC signature from ``WEBHOOK_SECRET`` and reports whether it +matches, so a single run verifies both the payload shape and the optional +HMAC body-signing end to end. + +Inspect +------- +* The console output for the recomputed-vs-received signature check. +* The printed JSON payload for ``B_apply_config`` — confirm it carries + ``task_name``, ``function``, ``args``, ``kwargs``, ``exception``, + ``traceback``, ``downstream_impact`` (containing ``C_notify_done``), + ``traced_vars`` and ``traced_vars_location``. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import sys +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any + +# Make the in-tree package importable when the script is run directly. +_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +if _PROJECT_ROOT not in sys.path: + sys.path.insert(0, _PROJECT_ROOT) + +from processes import Process, Task, TaskDependency, WebhookChannel, WebhookConfig # noqa: E402 + +WEBHOOK_SECRET = "manual-test-shared-secret" + + +class _WebhookRequestHandler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 (stdlib-mandated name) + length = int(self.headers.get("Content-Length", "0")) + body = self.rfile.read(length) + signature = self.headers.get("X-Signature-SHA256") + + print(f" [server] {self.command} {self.path}") + print(f" [server] Content-Type: {self.headers.get('Content-Type')}") + print(f" [server] X-Signature-SHA256: {signature}") + + expected = hmac.new(WEBHOOK_SECRET.encode("utf-8"), body, hashlib.sha256).hexdigest() + match = "OK" if signature == expected else "MISMATCH" + print(f" [server] signature check: {match} (expected={expected})") + + payload = json.loads(body) + print(" [server] payload:") + print(json.dumps(payload, indent=2)) + + self.send_response(200) + self.end_headers() + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + pass # silence default request logging; we print our own summary above + + +def _start_server() -> tuple[HTTPServer, int]: + server = HTTPServer(("127.0.0.1", 0), _WebhookRequestHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, server.server_address[1] + + +# --------------------------------------------------------------------------- # +# Task functions # +# --------------------------------------------------------------------------- # + + +def load_config(env: str) -> dict[str, Any]: + """Load configuration for the given environment. Always succeeds.""" + print(f" [load_config] env={env!r}") + return {"env": env, "feature_flags": {"new_pricing": True}} + + +def apply_config( + target: str, + config: dict[str, Any] | None = None, + restart: bool = True, +) -> str: + """Apply the loaded configuration. **Fails** — target service rejects it.""" + print(f" [apply_config] target={target!r} restart={restart} config={config!r}") + raise RuntimeError(f"target service {target!r} rejected configuration") + + +def notify_done(*_args: Any, **_kwargs: Any) -> str: + """Notify that the rollout finished. Cascading-skip target.""" + print(" [notify_done] this should never run — FAILED upstream") + return "done-notified" + + +# --------------------------------------------------------------------------- # +# Entry point # +# --------------------------------------------------------------------------- # + + +def main() -> int: + here = os.path.dirname(os.path.abspath(__file__)) + logs_dir = os.path.join(here, "logs") + os.makedirs(logs_dir, exist_ok=True) + + server, port = _start_server() + print(f"webhook server: http://127.0.0.1:{port}/hook") + + webhook = WebhookChannel( + WebhookConfig(url=f"http://127.0.0.1:{port}/hook", secret=WEBHOOK_SECRET) + ) + + dep = TaskDependency + tasks = [ + Task( + name="A_load_config", + log_path=os.path.join(logs_dir, "A_load_config.log"), + func=load_config, + args=("staging",), + channels=[webhook], + ), + Task( + name="B_apply_config", + log_path=os.path.join(logs_dir, "B_apply_config.log"), + func=apply_config, + args=("pricing-service",), + kwargs={"restart": True}, + dependencies=[ + dep( + "A_load_config", + use_result_as_additional_kwargs=True, + additional_kwarg_name="config", + ) + ], + channels=[webhook], + ), + Task( + name="C_notify_done", + log_path=os.path.join(logs_dir, "C_notify_done.log"), + func=notify_done, + dependencies=[dep("B_apply_config")], + channels=[webhook], + ), + ] + + print(f"tasks: {len(tasks)}") + print("-" * 72) + + try: + with Process(tasks) as process: + result = process.run(parallel=False) + finally: + server.shutdown() + + print("-" * 72) + print("passed:") + for name in sorted(result.passed_tasks_results): + print(f" + {name}") + print("failed (includes cascading-skipped):") + for name in sorted(result.failed_tasks): + print(f" - {name}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_notification_channels.py b/tests/test_notification_channels.py index 78466a4..4eb5da2 100644 --- a/tests/test_notification_channels.py +++ b/tests/test_notification_channels.py @@ -15,9 +15,19 @@ import pytest -from processes import EmailChannel, HTMLEmailStyle, NotificationChannel, Process, SMTPConfig, Task +from processes import ( + EmailChannel, + HTMLEmailStyle, + NotificationChannel, + Process, + SMTPConfig, + Task, + WebhookChannel, + WebhookConfig, +) from processes._email_internals import _HTMLEmailFormatter from processes._logfile_formatting import _TaskLogfileFormatter +from processes._webhook_internals import _WebhookFormatter from processes.notification_channels import _FileChannel from .base_test import BaseTest @@ -107,6 +117,36 @@ def test_frame_filter_defaults_to_none(self) -> None: assert channel.frame_filter is None +class TestWebhookChannel(BaseTest): + def _webhook_config(self, **overrides: object) -> WebhookConfig: + defaults: dict[str, object] = {"url": "https://example.test/hook"} + defaults.update(overrides) + return WebhookConfig(**defaults) # type: ignore[arg-type] + + def test_build_handler_defaults_to_error_level(self) -> None: + channel = WebhookChannel(self._webhook_config()) + handler = channel.build_handler("webhook_task") + + assert handler.level == logging.ERROR + assert isinstance(handler.formatter, _WebhookFormatter) + + def test_config_round_trips_onto_channel(self) -> None: + config = self._webhook_config( + headers={"Authorization": "Bearer x"}, timeout=10, secret="shh" + ) + channel = WebhookChannel(config) + + assert channel.webhook_config is config + assert channel.webhook_config.url == "https://example.test/hook" + assert channel.webhook_config.headers == {"Authorization": "Bearer x"} + assert channel.webhook_config.timeout == 10 + assert channel.webhook_config.secret == "shh" + + def test_frame_filter_defaults_to_none(self) -> None: + channel = WebhookChannel(self._webhook_config()) + assert channel.frame_filter is None + + class _RecordingChannel(NotificationChannel): """Test-only channel that captures every record it receives.""" diff --git a/tests/test_webhook_channel.py b/tests/test_webhook_channel.py new file mode 100644 index 0000000..3761cb8 --- /dev/null +++ b/tests/test_webhook_channel.py @@ -0,0 +1,156 @@ +"""Tests for the generic ``WebhookChannel`` handler/formatter internals. + +Covers: + +* ``_WebhookFormatter.format`` renders the canonical ``task_context`` + payload as JSON with the expected keys. +* ``_WebhookHandler.emit`` POSTs the JSON body to the configured URL with + the right method, headers, and timeout. +* Optional HMAC-SHA256 body signing via ``WebhookConfig.secret``. +* Errors during emission are routed through ``handleError`` instead of + propagating. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import logging +from unittest.mock import MagicMock, patch + +from processes import WebhookChannel, WebhookConfig +from processes._webhook_internals import _build_task_webhook_handler, _WebhookFormatter + +from .base_test import BaseTest + + +def _make_record(task_name: str = "demo_task") -> logging.LogRecord: + """Build a LogRecord carrying the canonical task_context payload.""" + record = logging.LogRecord( + name=task_name, + level=logging.ERROR, + pathname=__file__, + lineno=1, + msg="planned failure in %s", + args=(task_name,), + exc_info=None, + ) + record.task_context = { + "task_name": task_name, + "function": "func_demo", + "args": ("a", 1), + "kwargs": {"flag": True}, + "downstream_impact": ["child_a", "child_b"], + "exception": "RuntimeError('boom')", + "traceback_str": "Traceback (most recent call last):\n...\nRuntimeError: boom\n", + "traced_vars": "a = 'a'\nflag = True", + "traced_vars_location": "demo.py:42", + } + return record + + +class TestWebhookFormatter(BaseTest): + def test_format_renders_expected_payload_keys(self) -> None: + formatter = _WebhookFormatter() + payload = json.loads(formatter.format(_make_record())) + + assert payload == { + "task_name": "demo_task", + "function": "func_demo", + "args": "('a', 1)", + "kwargs": "{'flag': True}", + "exception": "RuntimeError('boom')", + "traceback": "Traceback (most recent call last):\n...\nRuntimeError: boom\n", + "downstream_impact": ["child_a", "child_b"], + "traced_vars": "a = 'a'\nflag = True", + "traced_vars_location": "demo.py:42", + } + + +class TestWebhookHandlerEmit(BaseTest): + def _config(self, **overrides: object) -> WebhookConfig: + defaults: dict[str, object] = {"url": "https://example.test/hook"} + defaults.update(overrides) + return WebhookConfig(**defaults) # type: ignore[arg-type] + + def test_emit_posts_json_payload(self) -> None: + handler = _build_task_webhook_handler(self._config()) + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MagicMock() + handler.emit(_make_record()) + + assert mock_urlopen.call_count == 1 + request = mock_urlopen.call_args.args[0] + assert request.full_url == "https://example.test/hook" + assert request.get_method() == "POST" + assert json.loads(request.data)["task_name"] == "demo_task" + assert mock_urlopen.call_args.kwargs["timeout"] == 5 + + def test_default_content_type_header(self) -> None: + handler = _build_task_webhook_handler(self._config()) + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MagicMock() + handler.emit(_make_record()) + + request = mock_urlopen.call_args.args[0] + assert request.get_header("Content-type") == "application/json" + + def test_custom_headers_merge_with_default_content_type(self) -> None: + config = self._config(headers={"Authorization": "Bearer token123"}, timeout=10) + handler = _build_task_webhook_handler(config) + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MagicMock() + handler.emit(_make_record()) + + request = mock_urlopen.call_args.args[0] + assert request.get_header("Content-type") == "application/json" + assert request.get_header("Authorization") == "Bearer token123" + assert mock_urlopen.call_args.kwargs["timeout"] == 10 + + def test_hmac_signature_added_when_secret_set(self) -> None: + config = self._config(secret="shh") + handler = _build_task_webhook_handler(config) + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MagicMock() + handler.emit(_make_record()) + + request = mock_urlopen.call_args.args[0] + expected = hmac.new(b"shh", request.data, hashlib.sha256).hexdigest() + assert request.get_header("X-signature-sha256") == expected + + def test_no_signature_header_when_secret_none(self) -> None: + handler = _build_task_webhook_handler(self._config()) + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MagicMock() + handler.emit(_make_record()) + + request = mock_urlopen.call_args.args[0] + assert request.get_header("X-signature-sha256") is None + + def test_emit_routes_request_errors_through_handle_error(self) -> None: + handler = _build_task_webhook_handler(self._config()) + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.side_effect = OSError("connection refused") + with patch.object(handler, "handleError") as mock_handle_error: + handler.emit(_make_record()) + + mock_handle_error.assert_called_once() + + +class TestWebhookChannelHandlerWiring(BaseTest): + def test_channel_builds_handler_using_webhook_internals(self) -> None: + channel = WebhookChannel(WebhookConfig(url="https://example.test/hook")) + handler = channel.build_handler("webhook_task") + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MagicMock() + handler.emit(_make_record("webhook_task")) + + assert mock_urlopen.call_count == 1 From 722391593d4b2caac3f974f87b2def4eda0e222e Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 18:27:40 -0400 Subject: [PATCH 2/5] refactor: make traced_vars a plain {name: repr(value)} dict _build_traced_vars (formerly _build_traced_vars_html) no longer HTML-escapes results, so _ErrorData.traced_vars / task_context["traced_vars"] is now a presentation-agnostic dict. _HTMLEmailFormatter does the HTML escaping when rendering, and the webhook payload now emits structured JSON instead of an HTML-escaped string. --- src/processes/_email_internals.py | 7 ++++++- src/processes/_error_data.py | 9 +++++---- src/processes/_tb_utils.py | 22 ++++++++++------------ src/processes/task.py | 4 ++-- tests/test_complex_dag_failures.py | 12 +++++++++--- tests/test_email_themes.py | 8 ++++---- tests/test_webhook_channel.py | 4 ++-- 7 files changed, 38 insertions(+), 28 deletions(-) diff --git a/src/processes/_email_internals.py b/src/processes/_email_internals.py index a0fa8cc..2f33ced 100644 --- a/src/processes/_email_internals.py +++ b/src/processes/_email_internals.py @@ -127,6 +127,11 @@ def format(self, record: logging.LogRecord) -> str: f"
  • {html.escape(str(name), quote=True)}
    • " for name in error.downstream_impact ) + traced_vars_html = "\n".join( + html.escape(f"{name} = {value}", quote=True) + for name, value in error.traced_vars.items() + ) + substitutions = dict(_load_language_strings(self._email_language)) substitutions["lang_traced_vars_blurb"] = substitutions.get( "lang_traced_vars_blurb", "" @@ -141,7 +146,7 @@ def format(self, record: logging.LogRecord) -> str: "traceback_before": html.escape(tb_before, quote=True), "traceback_highlight": html.escape(tb_highlight, quote=True), "traceback_after": html.escape(tb_after, quote=True), - "traced_vars": error.traced_vars, + "traced_vars": traced_vars_html, "downstream_items": downstream_items, } ) diff --git a/src/processes/_error_data.py b/src/processes/_error_data.py index 5c96f93..061d3e1 100644 --- a/src/processes/_error_data.py +++ b/src/processes/_error_data.py @@ -25,8 +25,9 @@ class _ErrorData: String representation of the raised exception. Defaults to ``""``. traceback_str : str Full formatted traceback. Defaults to ``""``. - traced_vars : str - Rendered local variables of the traced frame. Defaults to ``""``. + traced_vars : dict[str, str] + Mapping of local variable names to ``repr(value)`` for the traced + frame. Defaults to ``{}``. traced_vars_location : str ``"filename:lineno"`` of the traced frame. Defaults to ``""``. """ @@ -38,7 +39,7 @@ class _ErrorData: downstream_impact: list[str] = field(default_factory=list) exception: str = "" traceback_str: str = "" - traced_vars: str = "" + traced_vars: dict[str, str] = field(default_factory=dict) traced_vars_location: str = "" @@ -69,6 +70,6 @@ def _error_data(self, record: logging.LogRecord) -> _ErrorData: downstream_impact=ctx.get("downstream_impact", []) or [], exception=ctx.get("exception", record.getMessage()), traceback_str=ctx.get("traceback_str", ""), - traced_vars=ctx.get("traced_vars", ""), + traced_vars=ctx.get("traced_vars", {}) or {}, traced_vars_location=ctx.get("traced_vars_location", ""), ) diff --git a/src/processes/_tb_utils.py b/src/processes/_tb_utils.py index 72549c2..c3da37b 100644 --- a/src/processes/_tb_utils.py +++ b/src/processes/_tb_utils.py @@ -1,6 +1,5 @@ from __future__ import annotations -import html import os import sys import traceback @@ -86,8 +85,8 @@ def _resolve_target_tb(exc_tb: Any, frame_filter: str | None) -> Any: return matches[-1] if matches else frames[-1] -def _build_traced_vars_html(exc_tb: Any, frame_filter: str | None) -> str: - """Return HTML-escaped ``name = repr(value)`` lines for the target frame's locals. +def _build_traced_vars(exc_tb: Any, frame_filter: str | None) -> dict[str, str]: + """Return ``{name: repr(value)}`` for the target frame's local variables. Parameters ---------- @@ -99,22 +98,21 @@ def _build_traced_vars_html(exc_tb: Any, frame_filter: str | None) -> str: Returns ------- - str - Newline-separated, HTML-escaped ``name = repr(value)`` lines for the - target frame's locals, or ``""`` if no frame is found. + dict[str, str] + Mapping of local variable names to ``repr(value)`` for the target + frame, in definition order, or ``{}`` if no frame is found. """ target = _resolve_target_tb(exc_tb, frame_filter) if target is None: - return "" + return {} frame = target.tb_frame - lines: list[str] = [] + traced_vars: dict[str, str] = {} for name, value in frame.f_locals.items(): try: - rendered = repr(value) + traced_vars[name] = repr(value) except Exception as exc: - rendered = f"" - lines.append(f"{name} = {rendered}") - return "\n".join(html.escape(line, quote=True) for line in lines) + traced_vars[name] = f"" + return traced_vars def _build_traced_vars_location(exc_tb: Any, frame_filter: str | None) -> str: diff --git a/src/processes/task.py b/src/processes/task.py index 2485e23..ff5c55c 100644 --- a/src/processes/task.py +++ b/src/processes/task.py @@ -9,7 +9,7 @@ import logging -from ._tb_utils import _build_traced_vars_html, _build_traced_vars_location, _format_traceback +from ._tb_utils import _build_traced_vars, _build_traced_vars_location, _format_traceback from .exceptions import CircularDependencyError from .notification_channels import NotificationChannel, _FileChannel @@ -367,7 +367,7 @@ def _build_failure_context( "downstream_impact": downstream_names, "exception": str(exc), "traceback_str": _format_traceback(exc), - "traced_vars": _build_traced_vars_html(exc_tb, self._frame_filter), + "traced_vars": _build_traced_vars(exc_tb, self._frame_filter), "traced_vars_location": _build_traced_vars_location(exc_tb, self._frame_filter), } diff --git a/tests/test_complex_dag_failures.py b/tests/test_complex_dag_failures.py index f78089f..0a75f32 100644 --- a/tests/test_complex_dag_failures.py +++ b/tests/test_complex_dag_failures.py @@ -210,11 +210,17 @@ def make_task(name: str, deps, fail: bool = False) -> Task: expected_downstream = {n for n in skipped_task_names if name in _ancestors_of(n, tasks)} assert set(ctx["downstream_impact"]) == expected_downstream - serialized = repr(ctx) - assert "<" not in serialized and ">" not in serialized, ( - f"Task {name} task_context contains HTML markers: {serialized}" + assert isinstance(ctx["traced_vars"], dict), ( + f"Task {name} task_context['traced_vars'] must be a plain " + f"{{name: repr(value)}} dict, got {type(ctx['traced_vars'])}" ) + serialized = repr(ctx) + for entity in ("<", ">", "&", "'", """): + assert entity not in serialized, ( + f"Task {name} task_context contains an HTML entity ({entity}): {serialized}" + ) + # OUTCOME #4 — Mocked Alert Validation assert mock_smtp_class.call_count == len(failing_task_names), ( f"smtplib.SMTP should be instantiated {len(failing_task_names)} times, " diff --git a/tests/test_email_themes.py b/tests/test_email_themes.py index 3795756..95e1da9 100644 --- a/tests/test_email_themes.py +++ b/tests/test_email_themes.py @@ -27,7 +27,7 @@ from processes import EmailChannel, HTMLEmailStyle, Process, SMTPConfig, Task from processes._email_internals import _HTMLEmailFormatter from processes._tb_utils import ( - _build_traced_vars_html, + _build_traced_vars, _build_traced_vars_location, _format_traceback, ) @@ -216,7 +216,7 @@ def _deep() -> None: { "exception": str(exc), "traceback_str": _format_traceback(exc), - "traced_vars": _build_traced_vars_html(exc.__traceback__, frame_filter), + "traced_vars": _build_traced_vars(exc.__traceback__, frame_filter), "traced_vars_location": _build_traced_vars_location( exc.__traceback__, frame_filter ), @@ -258,7 +258,7 @@ def _bad_hook(d: dict) -> None: { "exception": str(exc), "traceback_str": _format_traceback(exc), - "traced_vars": _build_traced_vars_html(exc.__traceback__, frame_filter), + "traced_vars": _build_traced_vars(exc.__traceback__, frame_filter), "traced_vars_location": _build_traced_vars_location( exc.__traceback__, frame_filter ), @@ -293,7 +293,7 @@ def _inner() -> None: { "exception": str(exc), "traceback_str": _format_traceback(exc), - "traced_vars": _build_traced_vars_html(exc.__traceback__, None), + "traced_vars": _build_traced_vars(exc.__traceback__, None), "traced_vars_location": _build_traced_vars_location(exc.__traceback__, None), } ) diff --git a/tests/test_webhook_channel.py b/tests/test_webhook_channel.py index 3761cb8..ce55f0b 100644 --- a/tests/test_webhook_channel.py +++ b/tests/test_webhook_channel.py @@ -44,7 +44,7 @@ def _make_record(task_name: str = "demo_task") -> logging.LogRecord: "downstream_impact": ["child_a", "child_b"], "exception": "RuntimeError('boom')", "traceback_str": "Traceback (most recent call last):\n...\nRuntimeError: boom\n", - "traced_vars": "a = 'a'\nflag = True", + "traced_vars": {"a": "'a'", "flag": "True"}, "traced_vars_location": "demo.py:42", } return record @@ -63,7 +63,7 @@ def test_format_renders_expected_payload_keys(self) -> None: "exception": "RuntimeError('boom')", "traceback": "Traceback (most recent call last):\n...\nRuntimeError: boom\n", "downstream_impact": ["child_a", "child_b"], - "traced_vars": "a = 'a'\nflag = True", + "traced_vars": {"a": "'a'", "flag": "True"}, "traced_vars_location": "demo.py:42", } From 774620d776b64ecef6808b8bbb94aa94f88cc788 Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 18:31:52 -0400 Subject: [PATCH 3/5] feat: include traced variables in task logfiles _TaskLogfileFormatter now appends a "Traced vars:" block listing each traced local variable's repr, now that traced_vars is a plain dict rather than a pre-escaped HTML string. --- src/processes/_logfile_formatting.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/processes/_logfile_formatting.py b/src/processes/_logfile_formatting.py index 8b34d0c..1f15650 100644 --- a/src/processes/_logfile_formatting.py +++ b/src/processes/_logfile_formatting.py @@ -11,8 +11,8 @@ class _TaskLogfileFormatter(_ErrorContextFormatter): """Plain-text formatter for task logfiles. On failure, appends the same failure context shown in the HTML email - (function, args, kwargs, downstream impact, traced-vars location, - traceback) as readable text. + (function, args, kwargs, downstream impact, traced variables and their + location, traceback) as readable text. """ def __init__(self) -> None: @@ -51,6 +51,9 @@ def format(self, record: logging.LogRecord) -> str: f"Downstream impact: {', '.join(error.downstream_impact) or '-'}", f"Traced vars location: {error.traced_vars_location or '-'}", ] + if error.traced_vars: + lines.append("Traced vars:") + lines.extend(f" {name} = {value}" for name, value in error.traced_vars.items()) if error.traceback_str: lines.append("") lines.append(error.traceback_str.rstrip("\n")) From 4f6f75b1e1abf5ebe502c0276f88d9cb075ac70f Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 18:49:36 -0400 Subject: [PATCH 4/5] feat: add extra_payload to WebhookConfig for service-specific routing keys Lets callers merge static top-level keys (e.g. Telegram chat_id, Slack channel/username) into the JSON body without subclassing the formatter. --- README.md | 6 ++++++ src/processes/_webhook_internals.py | 13 ++++++++++--- src/processes/webhook_config.py | 7 +++++++ tests/test_webhook_channel.py | 27 +++++++++++++++++++++++++++ 4 files changed, 50 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 3939ab3..1bd6cd6 100644 --- a/README.md +++ b/README.md @@ -356,6 +356,7 @@ WebhookConfig( headers: dict[str, str] = {}, # merged with default Content-Type: application/json timeout: int = 5, secret: str | None = None, # HMAC-SHA256 signs the body when set + extra_payload: dict[str, Any] = {}, # extra top-level keys merged into the JSON body ) ``` @@ -365,6 +366,11 @@ POSTs a generic JSON payload to `url` on `logging.ERROR` and above — to any specific service (Slack, Discord, etc.); subclass and override `_WebhookFormatter._build_payload` to reshape the payload for one. +`extra_payload` keys are merged into the JSON body and take precedence over +the generic fields on collision — useful for service-specific routing data +(e.g. a Telegram `chat_id` or a Slack `channel`/`username` override) without +subclassing. + If `secret` is set, the request carries an `X-Signature-SHA256` header with the hex-encoded `hmac.new(secret, body, hashlib.sha256)` digest of the JSON body, so receivers can verify the payload wasn't tampered with. diff --git a/src/processes/_webhook_internals.py b/src/processes/_webhook_internals.py index cd32f71..267b309 100644 --- a/src/processes/_webhook_internals.py +++ b/src/processes/_webhook_internals.py @@ -16,6 +16,10 @@ class _WebhookFormatter(_ErrorContextFormatter): """Pure renderer: builds a generic JSON payload from ``record.task_context``.""" + def __init__(self, extra_payload: dict[str, Any] | None = None) -> None: + super().__init__() + self._extra_payload = extra_payload or {} + def format(self, record: logging.LogRecord) -> str: """Render a log record as a JSON payload string. @@ -27,10 +31,13 @@ def format(self, record: logging.LogRecord) -> str: Returns ------- str - A JSON-encoded object describing the task failure. + A JSON-encoded object describing the task failure, merged with + any configured ``extra_payload`` keys (which take precedence on + collision). """ error = self._error_data(record) - return json.dumps(self._build_payload(error)) + payload = {**self._build_payload(error), **self._extra_payload} + return json.dumps(payload) def _build_payload(self, error: _ErrorData) -> dict[str, Any]: """Build the JSON-serializable payload dict from ``_ErrorData``. @@ -102,6 +109,6 @@ def _build_task_webhook_handler(config: WebhookConfig) -> _WebhookHandler: A handler at ``logging.ERROR`` level with a ``_WebhookFormatter``. """ handler = _WebhookHandler(config) - handler.setFormatter(_WebhookFormatter()) + handler.setFormatter(_WebhookFormatter(extra_payload=config.extra_payload)) handler.setLevel(logging.ERROR) return handler diff --git a/src/processes/webhook_config.py b/src/processes/webhook_config.py index a593e1c..13e1a04 100644 --- a/src/processes/webhook_config.py +++ b/src/processes/webhook_config.py @@ -1,6 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field +from typing import Any @dataclass @@ -21,9 +22,15 @@ class WebhookConfig: Shared secret used to HMAC-SHA256 sign the JSON request body. When set, the hex digest is sent in the ``X-Signature-SHA256`` header. ``None`` disables signing. Defaults to ``None``. + extra_payload : dict[str, Any] + Additional top-level keys merged into the JSON payload, taking + precedence over the generic fields if names collide. Useful for + service-specific routing fields (e.g. a Telegram ``chat_id``). + Defaults to ``{}``. """ url: str headers: dict[str, str] = field(default_factory=dict) timeout: int = 5 secret: str | None = None + extra_payload: dict[str, Any] = field(default_factory=dict) diff --git a/tests/test_webhook_channel.py b/tests/test_webhook_channel.py index ce55f0b..4c259a4 100644 --- a/tests/test_webhook_channel.py +++ b/tests/test_webhook_channel.py @@ -67,6 +67,20 @@ def test_format_renders_expected_payload_keys(self) -> None: "traced_vars_location": "demo.py:42", } + def test_extra_payload_merged_into_payload(self) -> None: + formatter = _WebhookFormatter(extra_payload={"chat_id": "12345", "channel": "#alerts"}) + payload = json.loads(formatter.format(_make_record())) + + assert payload["chat_id"] == "12345" + assert payload["channel"] == "#alerts" + assert payload["task_name"] == "demo_task" + + def test_extra_payload_overrides_colliding_keys(self) -> None: + formatter = _WebhookFormatter(extra_payload={"task_name": "overridden"}) + payload = json.loads(formatter.format(_make_record())) + + assert payload["task_name"] == "overridden" + class TestWebhookHandlerEmit(BaseTest): def _config(self, **overrides: object) -> WebhookConfig: @@ -133,6 +147,19 @@ def test_no_signature_header_when_secret_none(self) -> None: request = mock_urlopen.call_args.args[0] assert request.get_header("X-signature-sha256") is None + def test_extra_payload_from_config_is_merged_into_body(self) -> None: + config = self._config(extra_payload={"chat_id": "12345"}) + handler = _build_task_webhook_handler(config) + + with patch("processes._webhook_internals.urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MagicMock() + handler.emit(_make_record()) + + request = mock_urlopen.call_args.args[0] + body = json.loads(request.data) + assert body["chat_id"] == "12345" + assert body["task_name"] == "demo_task" + def test_emit_routes_request_errors_through_handle_error(self) -> None: handler = _build_task_webhook_handler(self._config()) From 8053f6807cc8005239355e977a514b28794b3ede Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 18:53:38 -0400 Subject: [PATCH 5/5] chore: ignore local PR_DESCRIPTION.md --- .gitignore | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 34b9592..47e92eb 100644 --- a/.gitignore +++ b/.gitignore @@ -180,4 +180,6 @@ mail_config.toml /logs/ logs/* -logs \ No newline at end of file +logs + +PR_DESCRIPTION.md \ No newline at end of file