logging

azure_bootstrap.logging

Tier 1 logging primitives.

Always-on, stdlib-only. The top-level azure_bootstrap package re-exports the most common entry-points (configure_logging, correlation_scope, mask_*). Deeper imports stay available for callers that want fine-grained access.

Modules:

Name	Description
config	Single entry-point for v2 logging setup.
correlation	Correlation context for structured logs.
formatter	v2 log formatter: appends extra={} fields as key=repr(value) pairs.
jsonformatter	v2.1 JSON log formatter — one JSON object per record.
masking	Secret/PII masking and log-injection sanitization primitives.
noise	Silence chatty third-party loggers.

Classes:

Name	Description
CorrelationFilter	Attach every set context var as a record attribute.
ExtraFieldsFormatter	Render structured extras after the base format string.
LoggingExtraConflictError	Raised in DEBUG when a caller's extra={} collides with a reserved key.
JsonLogFormatter	Render a log record as a single line of JSON.

Functions:

Name	Description
configure_logging	Install structured-logging defaults. Idempotent — replaces handlers.
debug_logging_enabled	Second-factor gate on DEBUG output.
correlation_scope	Push correlation context for the duration of the with-block.
mask_secrets_in_dict	Shallow copy with secret-keyed values replaced by '***'.
register_secret_keys	Extend the allowlist at runtime. Names are lowercased.
sanitize_for_log	Replace control chars (\x00-\x1f, \x7f) with '?' and truncate.
register_noisy_logger	Append a logger name to the default registry for the lifetime of the process.
silence_noisy_loggers	Clamp the named loggers to level.

CorrelationFilter

Bases: Filter

Attach every set context var as a record attribute.

Only sets attributes the record doesn't already carry — so explicit extra={"correlation_id": "..."} overrides win.

ExtraFieldsFormatter

ExtraFieldsFormatter(fmt: str | None = '%(asctime)s %(levelname)s %(name)s %(message)s', datefmt: str | None = None, style: str = '%', validate: bool = True)

Bases: Formatter

Render structured extras after the base format string.

Output: <base> key1='value1' key2=42 key3=<MyObj>. Two-space gap is intentional so grep ' key=' finds field hits without matching the message body.

Source code in azure_bootstrap/logging/formatter.py

def __init__(
    self,
    fmt: str | None = "%(asctime)s %(levelname)s %(name)s %(message)s",
    datefmt: str | None = None,
    style: str = "%",
    validate: bool = True,
) -> None:
    super().__init__(fmt=fmt, datefmt=datefmt, style=style, validate=validate)  # type: ignore[arg-type]

LoggingExtraConflictError

Bases: Exception

Raised in DEBUG when a caller's extra={} collides with a reserved key.

JsonLogFormatter

JsonLogFormatter(*, ensure_ascii: bool = False, mask_extras: bool = True)

Bases: Formatter

Render a log record as a single line of JSON.

Emitted fields: timestamp (ISO-8601 UTC), level, logger, message, exception (only when exc_info is present), plus every non-reserved extra={} field (correlation IDs included, since :class:CorrelationFilter attaches them as record attributes). Extra fields are passed through :func:mask_secrets_in_dict so secret-keyed values are redacted to ***.

Source code in azure_bootstrap/logging/jsonformatter.py

def __init__(self, *, ensure_ascii: bool = False, mask_extras: bool = True) -> None:
    super().__init__()
    self._ensure_ascii = ensure_ascii
    self._mask_extras = mask_extras

configure_logging

configure_logging(*, format_string: str = '%(asctime)s %(levelname)s %(name)s %(message)s', silence_defaults: bool = True, extra_noisy_loggers: tuple[str, ...] = ()) -> None

Install structured-logging defaults. Idempotent — replaces handlers.

Source code in azure_bootstrap/logging/config.py

def configure_logging(
    *,
    format_string: str = "%(asctime)s %(levelname)s %(name)s %(message)s",
    silence_defaults: bool = True,
    extra_noisy_loggers: tuple[str, ...] = (),
) -> None:
    """Install structured-logging defaults. Idempotent — replaces handlers."""
    logging.setLoggerClass(_StrictLogger)

    level = effective_log_level()
    handler = logging.StreamHandler()
    handler.setFormatter(ExtraFieldsFormatter(format_string))
    handler.addFilter(CorrelationFilter())

    logging.basicConfig(level=level, handlers=[handler], force=True)

    root = logging.getLogger()
    if not any(isinstance(f, CorrelationFilter) for f in root.filters):
        root.addFilter(CorrelationFilter())

    silence_noisy_loggers(*extra_noisy_loggers, include_defaults=silence_defaults)

debug_logging_enabled

debug_logging_enabled() -> bool

Second-factor gate on DEBUG output.

LOG_LEVEL=DEBUG alone is not enough — DEBUG_LOGGING_ENABLED must also be truthy. Belt-and-suspenders against a stray manifest leaking DEBUG into prod.

Source code in azure_bootstrap/logging/config.py

