Skip to content

API Reference

This reference documents the public API exported by azure_functions_logging.

Use this page together with:

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 handlers attached later also receive context fields). 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.

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 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.

 False
extra_context_vars dict[str, ContextVar[Any]] | None

Optional mapping of field_name -> ContextVar whose current values ContextFilter also copies onto each LogRecord, alongside the four built-in context fields. Field names must not collide with the built-in fields. Only applies to the ContextFilter strategy (ignored when use_record_factory=True).

 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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
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,
) -> 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 handlers attached later also receive context
      fields). 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.

    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 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.
        extra_context_vars: Optional mapping of ``field_name -> ContextVar``
            whose current values ``ContextFilter`` also copies onto each
            LogRecord, alongside the four built-in context fields. Field names
            must not collide with the built-in fields. Only applies to the
            ``ContextFilter`` strategy (ignored when ``use_record_factory=True``).

    .. 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(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 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(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)
            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 
42
43
44
45
46
47
48
49
50
51
52
53
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 
60
61
62
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 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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 
85
86
87
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 
131
132
133
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 
115
116
117
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 
127
128
129
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 
135
136
137
138
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 
148
149
150
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 
161
162
163
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 
119
120
121
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 
144
145
146
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 
152
153
154
155
156
157
158
159
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 
140
141
142
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 
123
124
125
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 
133
134
135
136
137
138
139
140
141
142
143
144
145
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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
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.
  • 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 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
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.

 ''
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 
221
222
223
224
225
226
227
228
229
230
231
def __init__(
    self,
    sensitive_keys: Iterable[str] | None = None,
    name: str = "",
) -> 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
    )

filter(record)

Redact sensitive fields on the record. Always returns True.

Source code in src/azure_functions_logging/_filters.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
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 _normalize_key(key) 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 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
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")

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:

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

Both sync and async handlers are supported.

Parameters:

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
Source code in src/azure_functions_logging/_decorator.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def with_context(
    func: _F | None = None,
    *,
    param: str = _DEFAULT_PARAM,
) -> _F | Callable[[_F], _F]:
    """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:

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

    Both sync and async handlers are supported.

    Args:
        func: The handler function (when used without parentheses).
        param: Name of the parameter that receives the Azure Functions
            context object. Defaults to ``"context"``.
    """

    def decorator(fn: _F) -> _F:
        if asyncio.iscoroutinefunction(fn):
            return _wrap_async(fn, param)
        return _wrap_sync(fn, param)

    if func is not None:
        # Called as @with_context (no parentheses)
        return decorator(func)

    # Called as @with_context(...) (with parentheses)
    return decorator

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 
167
168
169
170
171
172
173
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)
# -> {"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.

Source code in src/azure_functions_logging/_context.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
@contextmanager
def logging_context(context: Any) -> 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.
    """
    tokens = inject_context(context)
    try:
        yield
    finally:
        restore_context(tokens)

install_context_factory

Deprecated shim for the LogRecordFactory injection strategy.

.. deprecated:: Direct use of install_context_factory is deprecated in favour of the single setup_logging(use_record_factory=True) entry point, which selects the LogRecordFactory injection strategy and manages ContextFilter teardown consistently. This shim will be removed in a future release.

Source code in src/azure_functions_logging/_context.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def install_context_factory() -> None:
    """Deprecated shim for the LogRecordFactory injection strategy.

    .. deprecated::
        Direct use of ``install_context_factory`` is deprecated in favour of
        the single ``setup_logging(use_record_factory=True)`` entry point,
        which selects the ``LogRecordFactory`` injection strategy and manages
        ``ContextFilter`` teardown consistently. This shim will be removed in
        a future release.
    """
    warnings.warn(
        "install_context_factory() is deprecated; use "
        "setup_logging(use_record_factory=True) instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    _install_context_factory()

Deprecated

Direct use of install_context_factory() is deprecated. Prefer setup_logging(use_record_factory=True), which selects the LogRecordFactory injection strategy and manages teardown consistently.

uninstall_context_factory

Restore the LogRecordFactory that preceded :func:install_context_factory.

This is the inverse of :func:install_context_factory and is primarily intended for test teardown, where leaving a global factory installed would leak context injection into unrelated tests.

Only the factory installed by this package is removed: the previously active factory (captured at install time, including any chained customizations) is restored as the global LogRecordFactory.

Returns:

Type Description
bool

True if this package's context factory was active and has been
bool

uninstalled; False if no such factory was active (no-op).
Source code in src/azure_functions_logging/_context.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def uninstall_context_factory() -> bool:
    """Restore the ``LogRecordFactory`` that preceded :func:`install_context_factory`.

    This is the inverse of :func:`install_context_factory` and is primarily
    intended for test teardown, where leaving a global factory installed would
    leak context injection into unrelated tests.

    Only the factory installed by this package is removed: the previously
    active factory (captured at install time, including any chained
    customizations) is restored as the global ``LogRecordFactory``.

    Returns:
        ``True`` if this package's context factory was active and has been
        uninstalled; ``False`` if no such factory was active (no-op).
    """
    current_factory = logging.getLogRecordFactory()
    if not getattr(current_factory, _CONTEXT_FACTORY_MARKER, False):
        return False

    previous_factory = getattr(current_factory, _CONTEXT_FACTORY_PREVIOUS, None)
    if previous_factory is None:  # pragma: no cover - defensive; always set on install
        previous_factory = logging.LogRecord
    logging.setLogRecordFactory(previous_factory)
    return True

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 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
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)
    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 
214
215
216
217
218
219
220
221
222
223
224
225
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")

Cross-Reference