diff --git a/README.md b/README.md index 4916819..241f8d2 100644 --- a/README.md +++ b/README.md @@ -188,7 +188,6 @@ tasks = [ notify_slack, LOG_DIR / "notify_slack.log", dependencies=[TaskDependency("build_report", use_result_as_additional_args=True)], - channels=[EmailChannel(smtp)], ), Task( "archive_report", @@ -200,6 +199,7 @@ tasks = [ with Process(tasks) as process: result = process.run(parallel=True) + result.notify(EmailChannel(smtp), only_errors=True) # one report email for the failed task(s) print("passed:", sorted(result.successes)) # archive_report, build_report, compute_revenue, compute_stock, fetch_inventory, fetch_orders @@ -209,7 +209,7 @@ print("report:", result.successes["build_report"].result) # daily-report | revenue=59.50 stock=262.50 ``` -The failing `notify_slack` task does **not** abort the run. `archive_report` is a sibling of the failed task (both depend on the successful `build_report`), so it runs unaffected — the rest of the workflow is not blackholed by one broken step. The HTML email handler also fires on the `notify_slack` task, paging on-call with the full traceback and the list of downstream tasks that were skipped because of it. +The failing `notify_slack` task does **not** abort the run. `archive_report` is a sibling of the failed task (both depend on the successful `build_report`), so it runs unaffected — the rest of the workflow is not blackholed by one broken step. Calling `result.notify(EmailChannel(smtp), only_errors=True)` then delivers a single report email covering the failed task with its traceback and the downstream tasks that were skipped because of it. @@ -230,7 +230,7 @@ Task( args: tuple = (), kwargs: dict | None = None, dependencies: list[TaskDependency] | None = None, - channels: list[NotificationChannel] | None = None, + traced_vars_frame_filter: str | None = None, timeout: float | None = None, retries: int | None = 0, retry_on: tuple[type[Exception], ...] | None = None, @@ -238,9 +238,9 @@ Task( ``` - `name` — unique within the `Process`; no spaces. -- `log_path` — the file this task logs to (INFO level, format `%(asctime)s - %(name)s - %(levelname)s - %(message)s`); wired internally into a file `NotificationChannel`. `None` (the default) means no file logging; if no other `channels` are configured either, a `NullHandler` is attached. +- `log_path` — the file this task logs to (INFO level, format `%(asctime)s - %(name)s - %(levelname)s - %(message)s`), with the structured failure context appended on error. `None` (the default) means no file logging; a `NullHandler` is attached instead. A `Task` does not send notifications — error notification is delegated to `ProcessExecutionReport.notify`. - `func` — the callable; receives `func(*args, **kwargs)` after result-injection. -- `channels` — additional `NotificationChannel`s attached to the task's logger. Use `EmailChannel(smtp_config, style=None)` to fire an HTML email on `logging.ERROR`; body includes `task_name`, `function`, `args`, `kwargs`, and `downstream_impact`. `style` defaults to `HTMLEmailStyle()` (modern, neutral, English). +- `traced_vars_frame_filter` — substring selecting which traceback frame's locals are captured into the failure context (and thus into both the logfile and any report notification). `None` (default) captures the outermost user frame. - `timeout` — seconds allowed per attempt; `None` means no limit. When the timeout fires the underlying thread is detached (Python threading limitation). - `retries` — additional attempts after the first failure; `0` or `None` means a single attempt. Defaults to `0`. - `retry_on` — tuple of exception types that trigger a retry. When `retries >= 1` and `retry_on` is `None`, defaults to `(ConnectionError, TimeoutError)` at call time. @@ -311,6 +311,21 @@ for name, entry in report.entries.items(): print(f"{name} failed after {entry.attempts} attempt(s): {entry.error.exception}") ``` +Deliver the report through one or more channels with `notify`: + +```python +report.notify( + *channels: ReportChannel, + only_errors: bool = False, # restrict the payload to ERRORED tasks + tasks: list[str] | None = None, # restrict to these task names (case-insensitive) + show_warnings: bool = True, # warn (not raise) if a channel fails +) +``` + +`only_errors` and `tasks` compose (both filters apply). A failing channel never +aborts the others. Built-in channels: `EmailChannel` (HTML email) and +`WebhookChannel` (JSON POST). + ### `ErrorData` ```python @@ -347,56 +362,60 @@ SMTPConfig( ```python HTMLEmailStyle( - style="modern", # classic | modern | compact palette="neutral", # neutral | catppuccin | neobones | slate language="en", # en | es | pt | fr | de | it - traced_vars_frame_filter=None, # substring to pick the traced frame | None ) ``` -### `NotificationChannel` +### `ReportContent` ```python -NotificationChannel # ABC: subclass and implement build_handler(task_name) -> logging.Handler +ReportContent( + show_traceback=True, # include each failure's full traceback + show_traced_vars=True, # include each failure's traced local variables +) ``` -If `log_path` is set, every `Task` attaches an internal file channel built from it. Extra channels passed via `channels` are attached on top of it. If neither `log_path` nor `channels` is set, the task's logger gets a `NullHandler`. +Per-channel selection of how much per-task detail a report notification includes. +Pass the same instance to several channels for uniform content, or give each its +own. ### `EmailChannel` ```python EmailChannel( smtp_config: SMTPConfig, - style: HTMLEmailStyle | None = None, # defaults to HTMLEmailStyle() + style: HTMLEmailStyle | None = None, # defaults to HTMLEmailStyle() + content: ReportContent | None = None, # defaults to ReportContent() ) ``` -Fires a styled HTML email on `logging.ERROR` and above. - -All fields are optional — omit `HTMLEmailStyle` entirely to use the defaults. +A `ReportChannel` that delivers a finished report as a styled HTML email when +passed to `report.notify(...)`. The body lists every task with its status and, +for each failure, the exception, traceback, downstream impact, and traced +variables (subject to `content`). #### Traced Variables -On failure, the email body includes the local variables of the **outermost -user frame in the traceback** — i.e. the last frame that is not inside -`site-packages` or your virtualenv. A `file:line` reference next to the -section shows exactly where those values were captured. - -`traced_vars_frame_filter` lets you point this at a different frame: set it -to a path substring (e.g. one of your own package or module names) to -capture locals from the outermost frame whose filename contains that -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. +On failure, each task captures the local variables of the **outermost user +frame in the traceback** — i.e. the last frame that is not inside +`site-packages` or your virtualenv — into both its logfile and the report email. +A `file:line` reference next to the section shows exactly where those values +were captured. Point this at a different frame per task with +`Task(traced_vars_frame_filter=…)`. ### `WebhookChannel` ```python WebhookChannel( webhook_config: WebhookConfig, + content: ReportContent | None = None, # defaults to ReportContent() ) ``` +A `ReportChannel` that POSTs the finished report as JSON when passed to +`report.notify(...)`. + ```python WebhookConfig( url: str, @@ -408,11 +427,12 @@ WebhookConfig( ) ``` -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. +Transport configuration for `WebhookChannel`. When the report is delivered, +POSTs a generic JSON payload to `url` — an `entries` object mapping each task +name to its `status`, `function`, `elapsed_seconds`, `attempts`, and (for +failures) an `error` block with `exception`, `traceback`, `downstream_impact`, +and `traced_vars` (subject to `ReportContent`). Not coupled to any specific +service (Slack, Discord, etc.). `extra_payload` keys are merged into the JSON body and take precedence over the generic fields on collision — useful for service-specific routing data diff --git a/arquitectura.md b/arquitectura.md deleted file mode 100644 index 4ea114d..0000000 --- a/arquitectura.md +++ /dev/null @@ -1,339 +0,0 @@ -# Plan de Arquitectura — Reorganización de dominio y comunicación - -> Documento de **planificación**. No introduce cambios de código por sí mismo. -> Define los problemas actuales, la arquitectura objetivo y un plan por fases -> aprobables de forma independiente. - ---- - -## 1. Problemas actuales - -### Problema 1 — Inversión de dependencia dominio → infraestructura (y el workaround de import circular) - -`task.py` es el núcleo de dominio, pero importa la infraestructura de comunicación: - -```python -# task.py:17 -from .notification_channels import NotificationChannel, _FileChannel -``` - -`Task.__init__` construye los handlers de logging en el momento de la -construcción (`task.py:351-365`). Esto crea la cadena de importación: - -``` -task.py → notification_channels.py → _email_internals.py → task.py ✗ - (necesita TaskStatus) -``` - -El ciclo se "resuelve" hoy con un **workaround**: los internals de comunicación -no importan `TaskStatus`, sino que comparan contra el string del valor del enum: - -```python -# _webhook_internals.py -_STATUS_SUCCESS = "success" -... -if entry.status.value == _STATUS_SUCCESS: -``` - -```python -# _email_internals.py -_STATUS_ERRORED = "errored" -``` - -Esto es frágil (se rompe en silencio si cambia el `.value` del enum) y es el -síntoma visible de que la dirección de la dependencia está invertida: el dominio -no debería arrastrar la infraestructura. - -### Problema 2 — Canales con doble contrato y transporte duplicado - -`EmailChannel` y `WebhookChannel` implementan **dos** interfaces con modelos de -entrega distintos: - -| | Task (`NotificationChannel`) | Report (`ReportChannel`) | -|---|---|---| -| Disparo | Evento de logging (`LogRecord`) | Llamada explícita | -| Momento | Durante la ejecución | Al finalizar | -| Entidad | `logging.LogRecord` (streaming) | `ProcessExecutionReport` (one-shot) | -| Quién envía | El `Handler` (autónomo, en `emit`) | El canal (directo, en `send_report`) | - -Las dos interfaces son legítimamente diferentes. El problema **no** es que existan -ambas, sino que el **transporte está duplicado**: - -- **Email**: `_HTMLEmailHandler.emit` (`_email_internals.py:170-192`) abre SMTP, - arma el `MIMEText`, autentica, hace `sendmail` y `quit`. La función nueva - `send_report_email` repite exactamente esa misma secuencia. -- **Webhook**: ya fue parcialmente deduplicado — `_WebhookHandler.emit` y - `send_report_webhook` comparten `_post_json`. Falta cerrar la simetría del - lado email. - -### Problema 3 — Duplicación entre `Process` y `Task` (cierre de handlers y propiedad partida) - -El mismo bucle de cierre de handlers aparece **idéntico** dos veces en `process.py`: - -```python -# process.py:267-269 (remove_task) y process.py:376-379 (close_loggers) -for handler in list(task.logger.handlers): - handler.close() - task.logger.removeHandler(handler) -``` - -El problema de fondo es de **propiedad partida**: `Task` *crea* su logger y sus -handlers (`task.py:348-365`), pero `Process` los *destruye* metiendo la mano en -`task.logger.handlers`. La gestión del ciclo de vida del logger está repartida -entre dos clases. - -### Problema 4 — Namespace plano con roles muy distintos - -`src/processes/` mezcla en un solo nivel 2 módulos de dominio con ~10 de -comunicación: - -``` -process.py task.py execution_report.py exceptions.py ← dominio / orquestación -notification_channels.py _email_internals.py _webhook_internals.py -email_config.py webhook_config.py _error_data.py _tb_utils.py -_logfile_formatting.py exception_html_formatter.py html_logging.py ← comunicación -``` - -A medida que se añadan canales (Slack, SMS, etc.) el directorio raíz seguirá -creciendo con archivos cuyo rol no es evidente desde su ubicación. - ---- - -## 2. Invariantes (lo que NO debe romperse) - -1. **Definir un `Task` sigue siendo trivial para el usuario.** La firma pública - se mantiene: - ```python - Task("t", func, channels=[EmailChannel(smtp_cfg)]) - ``` - El usuario nunca construye handlers ni transportes a mano. -2. **API pública estable.** Todo lo exportado hoy en `processes/__init__.py` - (`Task`, `Process`, `TaskStatus`, `ErrorData`, `EmailChannel`, - `ProcessExecutionReport`, etc.) se sigue importando con `from processes import ...`. -3. **Rutas de módulo documentadas estables.** `processes.html_logging` - (`HTMLSMTPHandler`) está expuesto en `docs/reference`. Si el archivo se - reubica, se deja un *shim* de re-export en la ruta antigua para no romper - `from processes.html_logging import ...`. -4. **Sin regresiones.** `pytest`, `mypy` y `ruff` quedan verdes tras cada fase. - ---- - -## 3. Arquitectura objetivo - -### 3.1 Capas - -El dominio define **puertos** (clases abstractas); la comunicación provee -**adaptadores** (implementaciones concretas). Dentro de la comunicación se -separan tres responsabilidades hoy entremezcladas: - -``` - ┌──────────────────────────────────────────┐ - DOMINIO │ Task · Process · ProcessExecutionReport │ - (puertos) │ TaskStatus · TaskResult · ErrorData │ - │ NotificationChannel · ReportChannel (ABC)│ - └───────────────────┬──────────────────────┘ - │ depende solo de abstracciones - ┌───────────────────▼──────────────────────┐ - COMUNICACIÓN │ Channels (adaptadores user-facing) │ - (adaptadores) │ EmailChannel · WebhookChannel │ - │ Render (entidad → payload) │ - │ HTML email · JSON webhook │ - │ Transport (payload → destino) │ - │ _SMTPTransport · _WebhookTransport │ - └──────────────────────────────────────────┘ -``` - -- **Render** = "qué entregar": convierte un `LogRecord` (Task) o un - `ProcessExecutionReport` (Report) en un cuerpo (HTML / JSON). -- **Transport** = "cómo entregarlo": una sola implementación de SMTP-send y una - sola de HTTP-POST, usadas tanto por el handler de streaming como por el envío - one-shot del reporte. **Aquí se elimina la duplicación del Problema 2.** -- **Channel** = objeto de configuración que el usuario pasa; cablea render + - transport para sus dos roles. Al quedar render/transport como capas - separadas, el canal es delgado y puede implementar ambos puertos sin duplicar - nada (la doble capacidad pasa a ser una conveniencia honesta, no un smell). - -### 3.2 Paquete `comms/` - -Se agrupa toda la comunicación en un subpaquete plano (sin sub-subdirectorios: -`transports/` + `rendering/` con 2 archivos cada uno sería sobre-ingeniería): - -``` -src/processes/ - __init__.py # API pública (sin cambios de exports) - process.py # Process, ProcessRunner - task.py # Task - task_types.py # TaskStatus, TaskResult, TaskDependency ← LEAF (sin imports de comms) - error_data.py # ErrorData (dataclass puro) ← LEAF - _tb_utils.py # utilidades de traceback ← LEAF (dominio) - execution_report.py # ProcessExecutionReport, TaskReportEntry - exceptions.py - html_logging.py # shim de compatibilidad → comms - comms/ - __init__.py # re-exporta canales y ABCs públicos - base.py # PUERTOS: NotificationChannel, ReportChannel, ReportContent - channels.py # ADAPTADORES: EmailChannel, WebhookChannel, _FileChannel - config.py # SMTPConfig, HTMLEmailStyle, WebhookConfig - _smtp.py # _SMTPTransport (envío unificado) + handler de streaming - _webhook.py # _WebhookTransport (POST unificado) + handler de streaming - _email_render.py # _HTMLEmailFormatter (task) + render HTML del reporte - _webhook_render.py # _WebhookFormatter (task) + payload JSON del reporte - _error_context.py # _ErrorContextFormatter (lee LogRecord) - _logfile_formatting.py - exception_html_formatter.py - themes/ # styles / palettes / languages -``` - -**Por qué `ErrorData` y `_tb_utils` salen a leaf y `_ErrorContextFormatter` no:** -`ErrorData` es un dataclass de valor (contexto de fallo) que `TaskResult` referencia -— es dominio. `_tb_utils` construye ese contexto y lo usa `Task` -(`task.py:473-474`) — también dominio. En cambio `_ErrorContextFormatter` es un -`logging.Formatter` que lee un `LogRecord`: eso es comunicación y se queda en -`comms/`. Hoy `_error_data.py` los mezcla; se separan. - -### 3.3 Cómo se rompe el ciclo (sin workaround) - -Tras mover los tipos de valor a leaves sin dependencia de comms: - -``` -task.py → task_types, error_data, _tb_utils, comms.base (ABC), comms (_FileChannel) -comms/* → task_types, error_data, _tb_utils, comms.base, config -execution_report.py → task_types, error_data (+ TYPE_CHECKING: comms.base.ReportChannel) -process.py → task, task_types, error_data, execution_report, exceptions -``` - -`comms/` nunca importa `task.py` ni `process.py`. Los renderers importan -`TaskStatus` directamente desde `task_types` (leaf). **El ciclo desaparece y se -borran `_STATUS_SUCCESS` / `_STATUS_ERRORED`.** - -> **Nota de honestidad sobre el alcance:** esto rompe el *ciclo* y elimina el -> workaround, pero `task.py` sigue dependiendo de `comms.base` para el ABC -> `NotificationChannel` y de `_FileChannel`. **No** es una inversión total -> dominio→infra. Se adopta deliberadamente el enfoque **ports & adapters -> pragmático**: las clases abstractas `NotificationChannel` / `ReportChannel` se -> consideran *puertos propiedad del dominio*, y `comms.base` se mantiene como un -> verdadero leaf que no importa ningún internal de `comms`. Para una librería de -> este tamaño esto es suficiente; reubicar los ABC al lado del dominio (inversión -> total) sería más churn sin beneficio proporcional. - -### 3.4 Propiedad del logger unificada (Problema 3) - -`Task` gana la responsabilidad de su propio teardown: - -```python -# task.py -def close_handlers(self) -> None: - """Cierra y desadjunta los handlers del logger de la tarea.""" - for handler in list(self.logger.handlers): - handler.close() - self.logger.removeHandler(handler) -``` - -`Process.close_loggers` y `Process.remove_task` dejan de duplicar el bucle y -llaman `task.close_handlers()`. El que *crea* los handlers es el que los -*destruye*. - ---- - -## 4. Decisiones de diseño - -| # | Decisión | Razón | -|---|---|---| -| A | Extraer tipos de valor (`TaskStatus`, `TaskResult`, `TaskDependency`, `ErrorData`) a módulos leaf sin imports de comms | Rompe el ciclo en la raíz; elimina el workaround de strings; cero cambio de API | -| B | Una sola implementación de transporte por medio (`_SMTPTransport`, `_WebhookTransport`) usada por streaming y one-shot | Elimina la duplicación de SMTP entre handler y reporte | -| C | Mantener `NotificationChannel` y `ReportChannel` como **dos ABCs separados** | Son modelos de entrega genuinamente distintos (streaming vs one-shot); forzar uno solo mezclaría conceptos | -| D | Mantener la doble capacidad de `EmailChannel`/`WebhookChannel` | Conveniencia de "un objeto, dos roles". Deja de ser smell una vez que render/transport son capas compartidas. `_FileChannel` (solo `NotificationChannel`) demuestra que los ABCs no se imponen a la fuerza | -| E | `Task.close_handlers()` | Unifica la propiedad del ciclo de vida del logger; elimina la duplicación Process/Task | -| F | Paquete `comms/` plano + shim en `processes.html_logging` | Agrupa por rol sin romper rutas públicas ni sobre-anidar | - ---- - -## 5. Plan de implementación por fases - -Cada fase deja `pytest` / `mypy` / `ruff` verdes y es aprobable por separado. - -### Fase 0 — Romper el ciclo y la duplicación Process/Task (pequeña, sin cambio de API) -- Extraer `TaskStatus`, `TaskResult`, `TaskDependency` a `task_types.py`. -- Extraer `ErrorData` a `error_data.py`; dejar `_ErrorContextFormatter` donde está - (luego se moverá a `comms/`). -- Borrar `_STATUS_SUCCESS` / `_STATUS_ERRORED`; los renderers importan `TaskStatus`. -- Añadir `Task.close_handlers()`; `Process.close_loggers` y `remove_task` lo usan. -- **Impacto:** nulo en API pública. Es la fase que ataca directamente los puntos - que el usuario señaló (workaround + duplicación Process/Task). - -### Fase 1 — Unificar transporte -- Introducir `_SMTPTransport.send(subject, html_body)` con la conexión + MIME + - auth + `sendmail` + `quit` en un solo lugar. -- `_HTMLEmailHandler.emit` y `send_report_email` delegan en él. (El handler puede - simplificarse a un `logging.Handler` plano, ya que hoy sobreescribe casi todo - `SMTPHandler`.) -- Cerrar la simetría del webhook alrededor de `_WebhookTransport` (formaliza el - ya existente `_post_json`). -- **Impacto:** nulo en API pública; elimina la duplicación del Problema 2. - -### Fase 2 — Introducir el paquete `comms/` -- Mover los módulos de comunicación a `comms/` según §3.2. -- `comms/__init__.py` re-exporta los símbolos públicos; `processes/__init__.py` - pasa a importar desde `comms`. -- Dejar shim en `processes/html_logging.py`. -- **Impacto:** movimientos de archivo + actualización de imports. API pública y - rutas documentadas intactas (verificado por los tests existentes + invariante 3). - -### Fase 3 — (Opcional) Formalizar la capa de render -- Separar limpiamente render de transport donde aún convivan en un mismo archivo, - si la Fase 1/2 no lo dejó ya nítido. - ---- - -## 6. Extensibilidad — añadir un canal nuevo - -Con la estructura objetivo, agregar p. ej. un canal de Slack: - -1. ¿POSTea JSON? Reutiliza `_WebhookTransport` tal cual. -2. Render: añade un formateador/render en `comms/_webhook_render.py` (o uno nuevo) - si el payload difiere. -3. `class SlackChannel(NotificationChannel, ReportChannel)` en `comms/channels.py`, - cableando render + transport. Implementa `build_handler` y/o `send_report` - según los roles que soporte. -4. Exportar en `comms/__init__.py` y `processes/__init__.py`. - -No se toca el dominio (`task.py`, `process.py`, `execution_report.py`): solo se -añade un adaptador. Esa es la escalabilidad que se busca. - ---- - -## 7. Resumen - -| Problema | Solución | Fase | -|---|---|---| -| Workaround de import circular | Tipos de valor en leaves; renderers importan `TaskStatus` | 0 | -| Duplicación Process/Task | `Task.close_handlers()` | 0 | -| Transporte duplicado (email/report) | `_SMTPTransport` / `_WebhookTransport` compartidos | 1 | -| Namespace plano con roles mezclados | Paquete `comms/` + shims | 2 | -| Doble contrato de canal | Se conserva (2 ABCs) — deja de ser smell al compartir capas | 1–2 | - -Recomendación: aprobar e implementar **Fase 0 y 1** primero (correctitud y -deduplicación, riesgo mínimo, sin cambio de API). La **Fase 2** (paquete) es la -más "churny" y conviene revisarla como PR aparte. - ---- - -## 8. Estado de implementación (as-built) - -Fases 0, 1 y 2 implementadas y verdes (134 tests, mypy, ruff; wheel verificado -con `comms/themes/` incluido). Desviaciones menores respecto al plan, todas -deliberadas: - -- **Configs no se fusionaron** en un único `config.py`: se mantienen - `comms/email_config.py` y `comms/webhook_config.py` (menos churn, sin - beneficio real en fusionar). -- **Nombres de módulo de delivery**: `comms/_email.py` y `comms/_webhook.py` - (en vez de `_smtp.py`/`_webhook.py`); contienen render + transporte + - handler de su medio. `_logfile_formatting.py` → `comms/_logfile.py`. -- **Sin shims** en rutas viejas: los tests acoplados a rutas internas se - repuntaron a la nueva ubicación (o al import público cuando el símbolo es - público). `html_logging.py` ya no existía, así que no hubo shim que mantener. -- **Fase 3 (separar render/transport en archivos distintos): no realizada.** - Las clases de transporte (`_SMTPTransport`, `_WebhookTransport`) ya quedan - nítidamente delimitadas dentro de sus archivos; separarlas más sería - fragmentación sin beneficio. Queda como refinamiento opcional futuro. diff --git a/docs/examples/advanced.md b/docs/examples/advanced.md index e0b3e4d..60e99ca 100644 --- a/docs/examples/advanced.md +++ b/docs/examples/advanced.md @@ -191,15 +191,15 @@ dependencies=[ ## 📧 Email Notifications -If a task fails it can notify via email: +After a run, the report can notify via email. For each failure it includes: - The name of the failing task - The python function being executed with its args and kwargs - The traceback of the error - The tasks that could not be executed in the process due to this failure. -To set this up, pass an `EmailChannel` to the Task constructor via `channels`: +To set this up, pass an `EmailChannel` to `report.notify(...)`: ```python -from processes import SMTPConfig, HTMLEmailStyle, EmailChannel, Task +from processes import SMTPConfig, HTMLEmailStyle, EmailChannel, Process smtp = SMTPConfig( mailhost=('smtp_server', 587), @@ -211,12 +211,13 @@ smtp = SMTPConfig( # Optional: customise the HTML presentation (all fields have defaults) style = HTMLEmailStyle( - style='modern', # classic | modern | compact palette='neutral', # neutral | catppuccin | neobones | slate language='en', # en | es | pt | fr | de | it ) -t = Task("task_name", func_to_run, "logfile", channels=[EmailChannel(smtp, style)]) +with Process(tasks) as process: + report = process.run() + report.notify(EmailChannel(smtp, style), only_errors=True) ``` ## ⏱️ Retries & Timeouts @@ -253,5 +254,6 @@ t_fetch = Task( If every attempt fails, the task is marked failed with the **last** exception raised — `retries` only controls how many times `func` is -retried, not whether the failure is eventually reported. Combine with -an `EmailChannel` to be paged only once all attempts are exhausted. \ No newline at end of file +retried, not whether the failure is eventually reported. Notify the +report with an `EmailChannel` afterwards to be paged only once all +attempts are exhausted. \ No newline at end of file diff --git a/docs/index.md b/docs/index.md index f20caa5..8613594 100644 --- a/docs/index.md +++ b/docs/index.md @@ -73,7 +73,7 @@ from datetime import date from processes import Process, Task, TaskDependency, SMTPConfig, HTMLEmailStyle, EmailChannel -# 1. Setup Email Alerts (Optional) +# 1. Setup the report email (Optional) smtp_config = SMTPConfig( mailhost=('smtp_server', 587), fromaddr='sender@example.com', @@ -82,7 +82,6 @@ smtp_config = SMTPConfig( secure=(), # () = STARTTLS; omit for no encryption ) email_style = HTMLEmailStyle( - style='modern', # classic | modern | compact palette='neutral', # neutral | catppuccin | neobones | slate language='en', # en | es | pt | fr | de | it ) @@ -100,7 +99,7 @@ def sum_data_from_csv_and_x(x, a=1, b=2): # 3. Create the Task Graph (order is irrelevant, that is handled by Process) tasks = [ Task("t-1", get_previous_working_day, "etl.log"), - Task("intependent", indep_task, "indep.log", channels=[EmailChannel(smtp_config, email_style)]), # This task will send email on failure + Task("intependent", indep_task, "indep.log"), Task("sum_csv", search_and_sum_csv, "etl.log", dependencies= [ TaskDependency("t-1", @@ -117,9 +116,10 @@ tasks = [ ) ] -# 4. Run the Process +# 4. Run the Process and notify the report (only the errored tasks) with Process(tasks) as process: # Context Manager ensures correct disposal of loggers - process_result = process.run() # To enable parallelization use .run(parallel=True) + report = process.run() # To enable parallelization use .run(parallel=True) + report.notify(EmailChannel(smtp_config, email_style), only_errors=True) ``` @@ -127,26 +127,26 @@ with Process(tasks) as process: # Context Manager ensures correct disposal of lo ## 📧 Customizing the HTML email -When a task with an `EmailChannel` raises, the alert is a **styled HTML -email** built from a bundled template. The body includes the exception, -the traceback (with the user-frame highlighted), the task context, the -list of downstream tasks that were skipped because of the failure, and -the local variables at the failing frame (see [Traced Variables](#traced-variables) below). +When you deliver a finished report with `report.notify(EmailChannel(...))`, the +report is a **styled HTML email** built from a bundled template. It lists every +task with its status, and for each failure includes the exception, the traceback, +the downstream tasks that were skipped, and the local variables captured at the +failing frame (see [Traced Variables](#traced-variables) below). How much per-task +detail is included is controlled by `ReportContent(show_traceback=…, +show_traced_vars=…)`. Email delivery and presentation are configured with two separate dataclasses: - **`SMTPConfig`** — transport settings (host, credentials, sender/recipients, TLS). -- **`HTMLEmailStyle`** — presentation settings, all optional: +- **`HTMLEmailStyle`** — presentation settings, both optional: | Field | Values | Default | |---|---|---| -| `style` | `classic`, `modern`, `compact` | `modern` | | `palette` | `neutral`, `catppuccin`, `neobones`, `slate` | `neutral` | | `language` | `en`, `es`, `pt`, `fr`, `de`, `it` | `en` | -| `traced_vars_frame_filter` | any path substring, or `None` | `None` (outermost user frame) | ```python -from processes import SMTPConfig, HTMLEmailStyle, EmailChannel, Task +from processes import SMTPConfig, HTMLEmailStyle, EmailChannel, ReportContent smtp = SMTPConfig( mailhost=("smtp.example.com", 587), @@ -157,34 +157,33 @@ smtp = SMTPConfig( ) style = HTMLEmailStyle( - style="compact", # classic | modern | compact palette="catppuccin", # neutral | catppuccin | neobones | slate language="es", # en | es | pt | fr | de | it ) -t = Task("task_name", func_to_run, "logfile", channels=[EmailChannel(smtp, style)]) +channel = EmailChannel(smtp, style, content=ReportContent(show_traced_vars=False)) +report.notify(channel) # or notify(channel, only_errors=True) ``` If `style` is omitted, `EmailChannel` defaults to `HTMLEmailStyle()` -(modern, neutral, English). If no `EmailChannel` is included in `channels`, -no email handler is attached. +(neutral, English). -All assets ship inside the wheel — the styles are Jinja-style HTML -templates at `src/processes/themes/styles/` and the palettes are CSS -fragments at `src/processes/themes/palettes/`. No template engine or -extra install is required; the formatter composes them at send time. +All assets ship inside the wheel — the report layout is an HTML template at +`src/processes/comms/themes/styles/report.html` and the palettes are CSS +fragments at `src/processes/comms/themes/palettes/`. No template engine or +extra install is required; the renderer composes them at send time. ### Traced Variables -On failure, the email body includes the local variables of the -**outermost user frame in the traceback** — i.e. the last frame in -the chain that is not inside `site-packages` or your virtualenv. -A `file:line` reference next to the section tells you exactly where -in the source the listed values were captured, which is usually the -fastest way to figure out *why* a complex task broke deep inside a -wrapper. +On failure, each task captures the local variables of the **outermost user +frame in the traceback** — i.e. the last frame in the chain that is not inside +`site-packages` or your virtualenv. A `file:line` reference next to the section +tells you exactly where in the source the listed values were captured, which is +usually the fastest way to figure out *why* a complex task broke deep inside a +wrapper. These captured variables flow into both the task's logfile and the +report email. -This default can be overridden with `HTMLEmailStyle.traced_vars_frame_filter`. +This default can be overridden per task with `Task(traced_vars_frame_filter=…)`. Set it to a path substring (e.g. the name of one of your own packages or modules) to capture locals from the outermost frame whose filename contains that substring instead — useful for deep-debugging code that runs through diff --git a/report-notifications-design.md b/report-notifications-design.md deleted file mode 100644 index e62c1b2..0000000 --- a/report-notifications-design.md +++ /dev/null @@ -1,184 +0,0 @@ -# Diseño: notificación de `ProcessExecutionReport` (email + webhook) - -> Branch: `feature/report-notifications` -> Estado: stubs `notify` / `notify_errors` ya creados (lanzan `NotImplementedError`). - -## 1. Análisis de situación - -El proyecto ya tiene infraestructura de notificación, pero **orientada a `Task`** y -**dirigida por logging**: - -- `NotificationChannel` (ABC) expone un único método: `build_handler(task_name) -> logging.Handler`. -- `_FileChannel`, `EmailChannel`, `WebhookChannel` producen handlers que un `Task` - adjunta a su logger. Emiten **por cada error**, mientras la tarea corre. -- El render de email (`_HTMLEmailFormatter` + templates en `themes/styles`, - `themes/palettes`, `themes/languages`) está hecho para **un solo fallo**: - substituciones `exception`, `traceback_highlight`, `traced_vars`, - `downstream_items`, etc. -- Configs de transporte/presentación: `SMTPConfig`, `HTMLEmailStyle`, - `WebhookConfig`. - -`ProcessExecutionReport` ya sabe serializarse (`to_json()`, lossless), y se acaban -de agregar los stubs `notify()` / `notify_errors()`. - -**Tensión central:** los dos consumidores tienen modos de entrega distintos. - -| | `Task` | `ProcessExecutionReport` | -|---|---|---| -| Modo | streaming / push | one-shot / pull | -| Dirigido por | `logging.LogRecord` | objeto completo en memoria | -| Cuándo | cada error, durante el run | una vez, al terminar | -| Contenido | un fallo | tabla de N tasks (success/errored/skipped, tiempos) | - -Forzar el reporte por el mismo `build_handler`/`SMTPHandler` obligaría a -**fabricar `LogRecord` falsos** → impedance mismatch. Hay que evitarlo. - -## 2. Objetivo - -Permitir que un `ProcessExecutionReport` se notifique vía: - -- **Email**, configurable en **estilo** (layout/paleta/idioma, reusando el sistema - de themes) y en **información** enviada. -- **Webhook**, como **JSON** (reusando `to_json()`). - -Con dos entradas: `notify()` (reporte completo) y `notify_errors()` (solo entries -`ERRORED`). Idealmente, que un mismo "canal" se pueda anotar tanto a una `Task` -como a un `ProcessExecutionReport`, configurando el destino una sola vez, y -extensible a nuevos destinos (Slack/Teams/etc.). - -## 3. Solución propuesta - -### 3.1 Qué se reutiliza y qué no - -| Pieza | Reutilizar | Nota | -|---|---|---| -| `SMTPConfig` | ✅ tal cual | Transporte puro, desacoplado de Task/logging. | -| `WebhookConfig` | ✅ tal cual | Transporte puro (URL/JSON). | -| Sistema de themes (`themes/`) | ✅ | Assets de estilo/paleta/idioma. | -| `HTMLEmailStyle` | 🟡 como selector de estilo | `traced_vars_frame_filter` es específico de fallo; evaluar un `ReportEmailStyle` o ignorar ese campo. | -| `_HTMLEmailFormatter` + templates de error + handlers | ❌ | Renderizan un solo fallo y son `LogRecord`-driven. El reporte necesita template y formatter propios, y envío one-shot. | - -### 3.2 Abstracción de canales: dos interfaces angostas (no un ABC gordo) - -```text -TaskChannel -> build_handler(task_name) -> logging.Handler (= NotificationChannel actual) -ReportChannel -> send_report(report, *, errors_only) -> None -``` - -- `EmailChannel` / `WebhookChannel` implementan **ambas** (mismo destino, dos verbos), - reusando internamente `SMTPConfig` / `WebhookConfig`. -- `_FileChannel` implementa **solo** `TaskChannel`. -- Composición sobre un ABC único: así ningún canal se ve forzado a un método que no - tiene sentido (evita reintroducir `NotImplementedError`). - -### 3.3 Render del reporte (email) - -- Nuevo template "reporte" (tabla de tasks: nombre, estado, intentos, duración, - y sección de errores con su contexto) reusando los directorios de - styles/palettes/languages. -- Nuevo formatter que consume un `ProcessExecutionReport` (no un `LogRecord`). -- Envío SMTP one-shot (no vía `logging.handlers.SMTPHandler`). - -### 3.4 Render del reporte (webhook) - -- POST del payload `to_json()` (ya lossless). `errors_only` filtra a entries `ERRORED`. - -## 4. Razón - -- **Separar transporte de presentación de entrega.** El transporte - (`SMTPConfig`/`WebhookConfig`) es lo genuinamente desacoplado y reutilizable; el - render por-fallo no lo es. Reutilizar el transporte evita duplicar auth/URL/TLS. -- **Respetar streaming vs one-shot.** Son modos de entrega distintos; un único - `build_handler` no modela ambos sin hacks (LogRecords falsos). -- **Interfaces angostas > ABC gordo.** Permite "anotar el mismo canal a Task o a - Report" sin métodos irrelevantes, y deja extensibilidad limpia (un destino nuevo - implementa lo que aplique). -- **No abstraer con un solo caso.** Empezar con configs directas y extraer canales - cuando aparezca el 2º/3er destino reduce el riesgo de sobre-diseño. - -## 5. Plan de implementación - -**Fase 0 — hecho.** Stubs `notify` / `notify_errors` con firmas que referencian -`SMTPConfig`/`HTMLEmailStyle`/`WebhookConfig` (sin churn futuro). - -**Fase 1 — Webhook (lo más simple).** -- Implementar `notify`/`notify_errors` para el caso webhook usando `to_json()` + - `WebhookConfig`. `errors_only` filtra entries. -- Tests: payload correcto, filtrado de errores, no-op si no hay webhook. - -**Fase 2 — Email (reporte completo).** -- Nuevo template de reporte + nuevo formatter (tabla de tasks + detalle de errores), - reusando themes y `HTMLEmailStyle` como selector. -- Envío SMTP one-shot con `SMTPConfig`. -- Decidir: ¿qué es "información configurable"? (columnas/secciones, incluir - tracebacks sí/no, etc.). ¿`HTMLEmailStyle` o nuevo `ReportEmailStyle`? -- Tests: render por estilo/paleta/idioma; modo errors-only. - -**Fase 3 — Abstracción de canales.** -- Introducir `ReportChannel`; renombrar/alias `NotificationChannel` → `TaskChannel` - (manteniendo compat). -- `EmailChannel`/`WebhookChannel` implementan ambas interfaces. -- `notify`/`notify_errors` aceptan objetos canal además de configs. - -**Fase 4 — Docs + ejemplos.** -- Documentar en `docs/` y agregar ejemplo en `examples/`. - -## 6. Decisiones abiertas - -1. ¿`HTMLEmailStyle` reutilizado o `ReportEmailStyle` nuevo (por `traced_vars_frame_filter`)? -2. ¿Qué exactamente es "información configurable" del email del reporte? -3. ¿`send_report` síncrono y que propague errores de envío, o best-effort silencioso - como hacen los handlers de logging hoy? -4. ¿`notify` recibe configs (simple) o canales (unificado) en la primera versión? - -## 7. Alternativa simplificada (sin abstracción de canales) - -En lugar de introducir `TaskChannel` / `ReportChannel`, `notify` y `notify_errors` -reciben **directamente** los objetos de transporte que ya existen — exactamente las -firmas de los stubs actuales: - -```python -report.notify( - email=SMTPConfig(...), - email_style=HTMLEmailStyle(...), # o ReportEmailStyle, ver 7.1 - webhook=WebhookConfig(...), -) -report.notify_errors(webhook=WebhookConfig(...)) -``` - -El reporte arma internamente su payload (HTML para email, `to_json()` para webhook) -y lo envía. **No** hay interfaces de canal, **no** hay objeto que se anote a la vez a -`Task` y a `Report`. - -**Pros** -- Mínimo: es literalmente implementar los stubs tal cual; cero abstracción nueva. -- Aprovecha lo genuinamente reutilizable (`SMTPConfig`/`WebhookConfig` son transporte - puro), que es donde está casi todo el valor de reuso. -- En espíritu con una lib "lightweight, zero-dep". - -**Contras** -- No unifica "configurar un destino una vez y usarlo en Task y en Report". -- Cada destino nuevo (Slack/Teams/SNS) es **otro parámetro** de `notify`, no una clase - polimórfica. Escala mal si se esperan muchos destinos. -- Algo de duplicación entre el envío del lado-Task (handlers) y el lado-Report. - -**Cuándo preferirla:** si la notificación del reporte es la única necesidad nueva y no -se prevén muchos destinos. **Recomendación:** empezar por aquí; extraer canales -(sección 3.2) recién cuando aparezca el 2º/3er destino o se quiera la ergonomía de -"un canal para ambos". Es reversible: las firmas con configs pueden coexistir o migrar -a aceptar canales después. - -### 7.1 `ReportEmailStyle`: ¿subclase o hermanas? - -`HTMLEmailStyle` tiene `style` / `palette` / `language` (comunes a cualquier email) y -`traced_vars_frame_filter` (**específico de un fallo**, sin sentido en un reporte). - -| Opción | Veredicto | Razón | -|---|---|---| -| **A. `ReportEmailStyle(HTMLEmailStyle)`** (subclase) | ❌ Evitar | Heredaría `traced_vars_frame_filter`, que no aplica. Herencia debe **añadir**, no heredar-e-ignorar. Un reporte "no es un tipo de" estilo-de-error (LSP smell). | -| **B. Hermanas bajo base común `EmailStyle`** | ✅ Si aporta campos propios | Base `EmailStyle` = `style`/`palette`/`language`. `HTMLEmailStyle(EmailStyle)` añade `traced_vars_frame_filter`; `ReportEmailStyle(EmailStyle)` añade lo específico del reporte (qué columnas/secciones, incluir tracebacks, etc.). Sin campos muertos. | -| **C. Reusar `HTMLEmailStyle` tal cual** (ignorar el campo) | ✅ Si no aporta nada aún | Lo más simple. Trade-off: arrastra un campo semánticamente muerto en el contexto de reporte. | - -**Recomendación:** empezar con **C** (reusar `HTMLEmailStyle`) en la versión simplificada; -migrar a **B** (extraer base `EmailStyle` + hermanas) en cuanto el email del reporte -necesite sus propios knobs. **Nunca A.** diff --git a/report-notifications-implementation.md b/report-notifications-implementation.md deleted file mode 100644 index 6947c31..0000000 --- a/report-notifications-implementation.md +++ /dev/null @@ -1,88 +0,0 @@ -# Report Notifications — Implementation Decisions & Summary - -## Decisiones tomadas - -### 1. Template único `report.html` (no 3 variantes por estilo) - -**Decisión:** Se creó un único `themes/styles/report.html` en lugar de 3 variantes (`report_classic.html`, `report_modern.html`, `report_compact.html`). - -**Razón:** El layout classic/modern/compact diferencia la presentación de un *único* error por tarea. Para un reporte multi-tarea el layout es inherentemente diferente (grilla de resumen + lista colapsable de tareas), y las tres variantes habrían producido diferencias cosméticas mínimas sin valor real. La palette y el language sí se respetan. - -**Implicación:** `HTMLEmailStyle.style` ("classic", "modern", "compact") no tiene efecto en los emails de reporte. Está documentado en el docstring de `_build_report_html`. - ---- - -### 2. `
`/`` para secciones colapsables - -**Decisión:** Cada tarea se envuelve en `
`. Las tareas con error usan `
` (abiertas por defecto); success/skipped se generan cerradas. - -**Razón:** Es la única opción viable con CSS puro en email. El "checkbox hack" es eliminado por la mayoría de los clientes. `
` es soportado en Apple Mail, Thunderbird, Gmail web. En Outlook el contenido se muestra expandido (degradación aceptable: siempre visible, nunca oculto). - ---- - -### 3. Sin `process_name` en el reporte - -**Decisión:** El asunto y el encabezado del email no incluyen el nombre del proceso. - -**Razón:** `ProcessExecutionReport` es un dataclass frozen y su contrato está estabilizado. Añadir `process_name` requeriría modificar `from_results` y el constructor, lo cual está fuera del scope de esta branch. Se usa un título genérico localizado. - ---- - -### 4. Circular import resuelto con comparación por `.value` - -**Decisión:** `_email_internals.py` y `_webhook_internals.py` no importan `TaskStatus`. En su lugar comparan `entry.status.value == "errored"` (string). - -**Razón:** La cadena de importación era: `task.py` → `notification_channels.py` → `_email_internals.py` → `task.py`. Usar el string del enum value corta el ciclo sin necesidad de lazy imports ni reorganización del módulo. - ---- - -### 5. `_post_json()` extraído como helper compartido - -**Decisión:** El signing HMAC + POST urllib fue extraído de `_WebhookHandler.emit` a `_post_json(config, payload_str)`. - -**Razón:** Antes la lógica estaba duplicada en `emit` (por tarea) y tendría que haberse duplicado de nuevo en `send_report_webhook`. Ahora ambos llaman a `_post_json`. - ---- - -### 6. `show_warnings: bool = True` en `notify`/`notify_errors` - -**Decisión:** Se añadió `show_warnings: bool = True` como kwarg a ambos métodos. Los errores se capturan con `try/except` dentro del loop y emiten `warnings.warn(...)`. - -**Razón:** Un canal que falla no debe abortar los demás. `warnings.warn` es el idioma Python correcto para alertas no-fatales; por defecto visible, silenciable con `show_warnings=False`. - ---- - -### 7. Renderers puros + transporte separado - -**Decisión:** Se crearon funciones puras: -- `_build_report_html(report, style, content, *, errors_only) -> str` -- `_build_report_webhook_payload(entries, content, config) -> dict` -- `send_report_email(...)` y `send_report_webhook(...)` como thin wrappers de transporte - -**Razón:** Sigue el patrón existente (`_HTMLEmailFormatter` / `_HTMLEmailHandler`, `_WebhookFormatter` / `_WebhookHandler`). Los renderers son testables sin I/O (los tests de HTML los invocan directamente). - ---- - -## Resumen de implementación - -### Archivos nuevos -| Archivo | Descripción | -|---|---| -| `src/processes/themes/styles/report.html` | Template HTML del reporte (palette-aware, multi-tarea, `
` colapsables) | -| `tests/test_report_send.py` | 21 tests: webhook payload, flags de contenido, signing HMAC, renderer HTML, transporte SMTP | - -### Archivos modificados -| Archivo | Cambios | -|---|---| -| `_webhook_internals.py` | Extraído `_post_json()`; añadidos `_build_report_webhook_payload()` y `send_report_webhook()` | -| `_email_internals.py` | Añadidos `_build_task_section_html()`, `_build_report_html()`, `send_report_email()` | -| `notification_channels.py` | `EmailChannel.send_report` y `WebhookChannel.send_report` implementados (dejaron de lanzar `NotImplementedError`) | -| `execution_report.py` | `notify`/`notify_errors` con `show_warnings: bool = True` y try/except por canal | -| `themes/languages/*.json` | 12 keys nuevas de reporte en los 6 idiomas (en, es, pt, fr, de, it) | -| `tests/test_report_notify_dispatch.py` | Removido test de `NotImplementedError`; añadidos 4 tests de `show_warnings` | - -### Estado final -- **134 tests passed** (0 failed) -- **mypy**: no issues -- **ruff**: no issues -- Branch: `feature/report-notifications` @ `5692a4d` diff --git a/src/processes/__init__.py b/src/processes/__init__.py index 4127afb..67628be 100644 --- a/src/processes/__init__.py +++ b/src/processes/__init__.py @@ -3,7 +3,6 @@ from .comms import EmailChannel as EmailChannel from .comms import HTMLEmailStyle as HTMLEmailStyle -from .comms import NotificationChannel as NotificationChannel from .comms import ReportChannel as ReportChannel from .comms import ReportContent as ReportContent from .comms import SMTPConfig as SMTPConfig diff --git a/src/processes/_logfile.py b/src/processes/_logfile.py new file mode 100644 index 0000000..53e9192 --- /dev/null +++ b/src/processes/_logfile.py @@ -0,0 +1,100 @@ +"""Task logfile formatting — a domain concern, not a communication channel. + +A ``Task`` writes its own diagnostic logfile; on failure it appends the same +structured context (``ErrorData``) the report carries. This lives in the domain +(not in ``comms``) because it is about a task recording what happened, not about +delivering a report to an external destination. +""" + +from __future__ import annotations + +import logging + +from .error_data import ErrorData + +_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + + +class _ErrorContextFormatter(logging.Formatter): + """Base formatter providing typed access to a record's failure context.""" + + def _error_data(self, record: logging.LogRecord) -> ErrorData: + """Extract the failure context from a log record. + + Parameters + ---------- + record : logging.LogRecord + The record being formatted. May or may not carry a + ``task_context`` attribute. + + Returns + ------- + ErrorData + Typed view of ``record.task_context``, with defaults filled in + for any missing fields. + """ + ctx = getattr(record, "task_context", None) or {} + return ErrorData( + task_name=str(ctx.get("task_name", "?")), + function=str(ctx.get("function", "?")), + args=ctx.get("args", ()), + kwargs=ctx.get("kwargs", {}), + 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", {}) or {}, + traced_vars_location=ctx.get("traced_vars_location", ""), + ) + + +class _TaskLogfileFormatter(_ErrorContextFormatter): + """Plain-text formatter for task logfiles. + + On failure, appends the structured failure context (function, args, kwargs, + downstream impact, traced variables and their location, traceback) as + readable text. + """ + + def __init__(self) -> None: + super().__init__(_LOG_FORMAT) + + def format(self, record: logging.LogRecord) -> str: + """Render a log record as plain text, appending failure context if present. + + Parameters + ---------- + record : logging.LogRecord + The record being formatted. + + Returns + ------- + str + The formatted log line, with the failure context appended if + ``record.task_context`` is set. + """ + if not getattr(record, "task_context", None): + return super().format(record) + + exc_info, record.exc_info = record.exc_info, None + try: + base = super().format(record) + finally: + record.exc_info = exc_info + + error = self._error_data(record) + lines = [ + base, + "", + f"Function: {error.function}", + f"Args: {error.args!r}", + f"Kwargs: {error.kwargs!r}", + 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")) + return "\n".join(lines) diff --git a/src/processes/comms/__init__.py b/src/processes/comms/__init__.py index 96d96d6..0147e55 100644 --- a/src/processes/comms/__init__.py +++ b/src/processes/comms/__init__.py @@ -1,12 +1,11 @@ -"""Communication layer: channels, transports and renderers. +"""Communication layer: report channels, transports and renderers. -Public surface re-exported here (and, in turn, from ``processes``): -the channel ports (``NotificationChannel``, ``ReportChannel``), the shared -``ReportContent`` selector, the concrete ``EmailChannel`` / ``WebhookChannel``, -and their transport configs. +Pure report-delivery package — it never touches task execution or logfiles. +Public surface re-exported here (and, in turn, from ``processes``): the +``ReportChannel`` port, the shared ``ReportContent`` selector, the concrete +``EmailChannel`` / ``WebhookChannel``, and their transport configs. """ -from .base import NotificationChannel as NotificationChannel from .base import ReportChannel as ReportChannel from .base import ReportContent as ReportContent from .channels import EmailChannel as EmailChannel diff --git a/src/processes/comms/_email.py b/src/processes/comms/_email.py index 6b06c3e..9c50057 100644 --- a/src/processes/comms/_email.py +++ b/src/processes/comms/_email.py @@ -2,16 +2,15 @@ import html import json -import logging -import logging.handlers import os import smtplib +from datetime import date from email.mime.text import MIMEText from email.utils import formatdate +from functools import cache from typing import TYPE_CHECKING, cast from ..task_types import TaskStatus -from ._error_context import _ErrorContextFormatter from .email_config import HTMLEmailStyle, SMTPConfig if TYPE_CHECKING: @@ -26,9 +25,15 @@ _PALETTE_MARKER = "{{__palette_css__}}" +@cache def _load_language_strings(language: str) -> dict[str, str]: """Load translatable strings for the given ISO 639-1 language code. + Cached per language: the bundled JSON files never change at runtime, so the + file is read from disk at most once per language for the life of the + process. Callers only read the returned mapping (never mutate it), so the + shared cached dict is safe. + Parameters ---------- language : str @@ -44,127 +49,39 @@ def _load_language_strings(language: str) -> dict[str, str]: return cast(dict[str, str], json.load(fh)) -class _HTMLEmailFormatter(_ErrorContextFormatter): - """Pure renderer: reads all error context from ``record.task_context`` and - fills the HTML template. No exception-info parsing or frame walking here. - """ - - def __init__(self, style: HTMLEmailStyle) -> None: - super().__init__() - self._email_style = style.style - self._color_palette = style.palette - self._email_language = style.language - self._cached_template: str | None = None - - def _resolve_template(self) -> str: - style_path = os.path.join(_STYLES_DIR, f"{self._email_style}.html") - palette_path = os.path.join(_PALETTES_DIR, f"{self._color_palette}.css") - with open(style_path, encoding="utf-8") as fh: - style = fh.read() - with open(palette_path, encoding="utf-8") as fh: - palette = fh.read() - return style.replace(_PALETTE_MARKER, palette) - - def _get_template(self) -> str: - if self._cached_template is None: - self._cached_template = self._resolve_template() - return self._cached_template - - def _split_traceback_at_target(self, tb_str: str, location: str) -> tuple[str, str, str]: - """Split *tb_str* around the frame line matching *location*. - - Parameters - ---------- - tb_str : str - Full formatted traceback to split. - location : str - ``"filename:lineno"`` of the frame to highlight, as produced by - ``_build_traced_vars_location``. - - Returns - ------- - tuple[str, str, str] - ``(before, highlight, after)`` where ``highlight`` is the - ``File "", line , in `` line for that - frame. Returns ``("", "", tb_str)`` if no match is found. - """ - if not tb_str or not location: - return ("", "", tb_str) - try: - filename, lineno_str = location.rsplit(":", 1) - lineno = int(lineno_str) - except ValueError: - return ("", "", tb_str) - - needle = f' File "{filename}", line {lineno}, in' - lines = tb_str.splitlines(keepends=True) - for i, line in enumerate(lines): - if needle in line: - return ("".join(lines[:i]), line, "".join(lines[i + 1 :])) - return ("", "", tb_str) - - def _render(self, template: str, substitutions: dict[str, str]) -> str: - rendered = template - for key, value in substitutions.items(): - rendered = rendered.replace("{{" + key + "}}", value) - return rendered - - def format(self, record: logging.LogRecord) -> str: - """Render a log record as a complete HTML email body. - - Parameters - ---------- - record : logging.LogRecord - The record being formatted. - - Returns - ------- - str - The fully rendered HTML email body. - """ - error = self._error_data(record) +@cache +def _load_palette_css(palette: str) -> str: + """Load the CSS fragment for the given palette name. - tb_before, tb_highlight, tb_after = self._split_traceback_at_target( - error.traceback_str, error.traced_vars_location - ) + Cached per palette: the bundled CSS files never change at runtime, so each + palette is read from disk at most once per process. The returned string is + immutable, so sharing it across renders is safe. + """ + path = os.path.join(_PALETTES_DIR, f"{palette}.css") + with open(path, encoding="utf-8") as fh: + return fh.read() - downstream_items = "".join( - 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() - ) +@cache +def _load_report_template() -> str: + """Load the ``report.html`` body template. - substitutions = dict(_load_language_strings(self._email_language)) - substitutions["lang_traced_vars_blurb"] = substitutions.get( - "lang_traced_vars_blurb", "" - ).replace("{location}", error.traced_vars_location) - substitutions.update( - { - "task_name": html.escape(error.task_name, quote=True), - "function": html.escape(error.function, quote=True), - "args": html.escape(repr(error.args), quote=True), - "kwargs": html.escape(repr(error.kwargs), quote=True), - "exception": html.escape(error.exception, quote=True), - "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": traced_vars_html, - "downstream_items": downstream_items, - } - ) - return self._render(self._get_template(), substitutions) + Cached: it is a single invariant bundled file, read from disk at most once + per process. The returned string is immutable; callers compose it with + ``str.replace`` (which yields a new string), so the cached value is never + mutated. + """ + path = os.path.join(_STYLES_DIR, "report.html") + with open(path, encoding="utf-8") as fh: + return fh.read() class _SMTPTransport: """Sends one HTML email per call over a fresh SMTP connection. The single place that owns the SMTP conversation (connect, optional - STARTTLS + login, ``sendmail``, ``quit``). Both the streaming task handler - (``_HTMLEmailHandler``) and the one-shot report sender (``send_report_email``) - delegate here, so the transport logic exists exactly once. + STARTTLS + login, ``sendmail``, ``quit``); ``send_report_email`` delegates + here. """ def __init__(self, config: SMTPConfig) -> None: @@ -201,58 +118,6 @@ def send(self, subject: str, html_body: str) -> None: smtp.quit() -class _HTMLEmailHandler(logging.handlers.SMTPHandler): - """Internal SMTP handler that sends log records as HTML emails.""" - - def __init__(self, config: SMTPConfig) -> None: - super().__init__( - config.mailhost, - config.fromaddr, - config.toaddrs, - "", # subject set by _build_task_email_handler - credentials=config.credentials, - secure=config.secure, # type: ignore[arg-type] - timeout=config.timeout, - ) - self._transport = _SMTPTransport(config) - - def emit(self, record: logging.LogRecord) -> None: - try: - self._transport.send(self.getSubject(record), self.format(record)) - except Exception: - self.handleError(record) - - -def _build_task_email_handler( - smtp_config: SMTPConfig, - style: HTMLEmailStyle, - task_name: str, -) -> _HTMLEmailHandler: - """Create a fully configured email handler bound to one task. - - Parameters - ---------- - smtp_config : SMTPConfig - SMTP transport configuration for the handler. - style : HTMLEmailStyle - HTML presentation settings used by the handler's formatter. - task_name : str - Name of the task the handler is bound to, used in the email subject. - - Returns - ------- - _HTMLEmailHandler - A handler at ``logging.ERROR`` level, with its formatter and - localized subject configured. - """ - handler = _HTMLEmailHandler(smtp_config) - handler.setFormatter(_HTMLEmailFormatter(style)) - handler.setLevel(logging.ERROR) - lang_strings = _load_language_strings(style.language) - handler.subject = f"{lang_strings['lang_email_subject']}{task_name}" - return handler - - def _build_task_section_html( entry: TaskReportEntry, lang: dict[str, str], @@ -349,30 +214,21 @@ def _build_report_html( Uses ``themes/styles/report.html`` + the palette chosen in ``style``. Language strings from ``style.language`` are applied throughout. - ``style.style`` (layout variant) is not used for reports — a single - multi-task layout is defined in ``report.html`` regardless of the - classic/modern/compact setting. Parameters ---------- report : ProcessExecutionReport The finished report to render. style : HTMLEmailStyle - Palette and language are respected; layout variant is not. + Palette and language for the rendered body. content : ReportContent Controls whether traceback and traced-variables sections appear. errors_only : bool When ``True`` only ERRORED entries appear in the output. """ lang = _load_language_strings(style.language) - palette_path = os.path.join(_PALETTES_DIR, f"{style.palette}.css") - with open(palette_path, encoding="utf-8") as fh: - palette_css = fh.read() - report_template_path = os.path.join(_STYLES_DIR, "report.html") - with open(report_template_path, encoding="utf-8") as fh: - template = fh.read() - - template = template.replace(_PALETTE_MARKER, palette_css) + palette_css = _load_palette_css(style.palette) + template = _load_report_template().replace(_PALETTE_MARKER, palette_css) entries = report.errored if errors_only else report.entries header = ( @@ -433,5 +289,16 @@ def send_report_email( lang = _load_language_strings(style.language) subject_key = "lang_report_email_subject_errors" if errors_only else "lang_report_email_subject" subject = lang.get(subject_key, "Process Execution Report") + subject = _decorate_subject(subject, report.process_name) html_body = _build_report_html(report, style, content, errors_only=errors_only) _SMTPTransport(smtp_config).send(subject, html_body) + + +def _decorate_subject(base: str, process_name: str) -> str: + """Append the process name (when set) and the run date to the subject. + + Produces e.g. ``"Process Execution Report nightly-etl | 20260620"``, or + ``"Process Execution Report | 20260620"`` when the process is unnamed. + """ + head = f"{base} {process_name}".rstrip() if process_name else base + return f"{head} | {date.today():%Y%m%d}" diff --git a/src/processes/comms/_error_context.py b/src/processes/comms/_error_context.py deleted file mode 100644 index 7dea78b..0000000 --- a/src/processes/comms/_error_context.py +++ /dev/null @@ -1,37 +0,0 @@ -from __future__ import annotations - -import logging - -from ..error_data import ErrorData - - -class _ErrorContextFormatter(logging.Formatter): - """Base formatter providing typed access to a record's failure context.""" - - def _error_data(self, record: logging.LogRecord) -> ErrorData: - """Extract the failure context from a log record. - - Parameters - ---------- - record : logging.LogRecord - The record being formatted. May or may not carry a - ``task_context`` attribute. - - Returns - ------- - ErrorData - Typed view of ``record.task_context``, with defaults filled in - for any missing fields. - """ - ctx = getattr(record, "task_context", None) or {} - return ErrorData( - task_name=str(ctx.get("task_name", "?")), - function=str(ctx.get("function", "?")), - args=ctx.get("args", ()), - kwargs=ctx.get("kwargs", {}), - 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", {}) or {}, - traced_vars_location=ctx.get("traced_vars_location", ""), - ) diff --git a/src/processes/comms/_logfile.py b/src/processes/comms/_logfile.py deleted file mode 100644 index 82960b4..0000000 --- a/src/processes/comms/_logfile.py +++ /dev/null @@ -1,60 +0,0 @@ -from __future__ import annotations - -import logging - -from ._error_context import _ErrorContextFormatter - -_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" - - -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 variables and their - location, traceback) as readable text. - """ - - def __init__(self) -> None: - super().__init__(_LOG_FORMAT) - - def format(self, record: logging.LogRecord) -> str: - """Render a log record as plain text, appending failure context if present. - - Parameters - ---------- - record : logging.LogRecord - The record being formatted. - - Returns - ------- - str - The formatted log line, with the failure context appended if - ``record.task_context`` is set. - """ - if not getattr(record, "task_context", None): - return super().format(record) - - exc_info, record.exc_info = record.exc_info, None - try: - base = super().format(record) - finally: - record.exc_info = exc_info - - error = self._error_data(record) - lines = [ - base, - "", - f"Function: {error.function}", - f"Args: {error.args!r}", - f"Kwargs: {error.kwargs!r}", - 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")) - return "\n".join(lines) diff --git a/src/processes/comms/_webhook.py b/src/processes/comms/_webhook.py index 2fdfabe..85329f9 100644 --- a/src/processes/comms/_webhook.py +++ b/src/processes/comms/_webhook.py @@ -3,13 +3,10 @@ import hashlib import hmac import json -import logging import urllib.request from typing import TYPE_CHECKING, Any -from ..error_data import ErrorData from ..task_types import TaskStatus -from ._error_context import _ErrorContextFormatter from .webhook_config import WebhookConfig if TYPE_CHECKING: @@ -23,9 +20,7 @@ class _WebhookTransport: """Signs and POSTs a JSON string to the configured webhook URL. The single place that owns the HTTP conversation (HMAC-SHA256 signing, - headers, POST). Both the streaming task handler (``_WebhookHandler``) and - the one-shot report sender (``send_report_webhook``) delegate here, so the - transport logic exists exactly once. + headers, POST); ``send_report_webhook`` delegates here. """ def __init__(self, config: WebhookConfig) -> None: @@ -50,87 +45,11 @@ def post(self, payload: str) -> None: pass -class _WebhookFormatter(_ErrorContextFormatter): - """Pure renderer: builds a generic JSON payload from ``record.task_context``.""" - - def __init__( - self, extra_payload: dict[str, Any] | None = None, nest_under: str | None = None - ) -> None: - super().__init__() - self._extra_payload = extra_payload or {} - self._nest_under = nest_under or None - - 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, merged with - any configured ``extra_payload`` keys (which take precedence on - collision). If ``nest_under`` is set, the failure fields are - nested under that key instead of being top-level. - """ - error = self._error_data(record) - generic_payload = self._build_payload(error) - if self._nest_under is not None: - generic_payload = {self._nest_under: generic_payload} - payload = {**generic_payload, **self._extra_payload} - return json.dumps(payload) - - 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: - _WebhookTransport(self._config).post(self.format(record)) - except Exception: - self.handleError(record) - - def _build_report_webhook_payload( entries: dict[str, TaskReportEntry], content: ReportContent, config: WebhookConfig, + process_name: str, ) -> dict[str, Any]: """Build the JSON-serializable payload dict for a report POST. @@ -142,6 +61,9 @@ def _build_report_webhook_payload( Content selection flags (``show_traceback``, ``show_traced_vars``). config : WebhookConfig Transport config used for ``nest_under`` and ``extra_payload``. + process_name : str + Name of the process the report came from (``""`` if unnamed). Always + emitted alongside ``entries`` for a stable schema. Returns ------- @@ -172,7 +94,7 @@ def _build_report_webhook_payload( task_dict["error"] = error_dict tasks_payload[name] = task_dict - generic: dict[str, Any] = {"entries": tasks_payload} + generic: dict[str, Any] = {"process_name": process_name, "entries": tasks_payload} if config.nest_under: generic = {config.nest_under: generic} return {**generic, **config.extra_payload} @@ -200,26 +122,5 @@ def send_report_webhook( When ``True`` only ``ERRORED`` entries are included in the payload. """ entries = report.errored if errors_only else report.entries - payload = _build_report_webhook_payload(entries, content, config) + payload = _build_report_webhook_payload(entries, content, config, report.process_name) _WebhookTransport(config).post(json.dumps(payload)) - - -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(extra_payload=config.extra_payload, nest_under=config.nest_under) - ) - handler.setLevel(logging.ERROR) - return handler diff --git a/src/processes/comms/base.py b/src/processes/comms/base.py index b3302d7..a464059 100644 --- a/src/processes/comms/base.py +++ b/src/processes/comms/base.py @@ -1,14 +1,12 @@ -"""Communication ports: the abstract channel interfaces the domain depends on. +"""Communication port: the abstract report-channel interface the domain depends on. -These abstractions are deliberately a leaf within ``comms`` — they import no -concrete channel, transport, or renderer — so ``task.py`` (streaming) and -``execution_report.py`` (one-shot) can depend on the *interface* without pulling -in the email/webhook implementations. +``ReportChannel`` is deliberately a leaf within ``comms`` — it imports no concrete +channel, transport, or renderer — so ``execution_report.py`` can depend on the +*interface* without pulling in the email/webhook implementations. """ from __future__ import annotations -import logging from abc import ABC, abstractmethod from dataclasses import dataclass from typing import TYPE_CHECKING @@ -17,7 +15,7 @@ from ..execution_report import ProcessExecutionReport -@dataclass(frozen=True) +@dataclass(frozen=True, slots=True) class ReportContent: """What detail a report notification includes. @@ -40,10 +38,9 @@ class ReportContent: class ReportChannel(ABC): """Base class for channels that deliver a finished ``ProcessExecutionReport``. - Unlike ``NotificationChannel`` (which builds a streaming ``logging.Handler`` - for a single ``Task``), a report channel sends a complete report **once**, - after the run. ``ProcessExecutionReport.notify`` iterates the channels it is - given and calls ``send_report`` on each. + A report channel sends a complete report **once**, after the run. + ``ProcessExecutionReport.notify`` iterates the channels it is given and calls + ``send_report`` on each. """ @abstractmethod @@ -58,46 +55,3 @@ def send_report(self, report: ProcessExecutionReport, *, errors_only: bool) -> N If True, only the ``ERRORED`` entries are sent; otherwise the whole report is sent. """ - - -class NotificationChannel(ABC): - """Base class for task notification channels. - - A notification channel knows how to build a configured - ``logging.Handler`` that delivers a task's log records (and, on - failure, its structured failure context) to some destination. ``Task`` - attaches one handler per configured channel to its logger. - - Concrete channels wrap a specific delivery mechanism (e.g. a logfile or - an email alert). New channels can be added by subclassing - ``NotificationChannel`` and implementing ``build_handler``. - """ - - @abstractmethod - def build_handler(self, task_name: str) -> logging.Handler: - """Build a configured handler for the given task. - - Parameters - ---------- - task_name : str - Name of the task the handler will be attached to. - - Returns - ------- - logging.Handler - A handler ready to be added to the task's logger. - """ - - @property - def frame_filter(self) -> str | None: - """Substring selecting the traceback frame to trace local variables of. - - See ``HTMLEmailStyle.traced_vars_frame_filter``. Channels that don't - influence frame selection return ``None`` (the default). - - Returns - ------- - str | None - ``None`` unless overridden by a subclass. - """ - return None diff --git a/src/processes/comms/channels.py b/src/processes/comms/channels.py index 602b890..2a67826 100644 --- a/src/processes/comms/channels.py +++ b/src/processes/comms/channels.py @@ -1,95 +1,38 @@ from __future__ import annotations -import logging from typing import TYPE_CHECKING -from ._email import _build_task_email_handler, send_report_email -from ._logfile import _TaskLogfileFormatter -from ._webhook import _build_task_webhook_handler, send_report_webhook -from .base import NotificationChannel, ReportChannel, ReportContent +from ._email import send_report_email +from ._webhook import send_report_webhook +from .base import ReportChannel, ReportContent from .email_config import HTMLEmailStyle, SMTPConfig from .webhook_config import WebhookConfig if TYPE_CHECKING: from ..execution_report import ProcessExecutionReport -__all__ = [ - "EmailChannel", - "NotificationChannel", - "ReportChannel", - "ReportContent", - "WebhookChannel", -] +__all__ = ["EmailChannel", "ReportChannel", "ReportContent", "WebhookChannel"] -class _FileChannel(NotificationChannel): - """Notification channel that writes task log records to a plain-text file. - - Attributes - ---------- - log_path : str - File path the handler writes to. - level : int - Minimum log level handled. Defaults to ``logging.INFO``. - - Parameters - ---------- - log_path : str - File path the handler writes to. - level : int - Minimum log level handled. Defaults to ``logging.INFO``. - """ - - def __init__(self, log_path: str, level: int = logging.INFO): - self.log_path = log_path - self.level = level - - def build_handler(self, task_name: str) -> logging.Handler: - """Build a ``FileHandler`` writing to ``log_path``. - - 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 ``FileHandler`` at ``level``, formatted with - ``_TaskLogfileFormatter``. - """ - handler = logging.FileHandler(self.log_path) - handler.setLevel(self.level) - handler.setFormatter(_TaskLogfileFormatter()) - return handler - - -class EmailChannel(NotificationChannel, ReportChannel): - """Channel that sends HTML email: a per-task failure alert and/or a report. - - As a ``NotificationChannel`` it builds a streaming handler for a ``Task`` - (one email per failure). As a ``ReportChannel`` it sends a finished - ``ProcessExecutionReport`` once via :meth:`send_report`. The same instance - can serve both roles. +class EmailChannel(ReportChannel): + """Report channel that sends a finished report as a styled HTML email. Attributes ---------- smtp_config : SMTPConfig - SMTP transport configuration for the alert. + SMTP transport configuration. style : HTMLEmailStyle - HTML presentation settings used to render the alert. + HTML presentation settings (palette + language) for the report body. content : ReportContent - Content selection used by :meth:`send_report` (ignored by the per-task - handler). + Content selection used by :meth:`send_report`. Parameters ---------- smtp_config : SMTPConfig - SMTP transport configuration for the alert. + SMTP transport configuration. style : HTMLEmailStyle | None - HTML presentation settings used to render the alert. Defaults to - ``HTMLEmailStyle()`` (modern, neutral, English) when ``None``. + HTML presentation settings. Defaults to ``HTMLEmailStyle()`` + (neutral palette, English) when ``None``. content : ReportContent | None Content selection for report delivery. Defaults to ``ReportContent()`` (everything) when ``None``. @@ -105,34 +48,6 @@ def __init__( self.style = style or HTMLEmailStyle() self.content = content or ReportContent() - def build_handler(self, task_name: str) -> logging.Handler: - """Build an HTML email handler bound to ``task_name``. - - Parameters - ---------- - task_name : str - Name of the task the handler will be attached to, used in the - email subject. - - Returns - ------- - logging.Handler - A handler at ``logging.ERROR`` level that sends a styled HTML - email for each error log record. - """ - return _build_task_email_handler(self.smtp_config, self.style, task_name) - - @property - def frame_filter(self) -> str | None: - """Frame filter sourced from ``style.traced_vars_frame_filter``. - - Returns - ------- - str | None - The configured ``traced_vars_frame_filter``, or ``None``. - """ - return self.style.traced_vars_frame_filter - def send_report(self, report: ProcessExecutionReport, *, errors_only: bool) -> None: """Send the report as a styled HTML email via SMTP. @@ -152,28 +67,24 @@ def send_report(self, report: ProcessExecutionReport, *, errors_only: bool) -> N ) -class WebhookChannel(NotificationChannel, ReportChannel): - """Channel that POSTs JSON: a per-task failure alert and/or a report. +class WebhookChannel(ReportChannel): + """Report channel that POSTs a finished report as generic JSON. - As a ``NotificationChannel`` it builds a streaming handler for a ``Task`` - (one POST per failure). As a ``ReportChannel`` it POSTs a finished - ``ProcessExecutionReport`` once via :meth:`send_report`. The payload is - generic JSON, so it can be consumed directly or transformed by downstream - relays (Slack/Discord/Teams adapters, custom alerting servers); it is not - coupled to any specific service. + The payload is generic JSON, so it can be consumed directly or transformed + by downstream relays (Slack/Discord/Teams adapters, custom alerting + servers); it is not coupled to any specific service. Attributes ---------- webhook_config : WebhookConfig - Webhook transport configuration for the alert. + Webhook transport configuration. content : ReportContent - Content selection used by :meth:`send_report` (ignored by the per-task - handler). + Content selection used by :meth:`send_report`. Parameters ---------- webhook_config : WebhookConfig - Webhook transport configuration for the alert. + Webhook transport configuration. content : ReportContent | None Content selection for report delivery. Defaults to ``ReportContent()`` (everything) when ``None``. @@ -183,23 +94,6 @@ def __init__(self, webhook_config: WebhookConfig, content: ReportContent | None self.webhook_config = webhook_config self.content = content or ReportContent() - 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) - def send_report(self, report: ProcessExecutionReport, *, errors_only: bool) -> None: """POST the report as a signed JSON payload. diff --git a/src/processes/comms/email_config.py b/src/processes/comms/email_config.py index e10489c..3941ffe 100644 --- a/src/processes/comms/email_config.py +++ b/src/processes/comms/email_config.py @@ -3,11 +3,9 @@ import ssl from dataclasses import dataclass -_VALID_STYLES = frozenset({"classic", "modern", "compact"}) _VALID_PALETTES = frozenset({"neutral", "catppuccin", "neobones", "slate"}) _VALID_LANGUAGES = frozenset({"en", "es", "pt", "fr", "de", "it"}) -_DEFAULT_STYLE = "modern" _DEFAULT_PALETTE = "neutral" _DEFAULT_LANGUAGE = "en" @@ -43,47 +41,32 @@ class SMTPConfig: timeout: int = 5 -@dataclass(frozen=True) +@dataclass(frozen=True, slots=True) class HTMLEmailStyle: - """HTML presentation settings for email error alerts. + """HTML presentation settings for the report email. - All fields default to a modern, neutral, English email — pass only - what you want to override. + Both fields default to a neutral, English email — pass only what you want + to override. Attributes ---------- - style : str - Layout to use: ``"classic"``, ``"modern"``, or ``"compact"``. - Defaults to ``"modern"``. palette : str Color scheme: ``"neutral"``, ``"catppuccin"``, ``"neobones"``, or ``"slate"``. Defaults to ``"neutral"``. language : str ISO 639-1 code for the email body text: ``"en"``, ``"es"``, ``"pt"``, ``"fr"``, ``"de"``, or ``"it"``. Defaults to ``"en"``. - traced_vars_frame_filter : str | None - Substring used to select the traceback frame whose local variables - appear in the *Traced Variables* section. When ``None`` (default), - the outermost user frame is used; when set, the outermost frame whose - filename contains this substring is used instead. Raises ------ ValueError - If ``style``, ``palette``, or ``language`` is not one of the - supported values. - TypeError - If ``traced_vars_frame_filter`` is neither ``str`` nor ``None``. + If ``palette`` or ``language`` is not one of the supported values. """ - style: str = _DEFAULT_STYLE palette: str = _DEFAULT_PALETTE language: str = _DEFAULT_LANGUAGE - traced_vars_frame_filter: str | None = None def __post_init__(self) -> None: - if self.style not in _VALID_STYLES: - raise ValueError(f"style must be one of {sorted(_VALID_STYLES)}, got {self.style!r}") if self.palette not in _VALID_PALETTES: raise ValueError( f"palette must be one of {sorted(_VALID_PALETTES)}, got {self.palette!r}" @@ -92,10 +75,3 @@ def __post_init__(self) -> None: raise ValueError( f"language must be one of {sorted(_VALID_LANGUAGES)}, got {self.language!r}" ) - if self.traced_vars_frame_filter is not None and not isinstance( - self.traced_vars_frame_filter, str - ): - raise TypeError( - f"traced_vars_frame_filter must be a str or None. " - f"Got {type(self.traced_vars_frame_filter)}" - ) diff --git a/src/processes/comms/themes/languages/de.json b/src/processes/comms/themes/languages/de.json index 62a24e8..b43c9f1 100644 --- a/src/processes/comms/themes/languages/de.json +++ b/src/processes/comms/themes/languages/de.json @@ -1,17 +1,10 @@ { - "lang_title_prefix": "Pipeline-Fehler: ", - "lang_failure_header": "Pipeline-Fehlschlag: ", - "lang_failure_header_short": "fehler", "lang_function_label": "Funktion", - "lang_args_label": "Argumente", - "lang_kwargs_label": "Schlüsselwortargumente", "lang_exception_label": "Ausnahme", "lang_downstream_title": "Auswirkung auf nachgelagerte Aufgaben", - "lang_downstream_blurb": "Die folgenden nachgelagerten Aufgaben werden übersprungen:", "lang_traceback_title": "Fehlerverlauf", "lang_traced_vars_title": "Verfolgte Variablen", "lang_traced_vars_blurb": "Die folgenden lokalen Variablen hatten diese Werte in {location}:", - "lang_email_subject": "Fehler in Aufgabe ", "lang_report_title_prefix": "Prozessbericht", "lang_report_header": "Prozessausführungsbericht", "lang_report_header_errors_only": "Bericht fehlgeschlagener Aufgaben", @@ -24,4 +17,4 @@ "lang_status_success": "Erfolg", "lang_status_errored": "Fehler", "lang_status_skipped": "Übersprungen" -} \ No newline at end of file +} diff --git a/src/processes/comms/themes/languages/en.json b/src/processes/comms/themes/languages/en.json index 33af00b..85dbae7 100644 --- a/src/processes/comms/themes/languages/en.json +++ b/src/processes/comms/themes/languages/en.json @@ -1,17 +1,10 @@ { - "lang_title_prefix": "Pipeline Error: ", - "lang_failure_header": "Pipeline Failure: ", - "lang_failure_header_short": "failure", "lang_function_label": "Function", - "lang_args_label": "Args", - "lang_kwargs_label": "Kwargs", "lang_exception_label": "Exception", "lang_downstream_title": "Downstream Impact", - "lang_downstream_blurb": "The following downstream tasks will be skipped:", "lang_traceback_title": "Traceback", "lang_traced_vars_title": "Traced Variables", "lang_traced_vars_blurb": "The following local variables had these values at {location}:", - "lang_email_subject": "Error in task ", "lang_report_title_prefix": "Process Report", "lang_report_header": "Process Execution Report", "lang_report_header_errors_only": "Failed Tasks Report", @@ -24,4 +17,4 @@ "lang_status_success": "Success", "lang_status_errored": "Error", "lang_status_skipped": "Skipped" -} \ No newline at end of file +} diff --git a/src/processes/comms/themes/languages/es.json b/src/processes/comms/themes/languages/es.json index 0370bdd..61f2a43 100644 --- a/src/processes/comms/themes/languages/es.json +++ b/src/processes/comms/themes/languages/es.json @@ -1,17 +1,10 @@ { - "lang_title_prefix": "Error en el pipeline: ", - "lang_failure_header": "Fallo en el pipeline: ", - "lang_failure_header_short": "fallo", "lang_function_label": "Función", - "lang_args_label": "Argumentos", - "lang_kwargs_label": "Argumentos con nombre", "lang_exception_label": "Excepción", "lang_downstream_title": "Impacto en tareas dependientes", - "lang_downstream_blurb": "Las siguientes tareas dependientes se omitirán:", "lang_traceback_title": "Traza de error", "lang_traced_vars_title": "Variables rastreadas", "lang_traced_vars_blurb": "Las siguientes variables locales tenían estos valores en {location}:", - "lang_email_subject": "Error en la tarea ", "lang_report_title_prefix": "Informe del Proceso", "lang_report_header": "Informe de Ejecución del Proceso", "lang_report_header_errors_only": "Informe de Tareas Fallidas", @@ -24,4 +17,4 @@ "lang_status_success": "Éxito", "lang_status_errored": "Error", "lang_status_skipped": "Omitida" -} \ No newline at end of file +} diff --git a/src/processes/comms/themes/languages/fr.json b/src/processes/comms/themes/languages/fr.json index 1ed90e6..636fb81 100644 --- a/src/processes/comms/themes/languages/fr.json +++ b/src/processes/comms/themes/languages/fr.json @@ -1,17 +1,10 @@ { - "lang_title_prefix": "Erreur du pipeline : ", - "lang_failure_header": "Échec du pipeline : ", - "lang_failure_header_short": "échec", "lang_function_label": "Fonction", - "lang_args_label": "Arguments", - "lang_kwargs_label": "Arguments nommés", "lang_exception_label": "Exception", "lang_downstream_title": "Impact sur les tâches dépendantes", - "lang_downstream_blurb": "Les tâches dépendantes suivantes seront ignorées :", "lang_traceback_title": "Trace d'erreur", "lang_traced_vars_title": "Variables suivies", "lang_traced_vars_blurb": "Les variables locales suivantes avaient ces valeurs à {location} :", - "lang_email_subject": "Erreur dans la tâche ", "lang_report_title_prefix": "Rapport de Processus", "lang_report_header": "Rapport d'Exécution du Processus", "lang_report_header_errors_only": "Rapport des Tâches Échouées", @@ -24,4 +17,4 @@ "lang_status_success": "Succès", "lang_status_errored": "Erreur", "lang_status_skipped": "Ignorée" -} \ No newline at end of file +} diff --git a/src/processes/comms/themes/languages/it.json b/src/processes/comms/themes/languages/it.json index 51cc08e..6fd9230 100644 --- a/src/processes/comms/themes/languages/it.json +++ b/src/processes/comms/themes/languages/it.json @@ -1,17 +1,10 @@ { - "lang_title_prefix": "Errore del pipeline: ", - "lang_failure_header": "Fallimento del pipeline: ", - "lang_failure_header_short": "fallimento", "lang_function_label": "Funzione", - "lang_args_label": "Argomenti", - "lang_kwargs_label": "Argomenti con nome", "lang_exception_label": "Eccezione", "lang_downstream_title": "Impatto sulle attività dipendenti", - "lang_downstream_blurb": "Le seguenti attività dipendenti verranno saltate:", "lang_traceback_title": "Traccia dell'errore", "lang_traced_vars_title": "Variabili tracciate", "lang_traced_vars_blurb": "Le seguenti variabili locali avevano questi valori in {location}:", - "lang_email_subject": "Errore nell'attività ", "lang_report_title_prefix": "Rapporto di Processo", "lang_report_header": "Rapporto di Esecuzione del Processo", "lang_report_header_errors_only": "Rapporto Attività Fallite", @@ -24,4 +17,4 @@ "lang_status_success": "Successo", "lang_status_errored": "Errore", "lang_status_skipped": "Saltata" -} \ No newline at end of file +} diff --git a/src/processes/comms/themes/languages/pt.json b/src/processes/comms/themes/languages/pt.json index 7dbe12f..2f1633c 100644 --- a/src/processes/comms/themes/languages/pt.json +++ b/src/processes/comms/themes/languages/pt.json @@ -1,17 +1,10 @@ { - "lang_title_prefix": "Erro no pipeline: ", - "lang_failure_header": "Falha no pipeline: ", - "lang_failure_header_short": "falha", "lang_function_label": "Função", - "lang_args_label": "Argumentos", - "lang_kwargs_label": "Argumentos nomeados", "lang_exception_label": "Exceção", "lang_downstream_title": "Impacto nas tarefas dependentes", - "lang_downstream_blurb": "As seguintes tarefas dependentes serão ignoradas:", "lang_traceback_title": "Rastreamento de erro", "lang_traced_vars_title": "Variáveis rastreadas", "lang_traced_vars_blurb": "As seguintes variáveis locais tinham estes valores em {location}:", - "lang_email_subject": "Erro na tarefa ", "lang_report_title_prefix": "Relatório do Processo", "lang_report_header": "Relatório de Execução do Processo", "lang_report_header_errors_only": "Relatório de Tarefas com Falha", @@ -24,4 +17,4 @@ "lang_status_success": "Sucesso", "lang_status_errored": "Erro", "lang_status_skipped": "Ignorada" -} \ No newline at end of file +} diff --git a/src/processes/comms/themes/styles/classic.html b/src/processes/comms/themes/styles/classic.html deleted file mode 100644 index 13448c5..0000000 --- a/src/processes/comms/themes/styles/classic.html +++ /dev/null @@ -1,49 +0,0 @@ - - - - - {{lang_title_prefix}}{{task_name}} - - - -

    {{lang_failure_header}}{{task_name}}

    - -
    -

    {{lang_function_label}}: {{function}}

    -

    {{lang_args_label}}: {{args}}

    -

    {{lang_kwargs_label}}: {{kwargs}}

    -

    {{lang_exception_label}}: {{exception}}

    -
    - -
    -

    {{lang_downstream_title}}

    -

    {{lang_downstream_blurb}}

    -
      {{downstream_items}}
    -
    - -

    {{lang_traceback_title}}

    - 
    {{traceback_before}}{{traceback_highlight}}{{traceback_after}}
    - -

    {{lang_traced_vars_title}}

    -

    {{lang_traced_vars_blurb}}

    - 
    {{traced_vars}}
    - - \ No newline at end of file diff --git a/src/processes/comms/themes/styles/compact.html b/src/processes/comms/themes/styles/compact.html deleted file mode 100644 index d64845e..0000000 --- a/src/processes/comms/themes/styles/compact.html +++ /dev/null @@ -1,67 +0,0 @@ - - - - - {{lang_title_prefix}}{{task_name}} - - - -
    -

    [{{lang_failure_header_short}}] {{task_name}}

    - -
    [{{lang_function_label}}] {{function}}
    -
    [{{lang_args_label}}] {{args}}
    -
    [{{lang_kwargs_label}}] {{kwargs}}
    -
    [{{lang_exception_label}}] {{exception}}
    - -
    -

    [{{lang_downstream_title}}]

    -
      {{downstream_items}}
    -
    - -
    -
    [{{lang_traceback_title}}]
    - 
    {{traceback_before}}{{traceback_highlight}}{{traceback_after}}
    -
    - -
    -
    [{{lang_traced_vars_title}}]
    -
    {{lang_traced_vars_blurb}}
    - 
    {{traced_vars}}
    -
    -
    - - \ No newline at end of file diff --git a/src/processes/comms/themes/styles/modern.html b/src/processes/comms/themes/styles/modern.html deleted file mode 100644 index 0159975..0000000 --- a/src/processes/comms/themes/styles/modern.html +++ /dev/null @@ -1,115 +0,0 @@ - - - - - {{lang_title_prefix}}{{task_name}} - - - -
    -
    {{lang_failure_header}}{{task_name}}
    -
    -
    -
    {{lang_function_label}}
    -
    {{function}}
    -
    -
    -
    {{lang_args_label}}
    -
    {{args}}
    -
    -
    -
    {{lang_kwargs_label}}
    -
    {{kwargs}}
    -
    -
    -
    {{lang_exception_label}}
    -
    {{exception}}
    -
    - -
    -

    {{lang_downstream_title}}

    -

    {{lang_downstream_blurb}}

    -
      {{downstream_items}}
    -
    - -
    {{lang_traceback_title}}
    - 
    {{traceback_before}}{{traceback_highlight}}{{traceback_after}}
    - -
    {{lang_traced_vars_title}}
    -
    {{lang_traced_vars_blurb}}
    - 
    {{traced_vars}}
    -
    -
    - - \ No newline at end of file diff --git a/src/processes/error_data.py b/src/processes/error_data.py index c584280..08cd439 100644 --- a/src/processes/error_data.py +++ b/src/processes/error_data.py @@ -4,7 +4,7 @@ from typing import Any -@dataclass(frozen=True) +@dataclass(frozen=True, slots=True) class ErrorData: """Typed view of a task failure, extracted from ``record.task_context``. diff --git a/src/processes/execution_report.py b/src/processes/execution_report.py index 14e2d62..80f5e9f 100644 --- a/src/processes/execution_report.py +++ b/src/processes/execution_report.py @@ -35,7 +35,7 @@ def _json_default(obj: Any) -> Any: return repr(obj) -@dataclass(frozen=True) +@dataclass(frozen=True, slots=True) class TaskReportEntry: """Per-task entry in a :class:`ProcessExecutionReport`. @@ -73,7 +73,7 @@ class TaskReportEntry: error: ErrorData | None = None -@dataclass(frozen=True) +@dataclass(frozen=True, slots=True) class ProcessExecutionReport: """Per-task breakdown of a finished :meth:`Process.run` call. @@ -82,9 +82,13 @@ class ProcessExecutionReport: entries : dict[str, TaskReportEntry] Mapping of task name to its report entry, ordered the same way as ``process.tasks`` (topological order). + process_name : str + Name of the process the report came from (``""`` if unnamed). Used to + label notifications such as the email subject. """ entries: dict[str, TaskReportEntry] = field(default_factory=dict) + process_name: str = "" def _filter(self, status: TaskStatus) -> dict[str, TaskReportEntry]: return {name: entry for name, entry in self.entries.items() if entry.status == status} @@ -138,7 +142,7 @@ def from_results( result=res.result if res.worked else None, error=res.error_data if res.status == TaskStatus.ERRORED else None, ) - return cls(entries) + return cls(entries, process.name) def to_json(self, *, indent: int | None = None, **dumps_kwargs: Any) -> str: """Serialize the whole report to a JSON string without dropping any field. @@ -212,5 +216,6 @@ def _for_tasks(self, tasks: list[str]) -> ProcessExecutionReport: """A copy of this report restricted to ``tasks`` (case-insensitive names).""" wanted = {name.lower() for name in tasks} return ProcessExecutionReport( - {name: entry for name, entry in self.entries.items() if name.lower() in wanted} + {name: entry for name, entry in self.entries.items() if name.lower() in wanted}, + self.process_name, ) diff --git a/src/processes/process.py b/src/processes/process.py index 477cc79..ffd951e 100644 --- a/src/processes/process.py +++ b/src/processes/process.py @@ -26,6 +26,8 @@ class Process: ---------- tasks : list[Task] List of tasks to be executed, automatically sorted by dependencies. + name : str + Human-readable name for the process (``""`` if unnamed). runner : ProcessRunner The runner responsible for executing the tasks. @@ -35,6 +37,10 @@ class Process: The tasks to orchestrate. Order does not matter — the constructor topologically sorts the list in place. The list is mutated during construction; pass a copy if the original ordering matters. + name : str, optional + Human-readable name for the process. Recorded on the resulting + ``ProcessExecutionReport`` and used to label notifications (e.g. the + email subject). Defaults to ``""`` (unnamed). Raises ------ @@ -48,8 +54,9 @@ class Process: If circular dependencies are detected among tasks. """ - def __init__(self, tasks: list[Task]): + def __init__(self, tasks: list[Task], name: str = ""): self.tasks = tasks + self.name = name try: self._validate_and_sort() diff --git a/src/processes/task.py b/src/processes/task.py index 53aae5a..c1e5317 100644 --- a/src/processes/task.py +++ b/src/processes/task.py @@ -10,9 +10,8 @@ import logging +from ._logfile import _TaskLogfileFormatter from ._tb_utils import _build_traced_vars, _build_traced_vars_location, _format_traceback -from .comms.base import NotificationChannel -from .comms.channels import _FileChannel from .error_data import ErrorData from .exceptions import CircularDependencyError from .task_types import TaskDependency, TaskResult, TaskStatus @@ -43,10 +42,10 @@ class Task: Keyword arguments to pass to the function. Defaults to empty dict. dependencies : list[TaskDependency] List of tasks this task depends on. Defaults to empty list. - channels : list[NotificationChannel] - Additional notification channels attached to this task's logger, on - top of the implicit file channel built from ``log_path``. Defaults to - empty list. + traced_vars_frame_filter : str | None + Substring selecting which traceback frame's local variables are + captured into the failure context on error. ``None`` selects the + outermost user frame. Defaults to ``None``. timeout : float | None Seconds allowed per attempt before a ``TimeoutError`` is raised. ``None`` means no limit. Defaults to ``None``. @@ -63,15 +62,17 @@ class Task: Parameters ---------- name : str - Unique task name; must not contain spaces. + Unique task name; must not contain spaces. Normalized to lowercase, + so task names (and the dependency references that point at them) are + matched case-insensitively. func : Callable[..., Any] The callable executed when the task runs. log_path : str | None File path the task's log records are written to (one ``FileHandler`` at ``INFO`` level, format - ``"%(asctime)s - %(name)s - %(levelname)s - %(message)s"``). ``None`` - means no file logging is configured. If this leaves the task with no - notification channels at all, a ``NullHandler`` is attached instead. + ``"%(asctime)s - %(name)s - %(levelname)s - %(message)s"``, with the + structured failure context appended on error). ``None`` means no file + logging is configured, and a ``NullHandler`` is attached instead. Defaults to ``None``. args : tuple[Any, ...] Positional arguments forwarded to ``func``. Defaults to ``()``. @@ -80,13 +81,12 @@ class Task: an empty dict. dependencies : list[TaskDependency] | None Tasks this task depends on. ``None`` is treated as an empty list. - channels : list[NotificationChannel] | None - Additional notification channels whose handlers are attached to this - task's logger, alongside the implicit file channel built from - ``log_path``. Use ``EmailChannel`` for HTML email alerts on - ``logging.ERROR`` and above, or subclass ``NotificationChannel`` for - other destinations. ``None`` is treated as an empty list. Defaults to - ``None``. + traced_vars_frame_filter : str | None + Substring used to select the traceback frame whose local variables are + captured into the failure context (and thus into both the logfile and + any report notification). When ``None`` (default), the outermost user + frame is used; when set, the outermost frame whose filename contains + this substring is used instead. timeout : float | None Seconds allowed per attempt before ``TimeoutError`` is raised for that attempt. ``None`` means no limit. When a timeout fires, the underlying @@ -105,8 +105,8 @@ class Task: TypeError If any parameter is not of the expected type, ``timeout`` is not a positive number, ``retries`` is negative, ``retry_on`` is not a - tuple of ``Exception`` subclasses, or ``channels`` is not a list of - ``NotificationChannel`` instances. + tuple of ``Exception`` subclasses, or ``traced_vars_frame_filter`` is + neither ``str`` nor ``None``. ValueError If ``name`` contains a space, if the same dependency name is listed more than once, or if the task lists itself as a @@ -124,15 +124,18 @@ def __init__( args: tuple[Any, ...] = (), kwargs: dict[str, Any] | None = None, dependencies: list[TaskDependency] | None = None, - channels: list[NotificationChannel] | None = None, + traced_vars_frame_filter: str | None = None, timeout: float | None = None, retries: int | None = 0, retry_on: tuple[type[Exception], ...] | None = None, ): - self.name = name + if not isinstance(name, str): + raise TypeError(f"name must be str. Got {type(name)}") + self.name = name.lower() self.log_path = log_path self.func = func self.args = args + self.traced_vars_frame_filter = traced_vars_frame_filter self.timeout = timeout self.retries = retries if retries is not None else 0 self.retry_on = retry_on @@ -145,10 +148,6 @@ def __init__( self.dependencies = [] else: self.dependencies = dependencies - if channels is None: - self.channels = [] - else: - self.channels = channels self._check_input_types() if " " in self.name: @@ -165,20 +164,15 @@ def __init__( logger = logging.getLogger(f"processes.{self.name}.{id(self)}") logger.setLevel(logging.DEBUG) - file_channels: list[NotificationChannel] = ( - [_FileChannel(self.log_path)] if self.log_path is not None else [] - ) - all_channels: list[NotificationChannel] = [*file_channels, *self.channels] - - if all_channels: - for channel in all_channels: - logger.addHandler(channel.build_handler(self.name)) + if self.log_path is not None: + file_handler = logging.FileHandler(self.log_path) + file_handler.setLevel(logging.INFO) + file_handler.setFormatter(_TaskLogfileFormatter()) + logger.addHandler(file_handler) else: logger.addHandler(logging.NullHandler()) - self._frame_filter: str | None = next( - (c.frame_filter for c in all_channels if c.frame_filter is not None), None - ) + self._frame_filter: str | None = self.traced_vars_frame_filter self.logger = logger def _check_input_types(self) -> None: @@ -208,12 +202,13 @@ def _check_input_types(self) -> None: f"dependency must be of type TaskDependency. Got {type(dependency)}" ) - if not isinstance(self.channels, list): - raise TypeError(f"channels must be list. Got {type(self.channels)}") - - for channel in self.channels: - if not isinstance(channel, NotificationChannel): - raise TypeError(f"channel must be of type NotificationChannel. Got {type(channel)}") + if self.traced_vars_frame_filter is not None and not isinstance( + self.traced_vars_frame_filter, str + ): + raise TypeError( + f"traced_vars_frame_filter must be a str or None. " + f"Got {type(self.traced_vars_frame_filter)}" + ) if self.timeout is not None and ( not isinstance(self.timeout, (int, float)) or self.timeout <= 0 diff --git a/src/processes/task_types.py b/src/processes/task_types.py index 6f11eb0..f1d26bd 100644 --- a/src/processes/task_types.py +++ b/src/processes/task_types.py @@ -139,7 +139,8 @@ class TaskDependency: Attributes ---------- task_name : str - The name of the task this dependency refers to. + The name of the task this dependency refers to. Normalized to lowercase + to match :class:`Task` names case-insensitively. use_result_as_additional_args : bool If True, the result of the dependency task will be passed as an additional positional argument as the last argument. Defaults to False. @@ -165,13 +166,13 @@ def __init__( use_result_as_additional_kwargs: bool = False, additional_kwarg_name: str = "", ): - self.task_name = task_name + if not isinstance(task_name, str): + raise TypeError(f"task_name must be of type str. Got {type(task_name)}") + self.task_name = task_name.lower() self.use_result_as_additional_args = use_result_as_additional_args self.use_result_as_additional_kwargs = use_result_as_additional_kwargs self.additional_kwarg_name = additional_kwarg_name - if not isinstance(self.task_name, str): - raise TypeError(f"task_name must be of type str. Got {type(self.task_name)}") if not isinstance(self.use_result_as_additional_args, bool): raise TypeError( f"use_result_as_additional_args must be of type bool. " diff --git a/tests/manual_tests/manual_pipeline_inspect.py b/tests/manual_tests/manual_pipeline_inspect.py deleted file mode 100644 index 9982771..0000000 --- a/tests/manual_tests/manual_pipeline_inspect.py +++ /dev/null @@ -1,423 +0,0 @@ -"""Manual end-to-end inspection of a real pipeline run. - -This script is **not** picked up by pytest — see ``tests/conftest.py`` and -the ``python_files = "test_*.py"`` setting in ``pytest.ini``. Run it by -hand to inspect everything the framework emits in one shot: - - * Per-task console output (which functions fired, in what order). - * Per-task log files under ``tests/manual_tests/logs/`` (one ``.log`` - per task — see the ``FileHandler`` attached in ``Task.__init__``). - * The HTML email sent via ``SMTPConfig`` to maildev (open the - web UI and check the rendered "Downstream Impact" list). - -Prerequisites -------------- -You only need **Node.js + maildev** available somewhere on your system -(``npm install -g maildev`` is the usual install). The script will -launch maildev for you if it isn't already running; if you started it -yourself, the script will detect and reuse it. - - The script connects to ``127.0.0.1`` (not ``localhost``) on purpose: - on Windows, ``localhost`` often resolves to IPv6 ``::1`` first while - maildev binds IPv4 only, producing ``WinError 10061`` (connection - refused). If your maildev uses a different host/port, edit - ``SMTP_HOST`` / ``SMTP_PORT`` / ``WEB_PORT`` below. - -From the project root, run: - - python tests/manual_tests/manual_pipeline_inspect.py - -Use ``--keep-logs`` to preserve any previous ``.log`` files instead of -wiping them at startup. Use ``--no-start-maildev`` if you want to -manage maildev yourself and have the script only verify it's already -listening. - -Then inspect: - * The console output below for execution order. - * The per-task files in ``tests/manual_tests/logs/``. - * The maildev web UI at http://localhost:1080. - -Expected outcomes ------------------ -* Tasks ``A0_ingest_orders``, ``B_validate_orders``, ``C_enrich_with_users``, - ``E_compute_kpis`` and ``H_archive_run_metadata`` succeed (their - ``func`` is called exactly once). -* ``D_join_inventory`` raises a ``json.JSONDecodeError`` from inside - the stdlib ``json`` package after passing through 4 local frames - (``join_inventory`` + 3 helpers). The rendered email traceback - should therefore show our local frames interleaved with frames - from ``json.decoder`` / ``json.scanner`` / ``json.__init__``. -* ``F_publish_dashboard`` raises (its ``func`` is called once and - then re-raises). -* Task ``G_notify_ops`` never runs (cascading skip from - ``F_publish_dashboard``). -* Two emails arrive in maildev, one per failure, each with the matching - "Downstream Impact" entries. - -Why sequential mode? --------------------- -maildev's SMTP listener is single-threaded. When the pipeline runs in -parallel and two tasks fail at the same time, the handler tries to open -two SMTP connections simultaneously; maildev accepts the first and -refuses the second (``ConnectionRefusedError``). To keep this manual -script deterministic without changing the library, the pipeline runs -in sequential mode here. The library itself supports both modes — see -the integration test for the parallel path. -""" - -from __future__ import annotations - -import json -import os -import sys -import traceback -from collections.abc import Callable -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 EmailChannel, Process, SMTPConfig, Task, TaskDependency # noqa: E402 - -# --------------------------------------------------------------------------- # -# Maildev wiring # -# --------------------------------------------------------------------------- # - -SMTP_HOST = "127.0.0.1" # avoid Windows IPv6/localhost resolution quirks -SMTP_PORT = 1025 # maildev's default SMTP port (web UI runs on 1080) -WEB_PORT = 1080 # maildev's default web UI port -FROM_ADDR = "pipeline-alerts@enterprise.test" -TO_ADDRS = ["sre-oncall@enterprise.test"] - - -def _make_smtp_config() -> SMTPConfig: - return SMTPConfig( - mailhost=(SMTP_HOST, SMTP_PORT), - fromaddr=FROM_ADDR, - toaddrs=TO_ADDRS, - timeout=5, - ) - - -# --------------------------------------------------------------------------- # -# Task functions # -# --------------------------------------------------------------------------- # - - -def ingest_orders( - feed: str, - batch_size: int = 500, - dry_run: bool = False, -) -> dict[str, Any]: - """Pull raw orders from the source feed. Always succeeds.""" - print(f" [ingest_orders] feed={feed!r} batch_size={batch_size} dry_run={dry_run}") - return { - "feed": feed, - "rows": [ - {"order_id": "O-1001", "user_id": "U-7", "sku": "SKU-A", "amount": 49.90}, - {"order_id": "O-1002", "user_id": "U-12", "sku": "SKU-B", "amount": 12.00}, - {"order_id": "O-1003", "user_id": "U-7", "sku": "SKU-C", "amount": 3.50}, - ], - "batch_size": batch_size, - "dry_run": dry_run, - } - - -def validate_orders( - schema_version: str, - payload: dict[str, Any] | None = None, - strict: bool = True, -) -> dict[str, Any]: - """Validate the ingested batch against the schema. Uses kwarg injection.""" - rows = (payload or {}).get("rows", []) - print(f" [validate_orders] schema={schema_version} rows={len(rows)} strict={strict}") - if not rows: - raise ValueError("empty payload — schema validation cannot proceed") - return { - "schema_version": schema_version, - "row_count": len(rows), - "strict": strict, - } - - -def enrich_with_users( - users_source: str, - payload: dict[str, Any] | None = None, - timeout: int = 10, -) -> dict[str, Any]: - """Join each order with its user record. Uses positional injection.""" - rows = (payload or {}).get("rows", []) - print(f" [enrich_with_users] source={users_source} rows={len(rows)} timeout={timeout}s") - return { - "enriched_rows": [ - {**row, "user_email": f"user{row['user_id']}@example.test"} for row in rows - ], - "timeout": timeout, - } - - -def _inventory_chunk_header(warehouse: str, strategy: str) -> str: - """Local frame 1/4 — assemble the cached chunk's opening text and - forward the call down the chain so this frame stays on the stack - when the decoder later raises.""" - return _inventory_chunk_seal(f'{{"warehouse": "{warehouse}", "strategy": "{strategy}",') - - -def _inventory_chunk_seal(header: str) -> str: - """Local frame 2/4 — append the body. Intentionally truncated mid-row - to simulate a cache-layer corruption (a partial write that the - downstream decoder then rejects with ``json.JSONDecodeError``).""" - return _inventory_chunk_parse( - header + ' "rows": [{"sku": "SKU-A", "qty": 12}, {"sku": "SKU-B", "q' - ) - - -def _inventory_chunk_parse(chunk: str) -> Any: - """Local frame 3/4 — hand the chunk to the stdlib JSON decoder. The - raise lives a few frames deep in ``json.decoder`` / ``json.scanner``, - which is the point: the rendered email traceback should show our - 4 local frames interleaved with stdlib frames so it's obvious the - formatter isn't trimming either side.""" - return json.loads(chunk) - - -def join_inventory( - warehouse: str, - payload: dict[str, Any] | None = None, - strategy: str = "inner", -) -> dict[str, Any]: - """Join with inventory levels. **Fails** — a corrupt cached chunk - surfaces a ``json.JSONDecodeError`` from inside the stdlib ``json`` - package. Call chain on failure: ``join_inventory`` → - ``_inventory_chunk_header`` → ``_inventory_chunk_seal`` → - ``_inventory_chunk_parse`` → ``json.loads`` → ``json.scanner`` / - ``json.decoder``.""" - rows = (payload or {}).get("enriched_rows", []) - print(f" [join_inventory] warehouse={warehouse} rows={len(rows)} strategy={strategy}") - return _inventory_chunk_header(warehouse, strategy) - - -def compute_kpis( - window: str, - validated: dict[str, Any] | None = None, - currency: str = "USD", -) -> dict[str, Any]: - """Aggregate KPIs from the validated + enriched chain. Uses kwarg injection.""" - row_count = (validated or {}).get("row_count", 0) - print(f" [compute_kpis] window={window} rows={row_count} currency={currency}") - return { - "window": window, - "kpi_rows": row_count, - "currency": currency, - "gmv": 65.40, - } - - -def publish_dashboard( - audience: str, - kpis: dict[str, Any] | None = None, - cache_ttl: int = 60, -) -> dict[str, Any]: - """Push KPIs to the dashboard. **Fails** — service unreachable.""" - print(f" [publish_dashboard] audience={audience} cache_ttl={cache_ttl}s") - raise ConnectionError( - f"dashboard service unreachable (audience={audience!r}, cache_ttl={cache_ttl})" - ) - - -def notify_ops(*_args: Any, **_kwargs: Any) -> str: - """Post a message to the ops channel. Cascading-skip target.""" - print(" [notify_ops] this should never run — FAILED upstream") - return "ops-notified" - - -def archive_run_metadata( - destination: str, - retention_days: int = 30, -) -> str: - """Archive the run manifest. Runs in parallel with publish_dashboard.""" - print(f" [archive_run_metadata] destination={destination} retention_days={retention_days}") - return f"archived://{destination}?retention={retention_days}d" - - -# --------------------------------------------------------------------------- # -# Build the DAG # -# --------------------------------------------------------------------------- # - - -def _log_path(logs_dir: str, name: str) -> str: - return os.path.join(logs_dir, f"{name}.log") - - -def build_tasks(logs_dir: str, smtp: SMTPConfig) -> list[Task]: - dep = TaskDependency - - def _task( - name: str, - func: Callable[..., Any], - args: tuple[Any, ...] = (), - kwargs: dict[str, Any] | None = None, - deps: list[TaskDependency] | None = None, - ) -> Task: - return Task( - name=name, - log_path=_log_path(logs_dir, name), - func=func, - args=args, - kwargs=kwargs or {}, - dependencies=deps or [], - channels=[EmailChannel(smtp)], - ) - - return [ - # Root - _task( - "A0_ingest_orders", - ingest_orders, - args=("orders_feed",), - kwargs={"batch_size": 1000, "dry_run": False}, - ), - # Validation branch (kwarg injection) - _task( - "B_validate_orders", - validate_orders, - args=("v2",), - kwargs={"strict": True}, - deps=[ - dep( - "A0_ingest_orders", - use_result_as_additional_kwargs=True, - additional_kwarg_name="payload", - ), - ], - ), - # Enrichment branch (positional injection) - _task( - "C_enrich_with_users", - enrich_with_users, - args=("users_api",), - kwargs={"timeout": 30}, - deps=[ - dep( - "A0_ingest_orders", - use_result_as_additional_args=True, - ), - ], - ), - # Inventory join (positional injection) — **FAILS** - _task( - "D_join_inventory", - join_inventory, - args=("warehouse_eu",), - kwargs={"strategy": "outer"}, - deps=[ - dep( - "C_enrich_with_users", - use_result_as_additional_args=True, - ), - ], - ), - # KPI aggregation (kwarg injection from B, deps on B and C) - _task( - "E_compute_kpis", - compute_kpis, - args=("daily",), - kwargs={"currency": "USD"}, - deps=[ - dep( - "B_validate_orders", - use_result_as_additional_kwargs=True, - additional_kwarg_name="validated", - ), - dep("C_enrich_with_users"), - ], - ), - # Dashboard publish (positional injection) — **FAILS** - _task( - "F_publish_dashboard", - publish_dashboard, - args=("executive",), - kwargs={"cache_ttl": 60}, - deps=[ - dep( - "E_compute_kpis", - use_result_as_additional_args=True, - ), - ], - ), - # Ops notification — **SKIPPED** (F failed) - _task( - "G_notify_ops", - notify_ops, - kwargs={"channel": "#ops-firehose"}, - deps=[dep("F_publish_dashboard")], - ), - # Audit archive — runs in parallel with F, independent of failures - _task( - "H_archive_run_metadata", - archive_run_metadata, - args=("s3://audit-lake/",), - kwargs={"retention_days": 90}, - deps=[dep("E_compute_kpis")], - ), - ] - - -# --------------------------------------------------------------------------- # -# 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) - - cleared = 0 - for name in os.listdir(logs_dir): - if name.endswith(".log"): - os.remove(os.path.join(logs_dir, name)) - cleared += 1 - if cleared: - print(f"logs dir: {logs_dir} (cleared {cleared} stale .log file(s))") - else: - print(f"logs dir: {logs_dir} (empty)") - - print(f"recipients: {TO_ADDRS}") - - smtp = _make_smtp_config() - tasks = build_tasks(logs_dir, smtp) - print(f"tasks: {len(tasks)}") - print("-" * 72) - - exit_code = 0 - try: - with Process(tasks) as process: - # Sequential mode: maildev's SMTP listener is single-threaded, - # so concurrent connections from parallel failures race and the - # second one is refused. See the module docstring. - result = process.run(parallel=False) - except Exception: - print("Process raised an unexpected exception:") - traceback.print_exc() - exit_code = 2 - else: - print("-" * 72) - print("passed:") - for name in sorted(result.successes): - print(f" + {name}") - print("failed (includes cascading-skipped):") - for name in sorted(set(result.errored) | set(result.skipped)): - print(f" - {name}") - - if exit_code == 0: - print("-" * 72) - print(f"Inspect the per-task logs in:\n {logs_dir}") - print(f"And the rendered email at:\n http://localhost:{WEB_PORT}") - return exit_code - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/manual_tests/manual_report_notify.py b/tests/manual_tests/manual_report_notify.py index 1eb0340..ca7ffd5 100644 --- a/tests/manual_tests/manual_report_notify.py +++ b/tests/manual_tests/manual_report_notify.py @@ -1,10 +1,11 @@ """Manual inspection: a mixed DAG whose ProcessExecutionReport is delivered by email via SMTP, exercising every traced-variables configuration in one run. -Unlike the other manual scripts (which only send per-task *failure alerts*), -this one is about the **report** path — ``ProcessExecutionReport.notify`` -rendering the whole run as a single HTML email and sending it to the same -maildev server. +Tasks no longer send per-task alerts; notification is delegated entirely to the +**report** path — ``ProcessExecutionReport.notify`` rendering the whole run as a +single HTML email and sending it to maildev. The traced-vars frame filter is now +configured per ``Task`` (capture-time), which feeds both the report and the +per-task logfile. Run by hand and eyeball the result in maildev: @@ -32,20 +33,20 @@ Report delivery (the point of the script): -* ``report.notify(full)`` — full report, ``show_traced_vars= - True`` → the traced-variables sections are - present (rich user locals for - ``fetch_orders``, json internals for - ``decode_payload``, empty for - ``noop_validate``). -* ``report.notify(brief, only_errors=True)`` — errored entries only, - ``show_traced_vars=False`` → the **same** - failures rendered **WITHOUT** the - traced-variables sections. - -So a single run demonstrates, side by side: with/without traced variables (both -per task and via the report content flag) and with/without a custom traceback -frame filter. +The single finished report is delivered once per combination of the full +delivery matrix — **2 languages x 4 palettes x 3 content styles x 2 only_errors +modes = 48 emails** — each sent to a distinct recipient whose local-part encodes +the combination (``report---