core.structlog_utils

Utilities for structured logging with context binding.

Attributes

F

Functions

bind_context(→ Callable[[F], F])	Decorator to automatically bind function/method parameters to structlog context.
bind_method_context(→ Callable[[F], F])	Decorator for binding context in methods with self/cls awareness.

Module Contents

core.structlog_utils.F

core.structlog_utils.bind_context(*param_names: str, **renames: str) → Callable[[F], F]

Decorator to automatically bind function/method parameters to structlog context.

This decorator extracts specified parameters from a function call and binds them to the structlog context for the duration of the function execution. After the function completes (successfully or with an exception), the context is cleaned up.

Parameters:

• *param_names – Names of parameters to bind to context using their original names.

• **renames – Mapping of parameter names to custom context keys (e.g., ti_id=”task_instance_id”).

Returns:

Decorated function with automatic context binding.

Example

>>> @bind_context("workflow_run_id", "cluster_name")
... def process_workflow(workflow_run_id: int, cluster_name: str):
...     logger.info("processing")  # Includes workflow_run_id, cluster

>>> @bind_context(
...     "task_instance_id", wf_id="workflow_run_id"
... )
... def launch_task(task_instance_id: int, workflow_run_id: int):
...     logger.info("launching")

>>> # Works with methods too
... class Service:
...     @bind_context("status", "timeout")
...     def process_status(self, status: str, timeout: int):
...         logger.info("processing_status")  # Includes status and timeout

>>> # Bind nested attributes
... @bind_context(
...     task_id="task_instance.task_instance_id"
... )
... def process_task(task_instance: TaskInstance):
...     logger.info("processing")  # Includes task_id extracted from task_instance

core.structlog_utils.bind_method_context(*param_names: str, **renames: str) → Callable[[F], F]

Decorator for binding context in methods with self/cls awareness.

This is an alias for bind_context that makes it explicit when decorating methods. The decorator automatically handles ‘self’ and ‘cls’ parameters.

Parameters:

• *param_names – Names of parameters to bind (excluding self/cls).

• **renames – Mapping of parameter names to custom context keys.

Returns:

Decorated method with automatic context binding.

Example

>>> class DistributorService:
...     @bind_method_context("task_instance", wf_id="workflow_run.workflow_run_id")
...     def launch_task_instance(self, task_instance: DistributorTaskInstance):
...         logger.info("launching")  # Includes task_instance and wf_id from self