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: Installs ContextFilter on the root logger's handlers only. Does NOT add 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.
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.

This function is idempotent per logger_name — calling it multiple times for the same logger has no additional effect.

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 :func:`install_context_factory` 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.	`False`

.. 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,
) -> None:
    """Configure logging for the current environment.
    Behavior depends on the detected environment:

    - **Azure / Core Tools**: Installs ``ContextFilter`` on the root logger's
      handlers only. Does NOT add 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.
    - **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.

    This function is idempotent per ``logger_name`` — calling it multiple times
    for the same logger has no additional effect.

    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 :func:`install_context_factory`
            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.

    .. 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 does not leave persistent global side effects.
    if use_record_factory:
        install_context_factory()

    with _configured_lock:
        # 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()
                )
                _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 filter on root logger itself so handlers attached later
            # (before the next setup_logging() call) also inherit it.
            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)

        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()
            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)
            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 delegates all standard logging methods to the underlying logger. 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.

`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)

`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)

`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),
        "cold_start": getattr(record, "cold_start", None),
        "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.

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.	`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()
    self._counts: dict[str, int] = {}
    self._window_starts: dict[str, float] = {}

`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:
            window_start = self._window_starts.get(record.name)
            if window_start is None or now - window_start >= self._window:
                self._window_starts[record.name] = now
                self._counts[record.name] = 1
                return True
            self._counts[record.name] = self._counts.get(record.name, 0) + 1
            return self._counts[record.name] <= 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 lowercased name is in sensitive_keys with "***".

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). When None, uses the built-in default set (23 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`.	`None`
`name`	`str`	Optional logger-name scope. When set, only matching loggers are subject to redaction; non-matching records pass through unchanged.	`''`

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 = "",
) -> None:
    super().__init__(name)
    self._sensitive_keys: frozenset[str] = (
        frozenset(k.lower() for k in sensitive_keys)
        if sensitive_keys is not None
        else _DEFAULT_SENSITIVE_KEYS
    )

`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 key.lower() in self._sensitive_keys:
                    setattr(record, key, _MASK)
                else:
                    value = record.__dict__[key]
                    if isinstance(value, (dict, list)):
                        setattr(record, key, _redact_value(value, self._sensitive_keys))
            except Exception:  # nosec B110 — one broken field must not stop others
                pass
    except Exception:  # nosec B110 — filter must never raise
        pass
    return True

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
        tokens[trace_id_var] = trace_id_var.set(_extract_trace_id(trace_parent))
    except Exception:  # nosec B110 — Principle 3: context failures are silent
        tokens[trace_id_var] = trace_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")

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")

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 ¶

bind(**kwargs) ¶

clear_context() ¶

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

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

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

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

getEffectiveLevel() ¶

hasHandlers() ¶

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

isEnabledFor(level) ¶

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

setLevel(level) ¶

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) ¶

inject_context¶

Usage Notes¶

Example: Azure Function Entrypoint¶

Example: Safe with Partial Context Object¶

End-to-End API Example¶

Cross-Reference¶

`name` `property` ¶

`bind(**kwargs)` ¶

`clear_context()` ¶

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

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

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

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

`getEffectiveLevel()` ¶

`hasHandlers()` ¶

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

`isEnabledFor(level)` ¶

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

`setLevel(level)` ¶

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

`format(record)` ¶

`filter(record)` ¶

`filter(record)` ¶