manager.logger

manager.logger

Audit-grade logging for spotforecast2-safe.

This module sets up a dual-handler logger: a console handler for humans and a file handler that writes one JSON object per line, conforming to the schema at audit_log_schema.json next to this file. The file is the compliance-relevant sink (EU AI Act Article 12, IEC 62443-4-2 SAR 6.1, IEC 61508-3 §7.4.7); the console handler is plaintext for interactive use.

Schema versioning rule: SCHEMA_VERSION is loaded directly from audit_log_schema.json at import time, so there is exactly one source of truth. Any change to the schema file is guarded by the CI job audit-log-schema-gate (.github/workflows/ci.yml), which rejects a pull request that touches the schema without a Conventional-Commits breaking-change marker (type!:) on at least one commit. That marker cascades to a MAJOR semantic-release bump.

Classes

Name	Description
JsonAuditFormatter	Format `LogRecord` instances as single-line JSON per audit_log_schema.json.

JsonAuditFormatter

manager.logger.JsonAuditFormatter()

Format LogRecord instances as single-line JSON per audit_log_schema.json.

The formatter emits exactly the fields named in the schema’s properties section, never more. Callers pass optional structured context through the standard logging extra= mechanism; recognised extras are event, task, and context.

Examples

import io
import json
import logging

from spotforecast2_safe.manager.logger import JsonAuditFormatter, SCHEMA_VERSION

formatter = JsonAuditFormatter()
stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.setFormatter(formatter)

logger = logging.getLogger("example.audit")
logger.setLevel(logging.DEBUG)
# Avoid duplicate handlers across repeated runs in the same process
logger.handlers.clear()
logger.addHandler(handler)
logger.propagate = False

logger.info(
    "model fitted",
    extra={"event": "fit", "task": "demo", "context": {"lags": 3}},
)

line = stream.getvalue().strip()
record = json.loads(line)
assert record["schema_version"] == SCHEMA_VERSION
assert record["event"] == "fit"
assert record["task"] == "demo"
assert record["context"] == {"lags": 3}
print(f"schema_version={record['schema_version']} event={record['event']}")

schema_version=1.0.0 event=fit

Methods

Name	Description
format	Format a `LogRecord` as a single-line JSON string.

format

manager.logger.JsonAuditFormatter.format(record)

Format a LogRecord as a single-line JSON string.

Parameters

Name	Type	Description	Default
record	logging.LogRecord	The log record to format.	required

Returns

Name	Type	Description
	str	A JSON string containing the required audit fields plus any
	str	optional `event`, `task`, `context`, and `exception`
	str	extras carried by the record.

Examples

import json
import logging
import time

from spotforecast2_safe.manager.logger import JsonAuditFormatter, SCHEMA_VERSION

formatter = JsonAuditFormatter()

record = logging.LogRecord(
    name="test.logger",
    level=logging.WARNING,
    pathname="",
    lineno=0,
    msg="threshold exceeded",
    args=(),
    exc_info=None,
)
record.__dict__.update({"event": "threshold_check", "task": "predict"})

line = formatter.format(record)
payload = json.loads(line)

assert payload["schema_version"] == SCHEMA_VERSION
assert payload["level"] == "WARNING"
assert payload["message"] == "threshold exceeded"
assert payload["event"] == "threshold_check"
assert payload["task"] == "predict"
print(f"level={payload['level']} message={payload['message']}")

level=WARNING message=threshold exceeded

Functions

Name	Description
setup_logging	Configure dual-handler logging for safety-critical execution.

setup_logging

manager.logger.setup_logging(level=logging.INFO, log_dir=None)

Configure dual-handler logging for safety-critical execution.

Attaches a stream handler (stdout, human-readable plaintext) and, when log_dir is provided, a file handler that writes JSON records per audit_log_schema.json. The console handler honours level; the file handler is always at INFO so the audit trail stays complete even when the operator silences the console.

Parameters

Name	Type	Description	Default
level	int	Logging level for console output. Default: `logging.INFO`.	`logging.INFO`
log_dir	Optional[Path]	Optional directory for the audit log file. If provided, a timestamped `task_safe_n_to_1_YYYYMMDD_HHMMSS.log` file is created and receives JSON-formatted records.	`None`

Returns

Name	Type	Description
	logging.Logger	Tuple of the configured logger and the audit log file path (or
	Optional[Path]	`None` if `log_dir` was omitted).

Raises

Name	Type	Description
	OSError	If `log_dir` is provided but the directory cannot be created or the audit log file cannot be opened. In a safety-critical workflow a missing audit trail is not recoverable — the failure surfaces immediately rather than degrading silently to console-only logging.

Examples

import json
import logging
import tempfile
from pathlib import Path

from spotforecast2_safe.manager.logger import setup_logging

# Reset the named logger so the example is idempotent when the notebook
# kernel re-runs this cell.
named = logging.getLogger("task_safe_n_to_1")
named.handlers.clear()

with tempfile.TemporaryDirectory() as tmp:
    log_dir = Path(tmp)
    logger, log_path = setup_logging(level=logging.WARNING, log_dir=log_dir)

    assert log_path is not None
    assert log_path.exists()

    logger.info("pipeline started", extra={"event": "task_start"})

    lines = [l for l in log_path.read_text(encoding="utf-8").splitlines() if l]
    assert len(lines) >= 1
    record = json.loads(lines[0])
    assert record["event"] == "audit_log_init"
    print(f"log file created: {log_path.name}")
    print(f"first record event: {record['event']}")

# Tear down so subsequent cells start clean
named.handlers.clear()

log file created: task_safe_n_to_1_20260602_232925.log
first record event: audit_log_init