simet.log.logger

simet.log.logger

Structured logging utilities.

Provides: - JSONFormatter: a logging.Formatter that emits JSON lines with ISO-8601 UTC timestamps, optional field remapping, exception/stack info, and all custom LogRecord extras. - NonErrorFilter: a filter that only allows records up to and including INFO level (i.e., it filters out WARNING and above).

JSONFormatter

JSONFormatter(*, fmt_keys=None)

Bases: Formatter

Emit log records as JSON lines with UTC timestamps.

The formatter builds a dictionary from the LogRecord, adds a UTC ISO-8601 timestamp and the rendered message, optionally remaps fields according to fmt_keys, and finally injects any custom record attributes (i.e., extras not in the built-in set).

By default, the output includes: - "message": record.getMessage() - "timestamp": record.created formatted as ISO-8601 in UTC - "exc_info" and/or "stack_info" when present - all custom attributes found on the record (extras)

Parameters:

Name	Type	Description	Default
fmt_keys	dict[str, str] | None	Optional mapping output_key → source_name to rename or surface specific fields. For each pair: * If source_name is "message" or "timestamp", that value is taken from the computed fields and removed from the defaults (so it won’t also appear under the default key). * Otherwise, getattr(record, source_name) is used. Example: {"level": "levelname", "logger": "name", "time": "timestamp"}	None

Notes

• Non-JSON-serializable values are converted with default=str in json.dumps.
• Timestamps are generated from record.created in UTC: YYYY-MM-DDTHH:MM:SS.mmmmmm+00:00.
• Extras (e.g., provided via logger.info("..", extra={"run_id": 7})) are included as top-level keys unless they collide with remapped keys.

Example

handler = logging.StreamHandler() handler.setFormatter(JSONFormatter(fmt_keys={"level": "levelname", "logger": "name"})) logger = logging.getLogger("app") logger.addHandler(handler) logger.setLevel(logging.INFO) logger.info("hello", extra={"run_id": 123}) # doctest: +SKIP {"level": "INFO", "logger": "app", "message": "hello", "timestamp": "...", "run_id": 123}

Initialize the JSON formatter.

Parameters:

Name	Type	Description	Default
fmt_keys	dict[str, str] | None	Optional output_key → source_name mapping; see class docstring.	None

Source code in simet/log/logger.py

def __init__(
    self,
    *,
    fmt_keys: dict[str, str] | None = None,
):
    """Initialize the JSON formatter.

    Args:
        fmt_keys: Optional output_key → source_name mapping; see class docstring.
    """
    super().__init__()
    self.fmt_keys = fmt_keys if fmt_keys is not None else {}

format

format(record)

Format a LogRecord as a JSON string.

Parameters:

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

Returns:

Name	Type	Description
str	str	A single JSON line representing the record.

Source code in simet/log/logger.py

@override
def format(self, record: logging.LogRecord) -> str:
    """Format a `LogRecord` as a JSON string.

    Args:
        record: The log record to format.

    Returns:
        str: A single JSON line representing the record.
    """
    message = self._prepare_log_dict(record)
    return json.dumps(message, default=str)

NonErrorFilter

Bases: Filter

Allow only log records up to and including INFO level.

Useful to split output by severity, e.g., one handler for INFO-and-below and another handler for WARNING-and-above.

Returns True for levels <= INFO, causing the record to pass through the handler this filter is attached to; returns False otherwise.

Example

info_handler = logging.StreamHandler() info_handler.addFilter(NonErrorFilter()) # passes DEBUG/INFO err_handler = logging.StreamHandler() err_handler.setLevel(logging.WARNING) # handles WARNING+

log = logging.getLogger("app") log.addHandler(info_handler) log.addHandler(err_handler) log.info("hello") # goes to info_handler log.warning("oops") # goes only to err_handler

filter

filter(record)

Return True iff the record's level is <= INFO.

Parameters:

Name	Type	Description	Default
record	LogRecord	The log record to test.	required

Returns:

Type	Description
bool | LogRecord	bool | logging.LogRecord: True to allow the record; False to drop it.

Source code in simet/log/logger.py

@override
def filter(self, record: logging.LogRecord) -> bool | logging.LogRecord:
    """Return True iff the record's level is <= INFO.

    Args:
        record: The log record to test.

    Returns:
        bool | logging.LogRecord: True to allow the record; False to drop it.
    """
    return record.levelno <= logging.INFO