API Reference¶

This reference documents the public API exported by azure_functions_logging.

Use this page together with:

Configuration for setup behavior by environment.
Usage Guide for complete implementation patterns.
Examples for runnable snippets.

setup_logging¶

Configure logging for the current environment. Behavior depends on the detected environment:

Azure / Core Tools (use_record_factory=False, default): Installs ContextFilter on the root logger's existing handlers and on the root logger itself (so records emitted directly on the root logger also carry context fields; note this does not cover records that merely propagate up from named child loggers — those are filtered only by the handler-level filters). Does NOT add new handlers or modify the root logger level (respects host.json configuration). If functions_formatter is provided, it is applied to every root handler before the filter is added. When use_record_factory=True, no ContextFilter is attached; context injection happens via the global LogRecordFactory instead.
Standalone local development: Adds a StreamHandler with ColorFormatter or JsonFormatter to the specified logger (or root logger if logger_name is None). Sets the level. Pass an explicit logger_name to avoid modifying the root logger; when a handler is added to a named logger this way, its propagate flag is set to False so records are not also emitted by an ancestor (e.g. root) handler, which would double-log.

Reconfiguration (calling setup_logging() more than once). Repeat calls are safe and never double-install filters or handlers. The exact contract depends on the detected environment:

Standalone local mode is fully idempotent per logger_name: state is tracked in a process-wide set, so the first call for a given logger_name wins and later calls for that same name return early — level, format, and the added handler are applied only once. Configure a different logger_name to set up an additional logger.
Azure / Core Tools mode is idempotent per (logger_name, use_record_factory) at the handler level: each root handler is given the ContextFilter (and functions_formatter, if supplied) exactly once, tracked by a WeakSet. Repeat calls therefore recover by picking up any handlers the host attached after the previous call, without duplicating filters. The root logger's level and handler list are never modified (host.json owns the level).

Switching use_record_factory between calls. Enabling it (False -> True) installs the global LogRecordFactory and strips any ContextFilter this package previously installed on the target and root loggers, so context is injected by exactly one mechanism. The two modes keep isolated per-signature state, so a later ContextFilter call never reuses a factory-mode instance (or vice versa). Switching back (True -> False) installs a fresh ContextFilter but does not uninstall the already-installed global factory; prefer a single, consistent use_record_factory value per process.

Switching activate_trace_context. A bare setup_logging() (the default activate_trace_context=None) leaves the previously configured activation default untouched; pass True/False to change it.

Parameters:

Name	Type	Description	Default
`level`	`int`	Logging level for local development. Ignored in Azure/Core Tools.	`INFO`
`format`	`str`	Log output format for local development. Supported values are `"color"` (default) and `"json"`. Ignored when `functions_formatter` is provided. In Azure/Core Tools, passing `format="json"` without `functions_formatter` emits a warning.	`'color'`
`logger_name`	`str \| None`	Optional logger name to configure. When None, configures the root logger (local dev) or installs filter on root handlers (Azure).	`None`
`functions_formatter`	`Formatter \| None`	Optional custom formatter applied to all root handlers when running inside Azure/Core Tools. Useful for injecting a custom JSON formatter or third-party formatter without losing ContextFilter integration.	`None`
`host_json_path`	`Path \| str \| None`	Optional explicit path to a `host.json` file used by the host-level conflict warning. When `None` (default), `host.json` is auto-discovered by walking up from the current working directory (bounded). Pass an explicit path to disable auto-discovery in environments where it might pick the wrong file.	`None`
`use_record_factory`	`bool`	When True, install the global `LogRecordFactory` so context fields are injected at LogRecord creation time and are preserved through queued, delayed, or cross-thread handling. When this option is enabled, `ContextFilter` is not attached to handlers, because the global `LogRecordFactory` would be overwritten by the filter at handler dispatch time. Defaults to False to preserve the existing handler-filter-only behavior. See how correlation works, injection modes: https://yeongseon.dev/azure-functions-python/logging/how-correlation-works/#3-two-ways-the-id-lands-on-a-record	`False`
`extra_context_vars`	`dict[str, ContextVar[Any]] \| None`	Optional mapping of `field_name -> ContextVar` whose current values are copied onto each LogRecord, alongside the built-in context fields. Field names must not collide with the built-in fields (raises `ValueError` otherwise). Applies to both injection strategies: the default `ContextFilter` and the global `LogRecordFactory` (`use_record_factory=True`).	`None`
`activate_trace_context`	`bool \| None`	Sets the process-wide default that `logging_context` and `with_context` consult to attach the host's W3C trace context (via OpenTelemetry) — making OTel log records inherit the host span's `trace_id`/`span_id`. Requires the `[otel]` extra; degrades to a silent no-op when OpenTelemetry is not installed. `True`/`False` force the default on/off. The default `None` leaves the stored default unchanged (which itself defaults to `False` — activation is strictly opt-in). A per-call `activate_trace_context` argument overrides this default.	`None`

.. warning::

``use_record_factory=True`` modifies the **global**
``logging.LogRecordFactory``, which affects all loggers in the process
(including third-party libraries). The four context field names
(``invocation_id``, ``function_name``, ``trace_id``, ``cold_start``)
become reserved LogRecord attributes — passing them via ``extra=`` to
stdlib loggers will raise ``KeyError``. Prefer :class:`FunctionLogger`
(which sanitizes ``extra`` keys automatically) when this option is on.

Source code in src/azure_functions_logging/_setup.py

