plexus.utils.request_log_capture module

Request-scoped log capture utilities for FastAPI applications.

This module provides several approaches to capture logs per request: 1. MemoryHandler - Buffers logs and flushes on completion 2. StringIO handler - Captures logs to a string buffer 3. Context variable approach - Tracks logs per request context

class plexus.utils.request_log_capture.AsyncLogCapture

Bases: object

Async-friendly log capture that works with FastAPI

__init__()

async capture_logs_async(request_id: str, coro)

Capture logs during the execution of an async coroutine

class plexus.utils.request_log_capture.BufferingLogHandler(capacity: int = 50, target_handler: Handler | None = None)

Bases: Handler

Handler that buffers logs in memory and can flush them conditionally. Based on Python’s MemoryHandler but with more control.

Initializes the instance - basically setting the formatter to None and the filter list to empty.

__init__(capacity: int = 50, target_handler: Handler | None = None)

Initializes the instance - basically setting the formatter to None and the filter list to empty.

clear_buffer()

Clear the buffer.

emit(record)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

flush_to_target()

Flush buffered records to target handler.

get_buffered_logs() → List[str]

Get formatted logs from buffer.

class plexus.utils.request_log_capture.MemoryLogHandler(capacity: int = 1000)

Bases: Handler

Simple memory-based log handler that captures all logs to a string buffer. This doesn’t filter by request ID, it just captures everything during its lifetime.

Initializes the instance - basically setting the formatter to None and the filter list to empty.

__init__(capacity: int = 1000)

Initializes the instance - basically setting the formatter to None and the filter list to empty.

close()

Tidy up any resources used by the handler.

This version removes the handler from an internal map of handlers, _handlers, which is used for handler lookup by name. Subclasses should ensure that this gets called from overridden close() methods.

emit(record)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

get_logs() → str

Get all captured logs as a string.

class plexus.utils.request_log_capture.RequestLogCapture(capacity: int = 100)

Bases: object

Captures logs for a specific request using various methods.

__init__(capacity: int = 100)

class plexus.utils.request_log_capture.RequestLogManager(logger_name: str = None, buffer_size: int = 100)

Bases: object

High-level manager for request-scoped log capture. Now uses the simpler approach that doesn’t interfere with existing logging.

__init__(logger_name: str = None, buffer_size: int = 100)

finish_request_capture(request_id: str) → str

Finish capturing logs for a request and clean up.

get_request_logs(request_id: str) → str

Get captured logs for a specific request.

start_request_capture(request_id: str = None) → str

Start capturing logs for a request.

class plexus.utils.request_log_capture.RingBufferLogHandler(capacity: int = 100)

Bases: Handler

Log handler that maintains a ring buffer of recent log messages. Similar to spdlog’s backtrace feature.

Initializes the instance - basically setting the formatter to None and the filter list to empty.

__init__(capacity: int = 100)

Initializes the instance - basically setting the formatter to None and the filter list to empty.

clear()

Clear the ring buffer.

emit(record)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

get_recent_logs() → List[str]

Get all logs from the ring buffer.

class plexus.utils.request_log_capture.SimpleRequestLogCapture

Bases: object

Simple request-scoped log capture that doesn’t interfere with existing logging. Uses a memory handler approach to capture logs without modifying the root logger.

__init__()

finish_request_capture(request_id: str) → str

Finish capturing logs for a request and clean up.

Args:

request_id: The request ID to finish

Returns:

Final captured logs

get_request_logs(request_id: str) → str

Get captured logs for a specific request.

Args:

request_id: The request ID to get logs for

Returns:

Captured logs as a string

start_request_capture(request_id: str = None) → str

Start capturing logs for a request using a memory buffer approach.

Args:

request_id: Optional request ID, generates UUID if not provided

Returns:

The request ID being used

class plexus.utils.request_log_capture.ThreadLocalLogCapture

Bases: object

Thread-safe log capture that uses thread-local storage to avoid conflicts between concurrent requests.

__init__()

get_logs(request_id: str) → str | None

Get captured logs for a request

start_capture(request_id: str, level=20)

Start capturing logs for a specific request

stop_capture(request_id: str)

Stop capturing logs for a request and clean up

plexus.utils.request_log_capture.capture_request_logs(request_id: str | None = None)

Context manager to capture logs for a specific request.

Usage:

with capture_request_logs() as (request_id, get_logs):

# Do work that generates logs logs = get_logs()

plexus.utils.request_log_capture.finish_request_logging(request_id: str) → str

Global function to finish request logging.

plexus.utils.request_log_capture.get_request_logs(request_id: str) → str

Global function to get request logs.

plexus.utils.request_log_capture.setup_request_logging(logger_name: str = None) → Logger

Set up a logger with request-scoped capturing capabilities. Note: This function is deprecated. Use RequestLogManager directly instead.

Args:

logger_name: Name of the logger, defaults to calling module

Returns:

Configured logger instance

plexus.utils.request_log_capture.start_request_logging(request_id: str = None) → str

Global function to start request logging.