academy.logging.helpers

academy/logging/helpers.py

JSONHandler

JSONHandler(filename: Path)

Bases: Handler

A LogHandler which outputs records as JSON objects, one per line.

Source code in academy/logging/helpers.py

def __init__(self, filename: pathlib.Path) -> None:
    super().__init__()
    self.f = open(filename, 'w')  # noqa: SIM115

emit

emit(record: LogRecord) -> None

Emits the log record as a JSON object.

Each attribute (including extra attributes) of the log record becomes an entry in the JSON object. Each value is rendered using str.

Source code in academy/logging/helpers.py

def emit(self, record: logging.LogRecord) -> None:
    """Emits the log record as a JSON object.

    Each attribute (including extra attributes) of the log record becomes
    an entry in the JSON object. Each value is rendered using ``str``.
    """
    d = {}

    d['formatted'] = self.format(record)

    for k, v in record.__dict__.items():
        try:
            d[k] = str(v)
        except Exception as e:
            d[k] = f'Unrepresentable: {e!r}'

    json.dump(d, fp=self.f)
    print('', file=self.f)
    self.f.flush()

execute_and_log_traceback async

execute_and_log_traceback(fut: Future[Any]) -> Any

Await a future and log any exception..

Warning

For developers. Other functions rely on the first await call to be on the wrapped future to that task cancellation can be properly caught in the wrapped task.

Catches any exceptions raised by the coroutine, logs the traceback, and re-raises the exception.

Source code in academy/logging/helpers.py

async def execute_and_log_traceback(
    fut: Future[Any],
) -> Any:
    """Await a future and log any exception..

    Warning:
        For developers. Other functions rely on the first await call to be on
        the wrapped future to that task cancellation can be properly caught
        in the wrapped task.

    Catches any exceptions raised by the coroutine, logs the traceback,
    and re-raises the exception.
    """
    try:
        return await fut
    except Exception:
        logger.exception('Background task raised an exception.')
        raise

log_context

log_context(
    c: LogConfig | None,
) -> Generator[None, None, None]

Context manager for using an LogConfig.

Source code in academy/logging/helpers.py

@contextlib.contextmanager
def log_context(c: LogConfig | None) -> Generator[None, None, None]:
    """Context manager for using an LogConfig."""
    if c is None:
        # If there's no config, let the context body run without doing
        # anything interesting with logging. This allows log_context
        # to be used to wrap blocks when a user wants to only optionally
        # configure logging.
        yield
        return

    logger.debug(
        f'Entering log_context context manager, with log config {c!r}',
    )

    with log_context_lock:
        if c.uuid in initialized_log_contexts:
            initialized_log_contexts[c.uuid].count += 1
            logger.debug('Configuration already initialized')
        else:
            logger.debug(f'First entry to log config {c} (in this process)')
            initialized_log_contexts[c.uuid] = _ConfigReference(
                c.init_logging(),
            )
            assert callable(initialized_log_contexts[c.uuid].uninit), (
                f'Log config {c} should have returned a callable'
            )

        assert c.uuid in initialized_log_contexts

    yield

    with log_context_lock:
        # Buggy use of reference counting might have removed this from the
        # structure, if unpaired references have been dropped.
        assert c.uuid in initialized_log_contexts

        if initialized_log_contexts[c.uuid].count > 1:
            logger.debug('Configuration is still in use - not uninitializing')
            initialized_log_contexts[c.uuid].count -= 1
        else:
            logger.debug('Configuration has no more users - uninitializing')
            assert initialized_log_contexts[c.uuid].count == 1
            initialized_log_contexts[c.uuid].uninit()
            del initialized_log_contexts[c.uuid]