def setup_logging(
    *,
    level: int = logging.INFO,
    format: str = "color",
    logger_name: str | None = None,
    functions_formatter: logging.Formatter | None = None,
    host_json_path: Path | str | None = None,
    use_record_factory: bool = False,
    extra_context_vars: dict[str, contextvars.ContextVar[Any]] | None = None,
    activate_trace_context: bool | None = None,
) -> None:
    """Configure logging for the current environment.
    Behavior depends on the detected environment:

    - **Azure / Core Tools** (``use_record_factory=False``, default): Installs
      ``ContextFilter`` on the root logger's existing handlers **and** on the
      root logger itself (so records emitted directly on the root logger also
      carry context fields; note this does not cover records that merely
      propagate up from named child loggers — those are filtered only by the
      handler-level filters). Does NOT add new handlers or modify the root
      logger level (respects ``host.json`` configuration). If
      ``functions_formatter`` is provided, it is applied to every root handler
      before the filter is added.
      When ``use_record_factory=True``, no ``ContextFilter`` is attached;
      context injection happens via the global ``LogRecordFactory`` instead.
    - **Standalone local development**: Adds a ``StreamHandler`` with
      ``ColorFormatter`` or ``JsonFormatter`` to the specified logger
      (or root logger if ``logger_name`` is None). Sets the level.
      Pass an explicit ``logger_name`` to avoid modifying the root logger;
      when a handler is added to a named logger this way, its
      ``propagate`` flag is set to ``False`` so records are not also emitted
      by an ancestor (e.g. root) handler, which would double-log.

    **Reconfiguration (calling ``setup_logging()`` more than once).**
    Repeat calls are safe and never double-install filters or handlers. The
    exact contract depends on the detected environment:

    - **Standalone local mode** is fully idempotent per ``logger_name``: state
      is tracked in a process-wide set, so the *first* call for a given
      ``logger_name`` wins and later calls for that same name return early —
      ``level``, ``format``, and the added handler are applied only once.
      Configure a different ``logger_name`` to set up an additional logger.
    - **Azure / Core Tools mode** is idempotent per
      ``(logger_name, use_record_factory)`` at the *handler* level: each root
      handler is given the ``ContextFilter`` (and ``functions_formatter``, if
      supplied) exactly once, tracked by a ``WeakSet``. Repeat calls therefore
      *recover* by picking up any handlers the host attached after the previous
      call, without duplicating filters. The root logger's level and handler
      list are never modified (``host.json`` owns the level).

    **Switching ``use_record_factory`` between calls.** Enabling it
    (``False`` -> ``True``) installs the global ``LogRecordFactory`` and strips
    any ``ContextFilter`` this package previously installed on the target and
    root loggers, so context is injected by exactly one mechanism. The two
    modes keep isolated per-signature state, so a later ``ContextFilter`` call
    never reuses a factory-mode instance (or vice versa). Switching back
    (``True`` -> ``False``) installs a fresh ``ContextFilter`` but does **not**
    uninstall the already-installed global factory; prefer a single, consistent
    ``use_record_factory`` value per process.

    **Switching ``activate_trace_context``.** A bare ``setup_logging()`` (the
    default ``activate_trace_context=None``) leaves the previously configured
    activation default untouched; pass ``True``/``False`` to change it.

    Args:
        level: Logging level for local development. Ignored in Azure/Core Tools.
        format: Log output format for local development. Supported values are
            ``"color"`` (default) and ``"json"``. Ignored when
            ``functions_formatter`` is provided. In Azure/Core Tools, passing
            ``format="json"`` without ``functions_formatter`` emits a warning.
        logger_name: Optional logger name to configure. When None, configures
            the root logger (local dev) or installs filter on root handlers (Azure).
        functions_formatter: Optional custom formatter applied to all root
            handlers when running inside Azure/Core Tools. Useful for
            injecting a custom JSON formatter or third-party formatter
            without losing ContextFilter integration.
        host_json_path: Optional explicit path to a ``host.json`` file used by
            the host-level conflict warning. When ``None`` (default),
            ``host.json`` is auto-discovered by walking up from the current
            working directory (bounded). Pass an explicit path to disable
            auto-discovery in environments where it might pick the wrong file.
        use_record_factory: When True, install the global ``LogRecordFactory``
            so context fields are injected at LogRecord creation time and are
            preserved through queued, delayed, or cross-thread handling. When
            this option is enabled, ``ContextFilter`` is **not** attached to
            handlers, because the global ``LogRecordFactory`` would be
            overwritten by the filter at handler dispatch time. Defaults to
            False to preserve the existing handler-filter-only behavior.
            See how correlation works, injection modes:
            https://yeongseon.dev/azure-functions-python/logging/how-correlation-works/#3-two-ways-the-id-lands-on-a-record
        extra_context_vars: Optional mapping of ``field_name -> ContextVar``
            whose current values are copied onto each LogRecord, alongside the
            built-in context fields. Field names must not collide with the
            built-in fields (raises ``ValueError`` otherwise). Applies to both
            injection strategies: the default ``ContextFilter`` and the global
            ``LogRecordFactory`` (``use_record_factory=True``).
        activate_trace_context: Sets the process-wide default that
            ``logging_context`` and ``with_context`` consult to attach the
            host's W3C trace context (via OpenTelemetry) — making OTel log
            records inherit the host span's ``trace_id``/``span_id``. Requires
            the ``[otel]`` extra; degrades to a silent no-op when OpenTelemetry
            is not installed. ``True``/``False`` force the default on/off. The
            default ``None`` leaves the stored default unchanged (which itself
            defaults to ``False`` — activation is strictly opt-in). A per-call
            ``activate_trace_context`` argument overrides this default.

    .. warning::

        ``use_record_factory=True`` modifies the **global**
        ``logging.LogRecordFactory``, which affects all loggers in the process
        (including third-party libraries). The four context field names
        (``invocation_id``, ``function_name``, ``trace_id``, ``cold_start``)
        become reserved LogRecord attributes — passing them via ``extra=`` to
        stdlib loggers will raise ``KeyError``. Prefer :class:`FunctionLogger`
        (which sanitizes ``extra`` keys automatically) when this option is on.
    """
    if format not in {"color", "json"}:
        msg = "format must be 'color' or 'json'"
        raise ValueError(msg)

    # Install the global LogRecordFactory only after argument validation, so an
    # invalid call (e.g. an extra_context_vars collision) does not leave
    # persistent global side effects. In record-factory mode the factory now
    # carries extra_context_vars too, so both modes honor user fields.
    if use_record_factory:
        _install_context_factory(extra_context_vars)

    with _configured_lock:
        # Only override the process-wide activation default when the caller
        # explicitly opts in or out. A bare ``setup_logging()`` (where
        # ``activate_trace_context`` is ``None``) must not silently revert a
        # previously-configured default — that would break idempotency.
        if activate_trace_context is not None:
            set_default_trace_context_activation(activate_trace_context)

        # When switching to record-factory mode, strip any previously-installed
        # ContextFilter from the target logger and root logger.  Must run BEFORE
        # the idempotency guards (so re-entry for the same logger_name still
        # removes stale filters), and UNDER _configured_lock (so concurrent
        # setup_logging() calls cannot race on the filter lists).
        if use_record_factory:
            _target = logging.getLogger(logger_name)
            _root = logging.getLogger()
            _remove_context_filters(_target)
            if _root is not _target:
                _remove_context_filters(_root)

        is_functions_env = _is_functions_environment()

        if is_functions_env:
            # Azure or Core Tools: install filter on handlers, don't touch level.
            #
            # Recovery semantics: if the host attaches new handlers after the
            # first call, subsequent calls will pick them up. We track which
            # handler ids have already been configured so we don't add duplicate
            # filters/formatters, and reuse the same ContextFilter instance so
            # that root.addFilter() is idempotent (identity-based check).
            if format != "color" and functions_formatter is None:
                warnings.warn(
                    "The 'format' parameter is ignored in Azure Functions environment. "
                    "Pass functions_formatter=JsonFormatter() to set JSON output on host handlers.",
                    stacklevel=2,
                )

            # Retrieve or create the per-call-signature state.
            _state_key: _AzureStateKey = (logger_name, use_record_factory)
            if _state_key not in _azure_state:
                ctx_filter: ContextFilter | None = (
                    None if use_record_factory else ContextFilter(extra_context_vars)
                )
                _azure_state[_state_key] = (ctx_filter, weakref.WeakSet())
            context_filter, configured_handlers = _azure_state[_state_key]
            root = logging.getLogger()
            for handler in root.handlers:
                if handler in configured_handlers:
                    continue  # already configured — skip to avoid duplicates
                if functions_formatter is not None:
                    handler.setFormatter(functions_formatter)
                if context_filter is not None:
                    handler.addFilter(context_filter)
                configured_handlers.add(handler)
            # Install the filter on the root logger itself so records emitted
            # directly on the root logger also carry context. This does NOT
            # apply to records propagating up from named child loggers, which
            # are only seen by the handler-level filters added above.
            if context_filter is not None and context_filter not in root.filters:
                root.addFilter(context_filter)

            warn_host_json_level_conflict(level, host_json_path=host_json_path)
            warn_otel_logging_misconfig(
                functions_formatter=functions_formatter,
                host_json_path=host_json_path,
            )

        else:
            # Standalone local development: full idempotency via logger name.
            if logger_name in _configured_loggers:
                return

            # When the LogRecordFactory is active, attaching ContextFilter would
            # overwrite factory-injected fields at handler dispatch time.
            context_filter = None if use_record_factory else ContextFilter(extra_context_vars)
            target = logging.getLogger(logger_name)
            target.setLevel(level)

            # Add colored handler only if no handlers exist
            if not target.handlers:
                handler = logging.StreamHandler()
                handler.setFormatter(ColorFormatter() if format == "color" else JsonFormatter())
                if context_filter is not None:
                    handler.addFilter(context_filter)
                target.addHandler(handler)
                if logger_name is not None:
                    # This named logger now owns its output via the handler we
                    # just added. Stop propagation to ancestor loggers so a
                    # handler already on the root logger (e.g. from
                    # ``basicConfig``, pytest, or a framework) does not emit
                    # every record a SECOND time. The root logger
                    # (``logger_name=None``) has no ancestor to double-emit to,
                    # so its propagation is left untouched.
                    target.propagate = False
            elif context_filter is not None:
                # Add filter to existing handlers
                for handler in target.handlers:
                    handler.addFilter(context_filter)

            _configured_loggers.add(logger_name)

Usage Notes¶

Call once during startup.
Default format is "color".
In Azure/Core Tools runtime, filter-only behavior avoids duplicate handlers.

Example¶

import logging
from azure_functions_logging import setup_logging

setup_logging(level=logging.INFO, format="json")

Example: Named Target Logger¶

from azure_functions_logging import setup_logging

setup_logging(logger_name="my_service")

Example: Invalid Format Handling¶

from azure_functions_logging import setup_logging

try:
    setup_logging(format="pretty")
except ValueError:
    pass

get_logger¶

Create a FunctionLogger wrapping a standard logging.Logger.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Logger name. Typically `__name__`.	`None`

Returns:

Type	Description
`FunctionLogger`	A `FunctionLogger` instance.

Source code in src/azure_functions_logging/__init__.py

