Logging

Logger

class ioiocore.logging.Logger(directory=None)

Bases: Interface

Logger interface for writing and retrieving log entries.

Parameters:

directory (str | None)

__init__(directory=None)

Initialize a logger instance.

Parameters:

directory (Optional[str]) – Directory for log storage, by default None.

property attempted_file: str

The path file logging was attempted at, when it failed.

Returns:

The path, or None when the log file was created.

Return type:

str

property file_error: str

Why file logging is unavailable.

File logging is optional – an unwritable directory must not stop a pipeline – but without the reason a reader is left guessing between a permissions problem, a missing drive and a bad path.

Returns:

A description including the exception type, or None when the log file was created successfully.

Return type:

str

property file_name: str

Retrieve the log file name.

Returns:

The log file name, or None when file logging is unavailable.

Return type:

str

flush()

Flush the log buffer.

get_all()

Retrieve all log entries.

Returns:

A list of all log entries.

Return type:

list

get_by_type(type)

Retrieve log entries of a specific type.

Parameters:

type (LogTypes) – The log type to filter by.

Returns:

A list of matching log entries.

Return type:

list

get_incident(type, opaque=False)

Retrieve the last ERROR or WARNING from the never-evicted record.

Unlike get_last(), this is unaffected by ring-buffer eviction, so a burst of log traffic cannot hide an incident.

Parameters:

• type (LogTypes) – The incident type to retrieve (ERROR or WARNING).

• opaque (bool) – Whether to return a given incident only once.

Returns:

The last incident of the given type, if any.

Return type:

Optional[LogEntry]

get_last(type, opaque=False)

Retrieve the last logged entry of a specific type.

Parameters:

• type (LogTypes) – The log type to filter by.

• opaque (bool) – Whether to return the last log entry only once (returning None on repeated calls).

Returns:

The last log entry of the specified type, if available.

Return type:

Optional[LogEntry]

has_entries(type=None)

Check if there are log entries of a specific type.

Parameters:

type (LogTypes) – The log type to check, by default None.

Returns:

True if there are matching log entries, False otherwise.

Return type:

bool

is_persistent()

Whether log entries are written to a file.

Returns:

True if a log file is being written, False if the logger is operating in memory only (e.g. the log directory could not be created).

Return type:

bool

stop()

Stop the logger: flush pending entries, close the log file and join the writer thread. Safe to call more than once.

write(type, message)

Write a log entry.

Parameters:

• type (LogTypes) – Type of the log entry.

• message (Union[str, Exception]) – Log message.

Returns:

The created log entry.

Return type:

LogEntry

LogEntry

class ioiocore.logging.LogEntry(type, source, message)

Bases: Interface

Represents a log entry containing type, source trace, and message.

Parameters:

• type (LogTypes)

• source (dict)

• message (str)

__init__(type, source, message)

Initialize a log entry.

Parameters:

• type (LogTypes) – Type of the log entry.

• source (dict) – source trace information.

• message (str) – Log message.

keys()

Retrieve all keys in the log entry.

Returns:

An iterable of keys.

Return type:

Iterable

to_formatted_string()

Retrieve a formatted string representation of the log entry.

Returns:

Formatted log entry string.

Return type:

str

values()

Retrieve all values in the log entry.

Returns:

An iterable of values.

Return type:

Iterable