Skip to content

Logging service

consortium.server.services.logging_service

SinkInfo(handler_id, sink, level, label, is_server_default, sink_kwargs=dict()) dataclass

handler_id instance-attribute

sink instance-attribute

level instance-attribute

label instance-attribute

is_server_default instance-attribute

sink_kwargs = field(default_factory=dict) class-attribute instance-attribute

LoggingService()

logging_config = LoggingConfigModel() instance-attribute

add_sink(sink, level, label, format=None, is_server_default=False, **kwargs)

Adds a new log sink to the logging service.

All additional keyword arguments are forwarded directly to loguru.logger.add.

Parameters:

Name Type Description Default
sink Any

The sink target, for example sys.stdout, a file path, or a callable.

 required
level str

The minimum log level for this sink, for example "DEBUG" or "INFO".

 required
label str

A unique label for identifying and referencing this sink.

 required
format str | Callable[[Any], str] | None

A loguru format string or callable. When None, the service's default formatter is used.

 None
is_server_default bool

When True, marks this sink as a server-default sink. Defaults to False.

 False
**kwargs

Additional keyword arguments passed to loguru.logger.add.

 {}

Returns:

Type Description
int

The handler ID returned by loguru.logger.add.

Raises:

Type Description
SinkLabelAlreadyExistsError

If a sink with the given label is already registered.
SinkConfigurationError

If loguru.logger.add rejects the sink configuration, for example an invalid level, filter, or format string, or sink or one of the forwarded keyword arguments is of an invalid type.
SinkFileSystemError

If sink is a file path that cannot be opened, for example because the containing directory does not exist.

remove_sink(label)

Removes a registered log sink by its label.

Parameters:

Name Type Description Default
label str

The label of the sink to remove.

 required

Raises:

Type Description
SinkNotFoundError

If no sink with the given label is registered.
StaleSinkHandlerError

If the sink's handler ID is no longer registered with loguru.

remove_all_sinks()

Removes all registered log sinks.

Raises:

Type Description
StaleSinkHandlerError

If a sink's handler ID is no longer registered with loguru (raised via remove_sink).

modify_sink(label, sink=_UNSET, **overrides)

Modifies an existing log sink in-place by rebuilding it with updated parameters.

Because loguru provides no update API, the existing sink is torn down and reconstructed with the merged configuration. Successive calls layer correctly because SinkInfo.sink_kwargs is kept up to date after each modification.

The existing handler is removed via loguru.logger.remove before the replacement is added via loguru.logger.add. If rebuilding the replacement fails, the sink is re-added with its previous configuration so that logging through it continues uninterrupted, and the label keeps pointing at the (restored) handler's current ID, before the typed error describing the rebuild failure is raised. If that rollback itself fails, the sink is dropped from this service's bookkeeping (it is no longer registered with loguru either way) and the rollback failure is logged, while the original error is still raised to the caller so the reason the modification failed is never masked.

Parameters:

Name Type Description Default
label str

The label of the sink to modify.

 required
sink Any

A replacement sink target. When omitted, the existing sink target is preserved.

 _UNSET
**overrides

Additional loguru logger.add keyword arguments to update. Provided values are merged over the existing sink kwargs.

 {}

Raises:

Type Description
SinkNotFoundError

If no sink with the given label is registered.
StaleSinkHandlerError

If the existing handler's ID is no longer registered with loguru.
SinkConfigurationError

If loguru.logger.add rejects the merged configuration when rebuilding the sink, for example an invalid level, filter, or format string, or the replacement sink or one of the merged keyword arguments is of an invalid type.
SinkFileSystemError

If the replacement sink is a file path that cannot be opened.

get_all_sinks()

Returns all currently registered log sinks.

Returns:

Type Description
list[SinkInfo]

A list of SinkInfo objects for all registered sinks, or an empty list if
list[SinkInfo]

none have been added.

configure_default_logging(logging_config)

Configures default log level colors and registers the server's default sinks.

Sets display colors for each log level and adds a stdout sink. A file sink is also added when logging_config.log_file is not None.

Parameters:

Name Type Description Default
logging_config LoggingConfigModel

The logging configuration model specifying the log level, colorize flag, and optional log file path, rotation policy, and retention policy.

 required

Raises:

Type Description
StaleSinkHandlerError

If a stale handler ID is encountered while removing the existing server-default sinks (raised via remove_sink).
SinkLabelAlreadyExistsError

If the stdout or file label collides with an already-registered non-default sink (raised via add_sink).
SinkConfigurationError

If loguru.logger.add rejects a sink configuration (for example, an invalid level or format string), or a sink or one of its keyword arguments is of an invalid type.
SinkFileSystemError

If logging_config.log_file is a path that cannot be opened.