def get_logger(name: str | None = None) -> FunctionLogger:
    """Create a ``FunctionLogger`` wrapping a standard ``logging.Logger``.

    Args:
        name: Logger name. Typically ``__name__``.

    Returns:
        A ``FunctionLogger`` instance.
    """
    import logging

    return FunctionLogger(logging.getLogger(name))

Usage Notes¶

Returns a FunctionLogger wrapper over a standard logger.
Pass __name__ for module-level identity.
Use the wrapper methods like standard logging methods.

Example¶

from azure_functions_logging import get_logger, setup_logging

setup_logging()
logger = get_logger(__name__)
logger.info("module logger ready")

Example: Root Logger Wrapper¶

from azure_functions_logging import get_logger, setup_logging

setup_logging()
root_logger = get_logger()
root_logger.warning("root logger event")

FunctionLogger¶

Wrapper around a standard logging.Logger with context binding.

FunctionLogger forwards the common logging methods (debug, info, warning, error, critical, exception, log) through :meth:_log, which merges bound context and sanitizes reserved extra keys. Any other attribute of the underlying logging.Logger (addHandler, handlers, propagate, level, getChild, ...) is delegated via :meth:__getattr__, so a FunctionLogger behaves like the stdlib Logger it wraps.

The bind() method returns a new wrapper with additional context fields that are merged into extra on each log call.

Context from bind() is supplementary to the ContextFilter-based context (invocation_id, function_name, etc.) which is set globally via inject_context().

Source code in src/azure_functions_logging/_logger.py

def __init__(self, logger: logging.Logger) -> None:
    self._logger = logger
    self._context: dict[str, Any] = {}

`name` `property` ¶

Return the name of the underlying logger.

`getattr(name)` ¶

Delegate unknown attributes to the wrapped logging.Logger.

__getattr__ only runs when normal lookup fails, so it never shadows FunctionLogger's own slots or methods. It forwards the rest of the stdlib Logger surface (addHandler, removeHandler, addFilter, handlers, propagate, level, disabled, parent, manager, getChild, ...) to self._logger.

Source code in src/azure_functions_logging/_logger.py

def __getattr__(self, name: str) -> Any:
    """Delegate unknown attributes to the wrapped ``logging.Logger``.

    ``__getattr__`` only runs when normal lookup fails, so it never
    shadows ``FunctionLogger``'s own slots or methods. It forwards the rest
    of the stdlib ``Logger`` surface (``addHandler``, ``removeHandler``,
    ``addFilter``, ``handlers``, ``propagate``, ``level``, ``disabled``,
    ``parent``, ``manager``, ``getChild``, ...) to ``self._logger``.
    """
    # Guard the slot names so a missing ``_logger`` (e.g. before __init__)
    # raises AttributeError instead of recursing infinitely.
    if name in ("_logger", "_context"):
        raise AttributeError(name)
    return getattr(self._logger, name)

`bind(**kwargs)` ¶

Return a new FunctionLogger with additional bound context.

The returned logger shares the same underlying logging.Logger but carries merged context fields. This is an immutable operation.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Context key-value pairs to bind.	`{}`

Returns:

Type	Description
`FunctionLogger`	A new `FunctionLogger` with merged context.

Source code in src/azure_functions_logging/_logger.py

def bind(self, **kwargs: Any) -> FunctionLogger:
    """Return a new ``FunctionLogger`` with additional bound context.

    The returned logger shares the same underlying ``logging.Logger``
    but carries merged context fields. This is an immutable operation.

    Args:
        **kwargs: Context key-value pairs to bind.

    Returns:
        A new ``FunctionLogger`` with merged context.
    """
    new = FunctionLogger(self._logger)
    new._context = {**self._context, **kwargs}
    return new

`clear_context()` ¶

Clear all bound context fields.

Source code in src/azure_functions_logging/_logger.py

def clear_context(self) -> None:
    """Clear all bound context fields."""
    self._context = {}

`critical(msg, *args, **kwargs)` ¶

Log a CRITICAL message.

Source code in src/azure_functions_logging/_logger.py