def debug_logging_enabled() -> bool:
    """Second-factor gate on DEBUG output.

    ``LOG_LEVEL=DEBUG`` alone is not enough — ``DEBUG_LOGGING_ENABLED`` must
    also be truthy. Belt-and-suspenders against a stray manifest leaking
    DEBUG into prod.
    """
    return env_flag("DEBUG_LOGGING_ENABLED", default=False)

correlation_scope

correlation_scope(correlation_id: str | None = None, **fields: str | None) -> Generator[str, None, None]

Push correlation context for the duration of the with-block.

Yields the resolved correlation_id, always a non-empty string. A fresh 12-char uuid hex is minted when correlation_id is None. Any keyword argument becomes a context var (e.g. email_id, request_id).

Source code in azure_bootstrap/logging/correlation.py

@contextmanager
def correlation_scope(
    correlation_id: str | None = None,
    **fields: str | None,
) -> Generator[str, None, None]:
    """Push correlation context for the duration of the with-block.

    Yields the resolved correlation_id, always a non-empty string. A fresh
    12-char uuid hex is minted when ``correlation_id`` is None. Any keyword
    argument becomes a context var (e.g. ``email_id``, ``request_id``).
    """
    resolved = correlation_id or uuid.uuid4().hex[:12]
    tokens: list[tuple[contextvars.ContextVar[str | None], contextvars.Token[str | None]]] = []
    tokens.append((_var_for("correlation_id"), _var_for("correlation_id").set(resolved)))
    for key, value in fields.items():
        if value is None:
            continue
        var = _var_for(key)
        tokens.append((var, var.set(value)))
    try:
        yield resolved
    finally:
        for var, token in reversed(tokens):
            try:
                var.reset(token)
            except Exception:
                pass

mask_secrets_in_dict

mask_secrets_in_dict(d: dict[str, Any]) -> dict[str, Any]

Shallow copy with secret-keyed values replaced by '***'.

A value is replaced only when truthy — empty strings, None, 0, etc. pass through unmodified so the caller can still see "field was empty".

Source code in azure_bootstrap/logging/masking.py

def mask_secrets_in_dict(d: dict[str, Any]) -> dict[str, Any]:
    """Shallow copy with secret-keyed values replaced by '***'.

    A value is replaced only when truthy — empty strings, None, 0, etc. pass
    through unmodified so the caller can still see "field was empty".
    """
    out: dict[str, Any] = {}
    for key, value in d.items():
        if isinstance(key, str) and key.lower() in _SECRET_KEY_ALLOWLIST and value:
            out[key] = "***"
        else:
            out[key] = value
    return out

register_secret_keys

register_secret_keys(*names: str) -> None

Extend the allowlist at runtime. Names are lowercased.

Source code in azure_bootstrap/logging/masking.py

def register_secret_keys(*names: str) -> None:
    """Extend the allowlist at runtime. Names are lowercased."""
    for name in names:
        if name:
            _SECRET_KEY_ALLOWLIST.add(name.lower())

sanitize_for_log

sanitize_for_log(value: str | None, *, max_len: int = 256) -> str

Replace control chars (\x00-\x1f, \x7f) with '?' and truncate.

Source code in azure_bootstrap/logging/masking.py

def sanitize_for_log(value: str | None, *, max_len: int = 256) -> str:
    """Replace control chars (\\x00-\\x1f, \\x7f) with '?' and truncate."""
    if value is None:
        return ""
    cleaned = _CONTROL_CHARS_FOR_LOGS.sub("?", value)
    if len(cleaned) > max_len:
        return cleaned[:max_len] + "...[truncated]"
    return cleaned

register_noisy_logger

register_noisy_logger(name: str) -> None

Append a logger name to the default registry for the lifetime of the process.

Source code in azure_bootstrap/logging/noise.py

def register_noisy_logger(name: str) -> None:
    """Append a logger name to the default registry for the lifetime of the process."""
    if name and name not in _DEFAULT_NOISY_LOGGERS:
        _DEFAULT_NOISY_LOGGERS.append(name)

silence_noisy_loggers

silence_noisy_loggers(*names: str, level: int = WARNING, include_defaults: bool = True) -> None

Clamp the named loggers to level.

Only the named loggers are affected — root is never touched, so callers can leave root at DEBUG while these stay quiet.

Source code in azure_bootstrap/logging/noise.py

def silence_noisy_loggers(
    *names: str,
    level: int = logging.WARNING,
    include_defaults: bool = True,
) -> None:
    """Clamp the named loggers to ``level``.

    Only the named loggers are affected — root is never touched, so callers
    can leave root at DEBUG while these stay quiet.
    """
    targets: list[str] = []
    if include_defaults:
        targets.extend(_DEFAULT_NOISY_LOGGERS)
    targets.extend(names)
    seen: set[str] = set()
    for name in targets:
        if not name or name in seen:
            continue
        seen.add(name)
        logging.getLogger(name).setLevel(level)