def critical(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Log a CRITICAL message."""
    self._log(logging.CRITICAL, msg, args, **kwargs)

`debug(msg, *args, **kwargs)` ¶

Log a DEBUG message.

Source code in src/azure_functions_logging/_logger.py

def debug(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Log a DEBUG message."""
    self._log(logging.DEBUG, msg, args, **kwargs)

`error(msg, *args, **kwargs)` ¶

Log an ERROR message.

Source code in src/azure_functions_logging/_logger.py

def error(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Log an ERROR message."""
    self._log(logging.ERROR, msg, args, **kwargs)

`exception(msg, *args, **kwargs)` ¶

Log an ERROR message with exception info.

Source code in src/azure_functions_logging/_logger.py

def exception(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Log an ERROR message with exception info."""
    kwargs["exc_info"] = kwargs.get("exc_info", True)
    self._log(logging.ERROR, msg, args, **kwargs)

`fatal(msg, *args, **kwargs)` ¶

Alias for :meth:critical (mirrors logging.Logger.fatal).

Implemented explicitly rather than delegated so bound context and reserved-key sanitization still apply.

Source code in src/azure_functions_logging/_logger.py

def fatal(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Alias for :meth:`critical` (mirrors ``logging.Logger.fatal``).

    Implemented explicitly rather than delegated so bound context and
    reserved-key sanitization still apply.
    """
    self._log(logging.CRITICAL, msg, args, **kwargs)

`getEffectiveLevel()` ¶

Return the effective level of the underlying logger.

Source code in src/azure_functions_logging/_logger.py

def getEffectiveLevel(self) -> int:
    """Return the effective level of the underlying logger."""
    return self._logger.getEffectiveLevel()

`hasHandlers()` ¶

Return whether the underlying logger has any handlers configured.

Source code in src/azure_functions_logging/_logger.py

def hasHandlers(self) -> bool:
    """Return whether the underlying logger has any handlers configured."""
    return self._logger.hasHandlers()

`info(msg, *args, **kwargs)` ¶

Log an INFO message.

Source code in src/azure_functions_logging/_logger.py

def info(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Log an INFO message."""
    self._log(logging.INFO, msg, args, **kwargs)

`isEnabledFor(level)` ¶

Check if the underlying logger is enabled for the given level.

Source code in src/azure_functions_logging/_logger.py

def isEnabledFor(self, level: int) -> bool:
    """Check if the underlying logger is enabled for the given level."""
    return self._logger.isEnabledFor(level)

`log(level, msg, *args, **kwargs)` ¶

Log msg at the given level, mirroring logging.Logger.log.

Honors the same bind < extra < kwargs merge precedence as :meth:info / :meth:warning / etc. and applies the same reserved-key sanitization.

Source code in src/azure_functions_logging/_logger.py

def log(self, level: int, msg: object, *args: Any, **kwargs: Any) -> None:
    """Log ``msg`` at the given ``level``, mirroring ``logging.Logger.log``.

    Honors the same ``bind`` < ``extra`` < ``kwargs`` merge precedence
    as :meth:`info` / :meth:`warning` / etc. and applies the same
    reserved-key sanitization.
    """
    self._log(level, msg, args, **kwargs)

`setLevel(level)` ¶

Set the logging level of the underlying logger.

Source code in src/azure_functions_logging/_logger.py

def setLevel(self, level: int | str) -> None:
    """Set the logging level of the underlying logger."""
    self._logger.setLevel(level)

`warn(msg, *args, **kwargs)` ¶

Deprecated alias for :meth:warning (mirrors logging.Logger.warn).

Implemented explicitly rather than delegated so bound context and reserved-key sanitization still apply.

Source code in src/azure_functions_logging/_logger.py

def warn(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Deprecated alias for :meth:`warning` (mirrors ``logging.Logger.warn``).

    Implemented explicitly rather than delegated so bound context and
    reserved-key sanitization still apply.
    """
    self._log(logging.WARNING, msg, args, **kwargs)

`warning(msg, *args, **kwargs)` ¶

Log a WARNING message.

Source code in src/azure_functions_logging/_logger.py

def warning(self, msg: object, *args: Any, **kwargs: Any) -> None:
    """Log a WARNING message."""
    self._log(logging.WARNING, msg, args, **kwargs)

Usage Notes¶

bind() returns a new immutable logger wrapper with merged context.
clear_context() clears bound context on that wrapper instance.
Logging methods mirror standard logger API.

Example: Binding Context¶

from azure_functions_logging import get_logger, setup_logging

setup_logging(format="json")
logger = get_logger("checkout")

request_logger = logger.bind(request_id="r-100", user_id="u-55")
request_logger.info("checkout started")

Example: Chained Binding¶

base = get_logger("service")
l1 = base.bind(tenant_id="tenant-a")
l2 = l1.bind(operation="import")
l2.info("import queued")

Example: Clearing Bound Context¶

log = get_logger("demo").bind(session="s-1")
log.info("before clear")
log.clear_context()
log.info("after clear")

Example: Exception Logging¶

log = get_logger("errors")

try:
    raise RuntimeError("boom")
except RuntimeError:
    log.exception("operation failed", phase="load")

JsonFormatter¶

Bases: Formatter

Structured JSON log formatter.

Output is newline-delimited JSON (NDJSON), with one JSON object per log line. Context fields (invocation_id, function_name, etc.) are included when present on the LogRecord (set by ContextFilter).

Unserializable values in extra are coerced to strings via :func:_json_default rather than dropping the log record.

Parameters:

Name	Type	Description	Default
`max_string_length`	`int`	Maximum character length for each native string value in `extra` when `truncate_native_strings=True`. Default: 2048.	`2048`
`truncate_native_strings`	`bool`	When `True`, recursively truncate string values inside `extra` (dicts and lists walked) to `max_string_length` characters. Truncated strings are suffixed with `…` so callers can detect the truncation. Non-string scalar values (int, float, bool) are not affected. Default: `False`.	`False`

Source code in src/azure_functions_logging/_json_formatter.py

def __init__(
    self,
    *,
    max_string_length: int = 2048,
    truncate_native_strings: bool = False,
) -> None:
    if max_string_length < 0:
        raise ValueError(f"max_string_length must be ≥ 0, got {max_string_length}")
    super().__init__()
    self._max_string_length = max_string_length
    self._truncate_native_strings = truncate_native_strings

`format(record)` ¶

Format a log record as one NDJSON object.

Fully fail-safe: every step (message rendering, exception formatting, timestamp, JSON serialization) is wrapped so that a hostile __str__ / __repr__, a malformed exc_info triple, or a cyclic extra payload never raises out of format(). Worst-case we still emit a single valid JSON object with sentinel values.

Source code in src/azure_functions_logging/_json_formatter.py

def format(self, record: logging.LogRecord) -> str:
    """Format a log record as one NDJSON object.

    Fully fail-safe: every step (message rendering, exception formatting,
    timestamp, JSON serialization) is wrapped so that a hostile
    ``__str__`` / ``__repr__``, a malformed ``exc_info`` triple, or a
    cyclic ``extra`` payload never raises out of ``format()``. Worst-case
    we still emit a single valid JSON object with sentinel values.
    """
    message = _safe_get_message(record)
    timestamp = _safe_timestamp(record)

    exception: str | None = None
    if record.exc_info:
        exception = _safe_format_exception(self, record.exc_info)

    excluded_fields = _STANDARD_RECORD_FIELDS | _CONTEXT_FIELDS
    extra = {key: value for key, value in record.__dict__.items() if key not in excluded_fields}
    extra = _to_json_safe(extra)
    if self._truncate_native_strings:
        extra = _truncate_native_strings(extra, self._max_string_length)

    payload = {
        "timestamp": timestamp,
        "level": record.levelname,
        "logger": record.name,
        "message": message,
        "invocation_id": getattr(record, "invocation_id", None),
        "function_name": getattr(record, "function_name", None),
        "trace_id": getattr(record, "trace_id", None),
        "span_id": getattr(record, "span_id", None),
        "cold_start": getattr(record, "cold_start", None),
        "host_instance_id": getattr(record, "host_instance_id", None) or get_host_instance_id(),
        "exception": exception,
        "extra": extra,
    }

    try:
        return json.dumps(payload, ensure_ascii=False, default=_json_default)
    except Exception:
        # Last-resort fallback: drop ``extra`` (the most likely culprit
        # for cyclic / unserializable payloads) and re-attempt. If even
        # that fails, emit a minimal hand-built JSON object so the
        # logging pipeline never breaks.
        payload["extra"] = {"__serialization_error__": True}
        try:
            return json.dumps(payload, ensure_ascii=False, default=_json_default)
        except Exception:
            return _emergency_payload(record)

Usage Notes¶

Use indirectly via setup_logging(format="json") for most cases.
Produces one JSON object per line (NDJSON style).
Includes context fields when available on log records.
Pass truncate_native_strings=True to clip string values in extra at max_string_length characters (default 2048); truncated values are suffixed with …. Only strings in extra are affected (recursively through dicts/lists) — message, ints, floats, and booleans are left intact.

Example: Automatic Selection¶

from azure_functions_logging import get_logger, setup_logging

setup_logging(format="json")
logger = get_logger("api")
logger.info("json formatter active", version="v1")

Example: Manual Formatter Wiring¶

import logging
from azure_functions_logging import JsonFormatter, get_logger

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())

target = logging.getLogger("manual")
target.handlers = [handler]
target.setLevel(logging.INFO)

logger = get_logger("manual")
logger.info("manual formatter configured")

SamplingFilter¶

Bases: Filter

Rate-limit a logger to emit at most rate records per window seconds.

Useful for high-frequency loggers (e.g. per-request HTTP logs, polling loops) that can saturate the Azure Functions gRPC channel.

All records that exceed the rate cap are silently dropped. Records at WARNING and above are always passed through, regardless of the cap.

Parameters:

Name	Type	Description	Default
`rate`	`int`	Maximum number of records to pass per window. Must be >= 1.	`100`
`window`	`float`	Rolling time window in seconds. Default: 1.0.	`1.0`
`name`	`str`	Optional logger-name scope. When set, only matching loggers are subject to sampling; non-matching records pass through unchanged. Empty string matches all loggers (default).	`''`
`per_logger`	`bool`	When False (default), all matching records share one rate bucket per filter instance. When True, each `record.name` has an independent bucket/window. Best suited for finite logger-name cardinality (e.g. module-based loggers). Stale buckets are automatically evicted to prevent unbounded memory growth.	`False`

Example::

filter = SamplingFilter(rate=10, window=1.0)
handler.addFilter(filter)

Source code in src/azure_functions_logging/_filters.py

def __init__(
    self,
    rate: int = 100,
    window: float = 1.0,
    name: str = "",
    *,
    per_logger: bool = False,
) -> None:
    super().__init__(name)
    if rate < 1:
        msg = "rate must be >= 1"
        raise ValueError(msg)
    if window <= 0:
        msg = "window must be > 0"
        raise ValueError(msg)
    self._rate: int = rate
    self._window: float = window
    self._lock: threading.Lock = threading.Lock()
    self._per_logger: bool = per_logger
    self._count: int = 0
    self._window_start: float = time.monotonic()
    # per_logger state: {record.name: (window_start, count)}
    self._buckets: dict[str, tuple[float, int]] = {}
    self._last_eviction: float = 0.0  # monotonic timestamp of last eviction

`filter(record)` ¶

Return True to emit the record, False to drop it.

Source code in src/azure_functions_logging/_filters.py

def filter(self, record: logging.LogRecord) -> bool:
    """Return True to emit the record, False to drop it."""
    # Honor name-based scoping from logging.Filter
    if not super().filter(record):
        return True  # bypass sampling for non-matching loggers

    # Always pass WARNING and above
    if record.levelno >= logging.WARNING:
        return True

    now = time.monotonic()
    with self._lock:
        if self._per_logger:
            bucket = self._buckets.get(record.name)
            if bucket is None or now - bucket[0] >= self._window:
                self._buckets[record.name] = (now, 1)
                # Opportunistic eviction when over capacity
                if len(self._buckets) > self._MAX_BUCKETS:
                    if now - self._last_eviction >= self._window:
                        # Full stale sweep (throttled to once per window)
                        self._evict_stale_buckets(now)
                        self._last_eviction = now
                    elif len(self._buckets) > self._MAX_BUCKETS:
                        # Throttle blocked full sweep; enforce hard cap
                        # by dropping the single oldest bucket
                        oldest_name = min(self._buckets, key=lambda k: self._buckets[k][0])
                        del self._buckets[oldest_name]
                return True
            count = bucket[1] + 1
            self._buckets[record.name] = (bucket[0], count)
            return count <= self._rate

        if now - self._window_start >= self._window:
            self._count = 0
            self._window_start = now
        self._count += 1
        return self._count <= self._rate

Usage Notes¶

WARNING and above always pass.
name= scopes sampling to matching logger names; non-matching records bypass sampling.
per_logger=False shares one bucket across all matching records on the filter instance.
per_logger=True gives each record.name an independent bucket/window.

Example: Per-Logger Buckets for Azure SDK Logs¶

import logging
from azure_functions_logging import SamplingFilter

for handler in logging.getLogger().handlers:
    handler.addFilter(SamplingFilter(rate=10, window=1.0, name="azure", per_logger=True))

RedactionFilter¶

Bases: Filter

Mask PII / sensitive values on LogRecord extra attributes in-place.

Iterates over all non-standard attributes on the LogRecord and replaces the value of any key whose normalized name is in sensitive_keys with "***".

Key normalization: lowercased, hyphens replaced with underscores. This means X-Functions-Key matches the entry x_functions_key.

This filter mutates the record in-place so both ColorFormatter and JsonFormatter see redacted values.

Parameters:

Name	Type	Description	Default
`sensitive_keys`	`Iterable[str] \| None`	Iterable of key names to redact (case-insensitive, hyphen-insensitive). When None, uses the built-in default set (25 keys): `password`, `passwd`, `pwd`, `token`, `access_token`, `refresh_token`, `id_token`, `authorization`, `auth`, `secret`, `client_secret`, `secret_key`, `api_key`, `apikey`, `subscription_key`, `connection_string`, `conn_str`, `sas_token`, `x_functions_key`, `function_key`, `master_key`, `private_key`, `credential`, `account_key`, `access_key`.	`None`
`name`	`str`	Optional logger-name scope. When set, only matching loggers are subject to redaction; non-matching records pass through unchanged.	`''`
`patterns`	`Iterable[str \| Pattern[str]] \| None`	Optional iterable of secret patterns (compiled `re.Pattern` or regex strings) enabling value-based redaction of the rendered message and string `extra` values — catching secrets embedded in free text that key-based redaction cannot see. Off by default (`None`) to avoid false positives and hot-path scanning cost; pass :data:`~azure_functions_logging._redaction.DEFAULT_PATTERNS` for a curated high-confidence set. A pattern may define a `keep` named group to preserve a readable prefix (e.g. `token=`) while masking only the value. Regex strings are compiled at construction; an invalid pattern raises `re.error` there (fail-fast on config), and runtime substitution failures are swallowed (Principle 3).	`None`

Note

The default set includes credential which may over-redact in codebases that use generic attribute names. Pass an explicit sensitive_keys set if false positives occur.

Example::

filter = RedactionFilter()
handler.addFilter(filter)

Source code in src/azure_functions_logging/_filters.py

def __init__(
    self,
    sensitive_keys: Iterable[str] | None = None,
    name: str = "",
    patterns: Iterable[str | re.Pattern[str]] | None = None,
) -> None:
    super().__init__(name)
    self._sensitive_keys: frozenset[str] = (
        frozenset(_normalize_key(k) for k in sensitive_keys)
        if sensitive_keys is not None
        else _DEFAULT_SENSITIVE_KEYS
    )
    self._patterns: tuple[re.Pattern[str], ...] = (
        tuple(p if isinstance(p, re.Pattern) else re.compile(p) for p in patterns)
        if patterns is not None
        else ()
    )

`filter(record)` ¶

Redact sensitive fields on the record. Always returns True.

Source code in src/azure_functions_logging/_filters.py

def filter(self, record: logging.LogRecord) -> bool:
    """Redact sensitive fields on the record. Always returns True."""
    # Honor name-based scoping from logging.Filter
    if not super().filter(record):
        return True  # bypass redaction for non-matching loggers

    try:
        for key in list(record.__dict__.keys()):
            try:
                if key in _RESERVED_LOG_RECORD_KEYS:
                    continue
                if _is_sensitive(key, self._sensitive_keys):
                    setattr(record, key, _MASK)
                    continue
                value = record.__dict__[key]
                if isinstance(value, (dict, list)):
                    setattr(record, key, _redact_value(value, self._sensitive_keys))
                elif self._patterns and isinstance(value, str):
                    setattr(record, key, _mask_patterns(value, self._patterns))
            except Exception:  # nosec B110 — one broken field must not stop others
                pass
        if self._patterns:
            self._mask_message(record)
    except Exception:  # nosec B110 — filter must never raise
        pass
    return True

Value / pattern-based redaction¶

By default RedactionFilter is key-based: it masks values whose key is sensitive, but not secrets embedded in a free-text message or string extra (e.g. logger.info("connecting with token=ghp_...")). Pass patterns= to enable opt-in value-based redaction of the rendered message and string extras:

from azure_functions_logging import DEFAULT_REDACTION_PATTERNS, RedactionFilter

flt = RedactionFilter(patterns=DEFAULT_REDACTION_PATTERNS)

It is off by default because pattern scanning has a hot-path cost and can produce false positives. DEFAULT_REDACTION_PATTERNS is a curated set of high-confidence secrets (bearer tokens, key=value secrets, Azure connection-string keys / SAS sig=, AWS access-key IDs, GitHub tokens, JWTs); patterns with a keep named group preserve a readable prefix (e.g. token=) while masking only the value. You may also pass your own regex strings or compiled patterns. Substitution failures never raise (see the design principle that redaction failures are silent).

DEFAULT_REDACTION_PATTERNS¶

AttributeFlattenFilter¶

Bases: Filter

Flatten nested dict extras into dotted scalar attributes in-place.

OpenTelemetry attributes only permit scalars and homogeneous arrays. A nested dict passed via extra (e.g. order={"id": 1}) is silently dropped by the OTel SDK. This filter rewrites such attributes into dotted scalar keys (order.id) so the data survives export.

The filter mutates the record in-place, removing the original nested-dict attribute and adding one attribute per leaf. It is opt-in: it has no effect unless explicitly attached to a handler/logger.

Behavior: - Nested dicts are flattened recursively to dotted keys. - Lists / heterogeneous arrays are left unchanged (emitted as-is under their dotted key). OTel accepts homogeneous scalar arrays; heterogeneous or nested-object arrays remain the caller's responsibility. - Scalar attributes and reserved LogRecord keys are never touched. - Empty dicts contribute no keys (the attribute is removed). - Cyclic references are dropped; over-deep structures (beyond max_depth) are emitted as-is at the depth boundary.

.. note::

This filter rewrites the record's attributes, so it also affects
**non-OTel** consumers reading the same record — e.g. a
:class:`JsonFormatter` on the root handler will emit ``order.id``
instead of a nested ``order`` object. To avoid changing your JSON log
shape, attach this filter only to the OpenTelemetry ``LoggingHandler``
rather than to the root handler.

Parameters:

Name	Type	Description	Default
`name`	`str`	Optional logger-name scope. When set, only matching loggers are flattened; non-matching records pass through unchanged.	`''`
`separator`	`str`	Delimiter joining nested keys. Default `"."`.	`'.'`
`max_depth`	`int`	Maximum recursion depth before a nested dict is emitted as-is. Default `10`.	`_FLATTEN_MAX_DEPTH`

Example::

filter = AttributeFlattenFilter()
handler.addFilter(filter)

Source code in src/azure_functions_logging/_filters.py

def __init__(
    self,
    name: str = "",
    *,
    separator: str = ".",
    max_depth: int = _FLATTEN_MAX_DEPTH,
) -> None:
    super().__init__(name)
    self._separator: str = separator
    self._max_depth: int = max_depth

`filter(record)` ¶

Flatten nested-dict fields on the record. Always returns True.

Source code in src/azure_functions_logging/_filters.py

def filter(self, record: logging.LogRecord) -> bool:
    """Flatten nested-dict fields on the record. Always returns True."""
    # Honor name-based scoping from logging.Filter
    if not super().filter(record):
        return True  # bypass flattening for non-matching loggers

    try:
        for key in list(record.__dict__.keys()):
            try:
                if key in _RESERVED_LOG_RECORD_KEYS:
                    continue
                value = record.__dict__[key]
                if not isinstance(value, dict):
                    continue
                flattened = _flatten_dict(key, value, self._separator, self._max_depth)
                del record.__dict__[key]
                for flat_key, flat_value in flattened.items():
                    setattr(record, flat_key, flat_value)
            except Exception:  # nosec B110 — one broken field must not stop others
                pass
    except Exception:  # nosec B110 — filter must never raise
        pass
    return True

Usage Notes¶

Opt-in only. Attach it to a handler/logger to flatten nested dict extras into dotted scalar keys (e.g. order={"id": 1} becomes order.id=1).
Intended for OpenTelemetry pipelines, where nested dict attributes are silently dropped by the OTel SDK.
Lists / heterogeneous arrays are left unchanged (emitted as-is under their dotted key). Flattening does not recurse into lists, so a list-of-dicts such as items=[{"id": 1}] is passed through verbatim rather than expanded into items.0.id.
Non-string dict keys are skipped (they cannot form a queryable dotted path).
On key collision — e.g. a nested {"a": {"b": 1}} and a literal {"a.b": 2} both mapping to a.b — the first value in iteration order wins and later collisions are dropped silently.

Attach to specific handlers, not the root logger blindly

This filter mutates the LogRecord in place, rewriting nested-dict attributes into new dotted-key attributes. Attach it only to the handlers that need flattened output (e.g. your OpenTelemetry handler). Installing it on the root logger changes the record schema for every downstream handler, which can surprise formatters that expect the original nested attribute.

from azure_functions_logging import AttributeFlattenFilter

handler.addFilter(AttributeFlattenFilter())

inject_context¶

Set invocation context from an Azure Functions context object.

Extracts invocation_id, function_name, trace_id, and cold_start from the provided context and stores them in contextvars.

This function is safe to call with any object. Missing or inaccessible attributes are silently ignored (Principle 3: context injection failures never cause application failures).

Parameters:

Name	Type	Description	Default
`context`	`Any`	An Azure Functions context object (func.Context).	required

Returns:

Type	Description
`ContextTokens`	A mapping of ContextVar to Token that can be passed to
`ContextTokens`	`restore_context()` to restore the previous state.

Source code in src/azure_functions_logging/_context.py

def inject_context(context: Any) -> ContextTokens:
    """Set invocation context from an Azure Functions context object.

    Extracts invocation_id, function_name, trace_id, and cold_start
    from the provided context and stores them in contextvars.

    This function is safe to call with any object. Missing or inaccessible
    attributes are silently ignored (Principle 3: context injection failures
    never cause application failures).

    Args:
        context: An Azure Functions context object (func.Context).

    Returns:
        A mapping of ContextVar to Token that can be passed to
        ``restore_context()`` to restore the previous state.
    """
    tokens: ContextTokens = {}
    try:
        tokens[invocation_id_var] = invocation_id_var.set(getattr(context, "invocation_id", None))
    except Exception:  # nosec B110 — Principle 3: context failures are silent
        tokens[invocation_id_var] = invocation_id_var.set(None)

    try:
        tokens[function_name_var] = function_name_var.set(getattr(context, "function_name", None))
    except Exception:  # nosec B110 — Principle 3: context failures are silent
        tokens[function_name_var] = function_name_var.set(None)

    try:
        trace_context = getattr(context, "trace_context", None)
        trace_parent = getattr(trace_context, "trace_parent", None) if trace_context else None
        parts = _extract_trace_context(trace_parent)
        tokens[trace_id_var] = trace_id_var.set(parts.trace_id if parts is not None else None)
        tokens[span_id_var] = span_id_var.set(parts.span_id if parts is not None else None)
    except Exception:  # nosec B110 — Principle 3: context failures are silent
        tokens[trace_id_var] = trace_id_var.set(None)
        tokens[span_id_var] = span_id_var.set(None)

    try:
        tokens[cold_start_var] = cold_start_var.set(_check_cold_start())
    except Exception:  # nosec B110 — Principle 3: context failures are silent
        pass
    return tokens

Usage Notes¶

Call at the start of every function invocation.
Sets invocation metadata in context variables.
Enables automatic cold start field in output.

Example: Azure Function Entrypoint¶

import azure.functions as func
from azure_functions_logging import get_logger, inject_context, setup_logging

setup_logging(format="json")
logger = get_logger(__name__)

app = func.FunctionApp()


@app.route(route="status")
def status(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    inject_context(context)
    logger.info("status request")
    return func.HttpResponse("ok")

Example: Safe with Partial Context Object¶

from azure_functions_logging import get_logger, inject_context, setup_logging

class PartialContext:
    invocation_id = "local-123"


setup_logging(format="json")
logger = get_logger("partial")

inject_context(PartialContext())
logger.info("partial context accepted")

with_context¶

Decorator that automatically injects invocation context.

Can be used with or without arguments::

@with_context
def handler(req, context):
    ...

@with_context(param="ctx")
def handler(req, ctx):
    ...

The decorator:

Finds the context parameter (by name, default "context")
Calls inject_context(context) before the handler body
Restores the previous context in finally after the handler returns

Both sync and async handlers are supported.

get_logging_metadata¶

Return logging metadata if the function was decorated with with_context.

Returns None if the function has no logging metadata attached.

Source code in src/azure_functions_logging/_decorator.py

def get_logging_metadata(func: Any) -> dict[str, Any] | None:
    """Return logging metadata if the function was decorated with ``with_context``.

    Returns ``None`` if the function has no logging metadata attached.
    """
    meta = read_logging_metadata(func)
    return dict(meta) if meta is not None else None

Usage Notes¶

Returns the logging metadata dict attached by the with_context decorator, or None when the function was not decorated.
Signature: get_logging_metadata(func: Any) -> dict[str, Any] | None.

Example¶

from azure_functions_logging import get_logging_metadata, with_context


@with_context
def handler(req, context=None):
    ...


metadata = get_logging_metadata(handler)
# -> {"version": 1, "context_param": "context"} or None if not decorated

logging_context¶

Context manager wrapping inject_context + restore_context.

Recommended pattern when handlers don't use the with_context decorator::

def handler(req, context):
    with logging_context(context):
        logger.info("processing")
        ...

Guarantees context is restored to its previous state even if the body raises, supporting safe nesting of contexts.

Parameters:

Name	Type	Description	Default
`context`	`Any`	An Azure Functions context object (func.Context).	required
`activate_trace_context`	`bool \| None`	When `True`, also attach the host's W3C trace context (from `context.trace_context`) via OpenTelemetry so log records emitted through an OTel `LoggingHandler` inherit the host span's `trace_id`/`span_id`. Requires the `[otel]` extra; `None` (default), the process-wide default configured via `setup_logging(activate_trace_context=...)` is used, which itself defaults to `False` (activation is strictly opt-in).	`None`

Source code in src/azure_functions_logging/_context.py

@contextmanager
def logging_context(context: Any, *, activate_trace_context: bool | None = None) -> Iterator[None]:
    """Context manager wrapping ``inject_context`` + ``restore_context``.

    Recommended pattern when handlers don't use the ``with_context`` decorator::

        def handler(req, context):
            with logging_context(context):
                logger.info("processing")
                ...

    Guarantees context is restored to its previous state even if the body raises,
    supporting safe nesting of contexts.

    Args:
        context: An Azure Functions context object (func.Context).
        activate_trace_context: When ``True``, also attach the host's W3C trace
            context (from ``context.trace_context``) via OpenTelemetry so log
            records emitted through an OTel ``LoggingHandler`` inherit the host
            span's ``trace_id``/``span_id``. Requires the ``[otel]`` extra;
            ``None`` (default), the process-wide default configured via
            ``setup_logging(activate_trace_context=...)`` is used, which itself
            defaults to ``False`` (activation is strictly opt-in).
    """
    should_activate = (
        get_default_trace_context_activation()
        if activate_trace_context is None
        else activate_trace_context
    )
    tokens = inject_context(context)
    try:
        if should_activate:
            trace_parent, trace_state = _read_trace_headers(context)
            from ._otel import activated_trace_context

            with activated_trace_context(trace_parent, trace_state):
                yield
        else:
            yield
    finally:
        restore_context(tokens)

propagate_context¶

Bind the current invocation context to func for background-thread execution.

Invocation context is stored in :mod:contextvars, which do not propagate to :class:~concurrent.futures.ThreadPoolExecutor workers or manually created :class:threading.Thread targets. Wrapping a callable with propagate_context snapshots the current invocation context fields (invocation_id, function_name, trace_id, span_id, cold_start) at wrap time and re-applies them inside the wrapper when it later runs on another thread, then restores the previous values on exit so pooled threads never leak context between tasks.

Wrap the callable inside the invocation whose context should be propagated, immediately before handing work to a thread or executor::

from concurrent.futures import ThreadPoolExecutor

def handler(req, context):
    with logging_context(context):
        with ThreadPoolExecutor() as pool:
            pool.submit(propagate_context(do_work, context=context), payload)

When an Azure Functions context object is supplied, the worker's thread_local_storage.invocation_id is also set for the duration of the call (and restored afterwards) so the worker's own logging handler correlates records emitted from the background thread. This is best-effort and duck-typed: any missing attribute or error is silently ignored (Principle 3: context propagation failures never crash the caller). No azure-functions import is required.

Parameters:

Name	Type	Description	Default
`func`	`Callable[_P, _R]`	The callable to run on a background thread. Called with whatever positional/keyword arguments the returned wrapper receives.	required
`context`	`Any`	Optional Azure Functions context object (`func.Context`). When provided, its `invocation_id` is propagated to the worker's `thread_local_storage` in addition to the `contextvars` snapshot.	`None`

Returns:

Type	Description
`Callable[_P, _R]`	A wrapper around func that applies the snapshotted context on entry and
`Callable[_P, _R]`	restores the previous state on exit. Reusable and concurrency-safe: it may
`Callable[_P, _R]`	be submitted to multiple threads simultaneously.

Usage Notes¶

Binds the current invocation context to a callable so it can run on a ThreadPoolExecutor worker or a manually created threading.Thread, which contextvars do not reach on their own.
Wrap inside the invocation, immediately before submitting the work; the context is snapshotted at wrap time.
Pass context=context to also propagate the Azure worker's thread_local_storage.invocation_id; propagation failures are silent and never crash the caller.

Example: ThreadPoolExecutor¶

from concurrent.futures import ThreadPoolExecutor

import azure.functions as func
from azure_functions_logging import logging_context, propagate_context


def handler(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    with logging_context(context):
        with ThreadPoolExecutor() as pool:
            future = pool.submit(propagate_context(do_work, context=context), payload)
            future.result()  # log records from do_work carry the invocation context
    return func.HttpResponse("ok")

propagating_executor¶

Return a :class:PropagatingExecutor wrapping pool (or a new pool).

Ergonomic entry point for background-thread context propagation that removes the per-submit :func:propagate_context boilerplate. See :class:PropagatingExecutor for full semantics.

Example::

from concurrent.futures import ThreadPoolExecutor
from azure_functions_logging import logging_context, propagating_executor

with logging_context(context):
    # Wrap an existing pool ...
    pool = propagating_executor(ThreadPoolExecutor(max_workers=4), context=context)
    # ... or let the helper create and own one:
    with propagating_executor(max_workers=4, context=context) as pool:
        pool.submit(do_work, payload)

Parameters:

Name	Type	Description	Default
`pool`	`Executor \| None`	Existing executor to wrap, or `None` to create a new :class:`~concurrent.futures.ThreadPoolExecutor`.	`None`
`context`	`Any`	Optional Azure Functions `context` for worker `thread_local_storage` propagation.	`None`
`**pool_kwargs`	`Any`	Forwarded to a newly created pool; rejected when pool is provided.	`{}`

Returns:

Name	Type	Description
`A`	`PropagatingExecutor`	class:`PropagatingExecutor` ready for :meth:`~PropagatingExecutor.submit`
	`PropagatingExecutor`	/ :meth:`~PropagatingExecutor.map`.

Source code in src/azure_functions_logging/_context.py

def propagating_executor(
    pool: Executor | None = None,
    *,
    context: Any = None,
    **pool_kwargs: Any,
) -> PropagatingExecutor:
    """Return a :class:`PropagatingExecutor` wrapping *pool* (or a new pool).

    Ergonomic entry point for background-thread context propagation that removes
    the per-submit :func:`propagate_context` boilerplate. See
    :class:`PropagatingExecutor` for full semantics.

    Example::

        from concurrent.futures import ThreadPoolExecutor
        from azure_functions_logging import logging_context, propagating_executor

        with logging_context(context):
            # Wrap an existing pool ...
            pool = propagating_executor(ThreadPoolExecutor(max_workers=4), context=context)
            # ... or let the helper create and own one:
            with propagating_executor(max_workers=4, context=context) as pool:
                pool.submit(do_work, payload)

    Args:
        pool: Existing executor to wrap, or ``None`` to create a new
            :class:`~concurrent.futures.ThreadPoolExecutor`.
        context: Optional Azure Functions ``context`` for worker
            ``thread_local_storage`` propagation.
        **pool_kwargs: Forwarded to a newly created pool; rejected when *pool* is
            provided.

    Returns:
        A :class:`PropagatingExecutor` ready for :meth:`~PropagatingExecutor.submit`
        / :meth:`~PropagatingExecutor.map`.
    """
    return PropagatingExecutor(pool, context=context, **pool_kwargs)

PropagatingExecutor¶

Bases: Executor

An :class:~concurrent.futures.Executor that auto-propagates invocation context.

contextvars do not follow work handed to a :class:~concurrent.futures.ThreadPoolExecutor worker, so records emitted from pooled threads lose their invocation_id unless every submitted callable is wrapped with :func:propagate_context. This executor removes that per-submit boilerplate: it wraps each callable passed to :meth:submit / :meth:map with :func:propagate_context at submission time, snapshotting the invocation context bound on the submitting thread.

Propagation stays explicit and opt-in at the executor boundary — the library never monkeypatches :mod:threading or :mod:concurrent.futures. Work submitted to a plain executor is unaffected. Propagation failures never crash the caller (Principle 3): a failed context application degrades to running the callable without context rather than raising.

The executor either wraps an existing pool or lazily creates a :class:~concurrent.futures.ThreadPoolExecutor::

from azure_functions_logging import logging_context, propagating_executor

def handler(req, context):
    with logging_context(context):
        with propagating_executor(context=context) as pool:
            pool.submit(do_work, payload)  # record carries invocation_id

Passing context= also propagates the Azure worker's thread_local_storage.invocation_id for the duration of each call, matching :func:propagate_context.

Parameters:

Name	Type	Description	Default
`pool`	`Executor \| None`	An existing :class:`~concurrent.futures.Executor` to wrap. When `None` (default), a new :class:`~concurrent.futures.ThreadPoolExecutor` is created from `**pool_kwargs` and owned by this instance (shut down on :meth:`shutdown` / context-manager exit).	`None`
`context`	`Any`	Optional Azure Functions `context` object. When supplied, its `invocation_id` is propagated to each worker's `thread_local_storage` in addition to the `contextvars` snapshot.	`None`
`**pool_kwargs`	`Any`	Forwarded to :class:`~concurrent.futures.ThreadPoolExecutor` when pool is `None`. Rejected (`TypeError`) when wrapping an existing pool.	`{}`

Source code in src/azure_functions_logging/_context.py

def __init__(
    self,
    pool: Executor | None = None,
    *,
    context: Any = None,
    **pool_kwargs: Any,
) -> None:
    if pool is None:
        self._pool: Executor = ThreadPoolExecutor(**pool_kwargs)
        self._owns_pool = True
    else:
        if pool_kwargs:
            raise TypeError(
                "pool_kwargs are only accepted when creating a new pool; "
                "they cannot be applied to an existing executor"
            )
        self._pool = pool
        self._owns_pool = False
    self._context = context

`pool` `property` ¶

The wrapped (or lazily created) underlying executor.

`map(fn, *iterables, timeout=None, chunksize=1)` ¶

Like :meth:Executor.map, wrapping fn for context propagation.

Source code in src/azure_functions_logging/_context.py

def map(
    self,
    fn: Callable[..., _R],
    *iterables: Iterable[Any],
    timeout: float | None = None,
    chunksize: int = 1,
) -> Iterator[_R]:
    """Like :meth:`Executor.map`, wrapping *fn* for context propagation."""
    wrapped = propagate_context(fn, context=self._context)
    return self._pool.map(wrapped, *iterables, timeout=timeout, chunksize=chunksize)

`shutdown(wait=True, *, cancel_futures=False)` ¶

Shut down the underlying executor (see :meth:Executor.shutdown).

Source code in src/azure_functions_logging/_context.py

def shutdown(self, wait: bool = True, *, cancel_futures: bool = False) -> None:
    """Shut down the underlying executor (see :meth:`Executor.shutdown`)."""
    self._pool.shutdown(wait=wait, cancel_futures=cancel_futures)

`submit(fn, /, *args, **kwargs)` ¶

Submit fn, wrapping it so the current invocation context propagates.

Source code in src/azure_functions_logging/_context.py

def submit(self, fn: Callable[..., _R], /, *args: Any, **kwargs: Any) -> Future[_R]:
    """Submit *fn*, wrapping it so the current invocation context propagates."""
    wrapped = propagate_context(fn, context=self._context)
    return self._pool.submit(wrapped, *args, **kwargs)

Usage Notes¶

Ergonomic alternative to wrapping every callable with propagate_context: the executor auto-wraps each callable passed to submit / map at submission time, snapshotting the invocation context bound on the submitting thread.
Propagation stays explicit and opt-in at the executor boundary — the library never monkeypatches threading / concurrent.futures. Work submitted to a plain executor is unaffected.
propagating_executor(pool) wraps an existing executor; with no pool it creates and owns a new ThreadPoolExecutor from the forwarded keyword arguments (e.g. max_workers=).
Pass context=context to also propagate the Azure worker's thread_local_storage.invocation_id; propagation failures never crash the caller.

Example: propagating executor¶

import azure.functions as func
from azure_functions_logging import logging_context, propagating_executor


def handler(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    with logging_context(context):
        with propagating_executor(max_workers=4, context=context) as pool:
            # No per-submit propagate_context() needed — records from do_work
            # carry the invocation context.
            futures = [pool.submit(do_work, item) for item in payload]
            for future in futures:
                future.result()
    return func.HttpResponse("ok")

reset_context¶

Clear every invocation context variable.

Use this for test teardown or defensive full cleanup. For normal context management, prefer token-based restore::

tokens = inject_context(context)
try:
    ...
finally:
    restore_context(tokens)

because token-based restore preserves any outer context.

Safe to call repeatedly. Setting to None is the documented "absent" state for every context field (matches ContextVar defaults).

Source code in src/azure_functions_logging/_context.py

def reset_context() -> None:
    """Clear every invocation context variable.

    Use this for test teardown or defensive full cleanup. For normal
    context management, prefer token-based restore::

        tokens = inject_context(context)
        try:
            ...
        finally:
            restore_context(tokens)

    because token-based restore preserves any outer context.

    Safe to call repeatedly. Setting to ``None`` is the documented \"absent\"
    state for every context field (matches ``ContextVar`` defaults).
    """
    invocation_id_var.set(None)
    function_name_var.set(None)
    trace_id_var.set(None)
    span_id_var.set(None)
    cold_start_var.set(None)

restore_context¶

Restore context variables to their previous state using tokens.

Tokens are single-use and must be restored in the same context where they were created. Calling this function twice with the same tokens raises RuntimeError from contextvars.

Parameters:

Name	Type	Description	Default
`tokens`	`ContextTokens`	Mapping returned by `inject_context()`.	required

Source code in src/azure_functions_logging/_context.py

def restore_context(tokens: ContextTokens) -> None:
    """Restore context variables to their previous state using tokens.

    Tokens are single-use and must be restored in the same context where
    they were created. Calling this function twice with the same tokens
    raises ``RuntimeError`` from ``contextvars``.

    Args:
        tokens: Mapping returned by ``inject_context()``.
    """
    for var, token in tokens.items():
        var.reset(token)

ContextTokens¶

End-to-End API Example¶

import logging
import azure.functions as func
from azure_functions_logging import get_logger, inject_context, setup_logging

setup_logging(level=logging.INFO, format="json")
logger = get_logger(__name__)

app = func.FunctionApp()


@app.route(route="orders")
def orders(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    inject_context(context)
    req_logger = logger.bind(route="/orders", method=req.method)
    req_logger.info("orders request started")
    req_logger.info("orders request completed")
    return func.HttpResponse("ok")

Name	Type	Description	Default
`func`	`_F \| None`	The handler function (when used without parentheses).	`None`
`param`	`str`	Name of the parameter that receives the Azure Functions context object. Defaults to `"context"`.	`_DEFAULT_PARAM`
`activate_trace_context`	`bool \| None`	When `True`, also attach the host's W3C trace context so OTel log records inherit the host span's `trace_id`/`span_id` (requires the `[otel]` extra; silent no-op otherwise). When `None` (default), the process-wide default configured via `setup_logging(activate_trace_context=...)` applies.	`None`
`strict`	`bool`	When `True`, raise :class:`ValueError` at decoration time if the handler signature declares no `param` parameter (and cannot receive it via `**kwargs`), since context injection would silently no-op. When `False` (default), the same condition emits a :class:`RuntimeWarning` instead, so the app keeps running.	`False`
`lifecycle`	`bool`	When `True`, emit opt-in invocation lifecycle records — an `"invocation start"` record before the handler runs and an `"invocation end"` record after it returns (or an `"invocation error"` record if it raises). End/error records carry `duration_ms` and `outcome` extras. Exceptions are logged then re-raised unchanged. Defaults to `False` (no output, zero overhead when disabled).	`False`
`lifecycle_level`	`int`	Log level for the start/end lifecycle records when `lifecycle=True`. Defaults to `logging.INFO`. Error records are always emitted at `logging.ERROR`.	`INFO`

API Reference¶

setup_logging¶

Usage Notes¶

Example¶

Example: Named Target Logger¶

Example: Invalid Format Handling¶

get_logger¶

Usage Notes¶

Example¶

Example: Root Logger Wrapper¶

FunctionLogger¶

name property ¶

__getattr__(name) ¶

bind(**kwargs) ¶

clear_context() ¶

critical(msg, *args, **kwargs) ¶

debug(msg, *args, **kwargs) ¶

error(msg, *args, **kwargs) ¶

exception(msg, *args, **kwargs) ¶

fatal(msg, *args, **kwargs) ¶

getEffectiveLevel() ¶

hasHandlers() ¶

info(msg, *args, **kwargs) ¶

isEnabledFor(level) ¶

log(level, msg, *args, **kwargs) ¶

setLevel(level) ¶

warn(msg, *args, **kwargs) ¶

warning(msg, *args, **kwargs) ¶

Usage Notes¶

Example: Binding Context¶

Example: Chained Binding¶

Example: Clearing Bound Context¶

Example: Exception Logging¶

JsonFormatter¶

format(record) ¶

Usage Notes¶

Example: Automatic Selection¶

Example: Manual Formatter Wiring¶

SamplingFilter¶

filter(record) ¶

Usage Notes¶

Example: Per-Logger Buckets for Azure SDK Logs¶

RedactionFilter¶

filter(record) ¶

Value / pattern-based redaction¶

DEFAULT_REDACTION_PATTERNS¶

AttributeFlattenFilter¶

filter(record) ¶

Usage Notes¶

inject_context¶

Usage Notes¶

Example: Azure Function Entrypoint¶

Example: Safe with Partial Context Object¶

with_context¶

get_logging_metadata¶

Usage Notes¶

Example¶

logging_context¶

propagate_context¶

Usage Notes¶

Example: ThreadPoolExecutor¶

propagating_executor¶

PropagatingExecutor¶

pool property ¶

map(fn, *iterables, timeout=None, chunksize=1) ¶

shutdown(wait=True, *, cancel_futures=False) ¶

submit(fn, /, *args, **kwargs) ¶

Usage Notes¶

Example: propagating executor¶

reset_context¶

restore_context¶

ContextTokens¶

End-to-End API Example¶

Cross-Reference¶

`name` `property` ¶

`getattr(name)` ¶

`bind(**kwargs)` ¶

`clear_context()` ¶

`critical(msg, *args, **kwargs)` ¶

`debug(msg, *args, **kwargs)` ¶

`error(msg, *args, **kwargs)` ¶

`exception(msg, *args, **kwargs)` ¶

`fatal(msg, *args, **kwargs)` ¶

`getEffectiveLevel()` ¶

`hasHandlers()` ¶

`info(msg, *args, **kwargs)` ¶

`isEnabledFor(level)` ¶

`log(level, msg, *args, **kwargs)` ¶

`setLevel(level)` ¶

`warn(msg, *args, **kwargs)` ¶

`warning(msg, *args, **kwargs)` ¶

`format(record)` ¶

`filter(record)` ¶

`filter(record)` ¶

`filter(record)` ¶

`pool` `property` ¶

`map(fn, *iterables, timeout=None, chunksize=1)` ¶

`shutdown(wait=True, *, cancel_futures=False)` ¶

`submit(fn, /, *args, **kwargs)` ¶