API reference

Everything below is auto-generated from the docstrings in the source. Private members (anything starting with _) are excluded.

Top-level exports

`get_logger`	Return a LogCoreLogger for the given name, creating it if needed.
`Sampler`	Decides whether to emit, drop, or buffer log records.
`LogLevel`
`set_correlation_id`
`get_correlation_id`
`configure_stdlib`	Route the stdlib root logger through LogCore's formatters.
`CorrelationIdMiddleware`	ASGI middleware binding a correlation ID for each request.
`WSGICorrelationIdMiddleware`	WSGI counterpart of `CorrelationIdMiddleware`.
`flush`	Drain pending records on every active queue listener.
`shutdown`	Stop all queue listeners after draining them.

Logger

logcore.get_logger(name, level=None, json=None, file=None, correlation_id=None, max_file_size=None, backup_count=None, redact_fields=None, sampler=None, sample_rate=None, propagate=None, console=None, console_stream=None, async_logging=None, queue_size=None) → LogCoreLogger[source]

Return a LogCoreLogger for the given name, creating it if needed.

Loggers are cached by name. Calling with the same name and no extra arguments returns the existing logger. Passing any configuration argument forces a new logger to be created and cached, replacing the old one. Environment variables (LOGCORE_*) are applied as defaults when a parameter is omitted.

Pass sampler for full control, or sample_rate as a shortcut for Sampler(rate=sample_rate). Passing both raises ValueError.

Parameters:

name (str)
level (Optional[str])
json (Optional[bool])
file (Optional[str])
correlation_id (Optional[str])
max_file_size (Optional[int])
backup_count (Optional[int])
redact_fields (Optional[Set[str]])
sampler (Optional[Sampler])
sample_rate (Optional[float])
propagate (Optional[bool])
console (Optional[bool])
console_stream (Optional[str])
async_logging (Optional[bool])
queue_size (Optional[int])

Return type:

LogCoreLogger

class logcore.logger.LogCoreLogger(config)[source]

Bases: object

Parameters:: config (LogCoreConfig)

debug(message, *args, **kwargs) → None[source]

Parameters:

message (str)
args (Any)
kwargs (Any)

Return type:

None

info(message, *args, **kwargs) → None[source]

Parameters:

message (str)
args (Any)
kwargs (Any)

Return type:

None

warning(message, *args, **kwargs) → None[source]

Parameters:

message (str)
args (Any)
kwargs (Any)

Return type:

None

warn(message, *args, **kwargs) → None[source]

Parameters:

message (str)
args (Any)
kwargs (Any)

Return type:

None

error(message, *args, **kwargs) → None[source]

Parameters:

message (str)
args (Any)
kwargs (Any)

Return type:

None

critical(message, *args, **kwargs) → None[source]

Parameters:

message (str)
args (Any)
kwargs (Any)

Return type:

None

exception(message, *args, **kwargs) → None[source]

Parameters:

message (str)
args (Any)
kwargs (Any)

Return type:

None

flush() → None[source]

Flush all handlers attached to this logger.

Return type:: None

time(operation_name, level='INFO', **kwargs) → Timer | AsyncTimer[source]

Return a context manager that logs start/complete and duration_ms.

Auto-detects async context: returns AsyncTimer inside a running event loop task, Timer otherwise. Use with async with or with accordingly.

Parameters:

operation_name (str)
level (str)
kwargs (Any)

Return type:

Union[Timer, AsyncTimer]

with_correlation_id(correlation_id=None) → Generator[str, None, None][source]

Return a context manager that sets a correlation ID for this scope.

Uses contextvars, so the ID is isolated per async task or thread. A UUID is generated automatically when correlation_id is omitted.

When a tail-based sampler is attached, any buffered records for this correlation_id are discarded on clean exit (the request didn’t error, so we drop the captured history).

Parameters:: correlation_id (Optional[str])
Return type:: Generator[str, None, None]

flush_sample_buffer(correlation_id=None) → int[source]

Discard any tail-buffered records for correlation_id.

Call this at request end when you set the correlation_id directly (e.g. via middleware) instead of using with_correlation_id. When correlation_id is omitted, the current contextvar value is used.

Returns the number of records discarded; 0 if no sampler is attached, sampling is not tail-based, or no buffer exists for the given cid.

Parameters:: correlation_id (Optional[str])
Return type:: int

set_level(level) → None[source]

Parameters:: level (Union[str, LogLevel])
Return type:: None

get_level() → LogLevel[source]

Return type:: LogLevel

is_enabled_for(level) → bool[source]

Parameters:: level (Union[str, LogLevel])
Return type:: bool

Sampling

class logcore.sampling.Sampler(rate=1.0, always_keep=None, tail_based=False, tail_buffer_size=100, max_active_buffers=1000, _rng=None) → None[source]

Bases: object

Decides whether to emit, drop, or buffer log records.

The full evaluation order inside decide():

If record.levelname is in always_keep → KEEP.
If tail_based is on AND a correlation_id is set: when the cid has already been flushed (an earlier error in this request drained the buffer) → KEEP (pass-through mode); otherwise → BUFFER.
Otherwise, a random draw against rate → KEEP or DROP.

The companion method flush_pending() returns any buffered records that should be emitted alongside a kept always_keep record; the caller (LogCoreLogger) is expected to invoke it and emit those records before emitting the triggering record itself.

Parameters:

rate (float) – Fraction of non-always-keep records to emit when rate-based sampling applies. Must be in [0.0, 1.0]. Defaults to 1.0 (keep all).
always_keep (Optional[Set[str]]) – Level names that are never sampled. Defaults to {“WARNING”, “ERROR”, “CRITICAL”}.
tail_based (bool) – When True, buffer records under the current correlation_id and flush on the first always_keep record. Defaults to False.
tail_buffer_size (int) – Max records buffered per correlation_id before the oldest is evicted. Defaults to 100.
_rng (Optional[Random]) – Optional random.Random for deterministic tests.
max_active_buffers (int)

decide_early(levelno, cid) → Decision[source]

Return the sampling decision without needing a built LogRecord.

Called before the record exists so a DROP costs nothing beyond a level comparison and possibly one RNG draw — previously a dropped record still paid for frame inspection, makeRecord and field coercion.

Takes the lock at most once; the counter bumps used to acquire it a second and third time per call.

Parameters:

levelno (int)
cid (Optional[str])

Return type:

Decision

decide(record) → Decision[source]

Return the sampling decision for record.

Retained for callers holding a record; decide_early() is the hot path. Note this does not bump the kept/dropped counters, matching the original contract where the logger bumped them separately.

Parameters:: record (LogRecord)
Return type:: Decision

buffer(record, cid=None) → None[source]

Append record to the ring buffer for cid.

Silently does nothing if no correlation_id is set (defensive — the logger should only call this after a BUFFER decision). Drops the oldest record if the buffer is full, and evicts the least recently used correlation_id once max_active_buffers is reached.

Parameters:

record (LogRecord)
cid (Optional[str])

Return type:

None

flush_pending(record, cid=None) → List[LogRecord][source]

Return buffered records to emit alongside record.

Returns a non-empty list only when tail-based is on, the record’s level is in always_keep, and there are buffered records for the current correlation_id. Marks the cid as flushed so subsequent records pass through directly.

Parameters:

record (LogRecord)
cid (Optional[str])

Return type:

List[LogRecord]

discard_buffer(correlation_id) → int[source]

Drop the buffer for correlation_id without emitting.

Called by LogCoreLogger.with_correlation_id() on clean exit. Returns the number of records discarded. Safe to call when no buffer exists.

Parameters:: correlation_id (str)
Return type:: int

stats() → SamplerStats[source]

Return a snapshot of sampler counters and buffer state.

Return type:: SamplerStats

class logcore.sampling.SamplerStats(active_buffers, buffered_records, dropped_overflow, kept, dropped, buffered, flushed, discarded, evicted_buffers=0) → None[source]

Bases: object

Snapshot of sampler runtime state.

Lifetime counters (since sampler creation):

kept: records emitted directly (passed through sampling).
dropped: records dropped by rate-based sampling.
buffered: records that entered the tail-based buffer (regardless of eventual fate). buffered == flushed + discarded + buffered_records.
flushed: buffered records that were later emitted via an always_keep flush.
discarded: buffered records that were dropped at clean request exit (no error fired).
dropped_overflow: records evicted from a full buffer (ring-buffer overflow before flush/discard).
evicted_buffers: whole correlation_id buffers dropped because max_active_buffers was exceeded. A non-zero value means correlation scopes are not being closed — see flush_sample_buffer.

Live state:

active_buffers: number of correlation_ids currently buffering.
buffered_records: total records currently sitting in buffers.

Parameters:

active_buffers (int)
buffered_records (int)
dropped_overflow (int)
kept (int)
dropped (int)
buffered (int)
flushed (int)
discarded (int)
evicted_buffers (int)

active_buffers: int

buffered_records: int

dropped_overflow: int

kept: int

dropped: int

buffered: int

flushed: int

discarded: int

evicted_buffers: int = 0

class logcore.sampling.Decision(value)[source]

Bases: Enum

The action a Sampler decides to take for a given LogRecord.

KEEP = 'keep'

DROP = 'drop'

BUFFER = 'buffer'

logcore.sampling.sampler_from_env() → Sampler | None[source]

Construct a Sampler from LOGCORE_SAMPLE_* env vars.

Returns None if no sampling env vars are set. Recognises:

LOGCORE_SAMPLE_RATE — float in [0.0, 1.0]
LOGCORE_SAMPLE_TAIL — truthy enables tail-based
LOGCORE_SAMPLE_BUFFER_SIZE — int, max records per correlation_id
LOGCORE_SAMPLE_ALWAYS_KEEP — comma-separated level names

Return type:: Optional[Sampler]

Correlation IDs

logcore.set_correlation_id(correlation_id=None) → str[source]

Parameters:: correlation_id (Optional[str])
Return type:: str

logcore.get_correlation_id() → str | None[source]

Return type:: Optional[str]

logcore.utils.correlation_id_context(correlation_id=None) → Generator[str, None, None][source]

Parameters:: correlation_id (Optional[str])
Return type:: Generator[str, None, None]

Timing helpers

class logcore.utils.Timer(logger, operation_name, level='INFO', **kwargs) → None[source]

Bases: object

Parameters:

logger (Any)
operation_name (str)
level (str)
kwargs (Any)

property elapsed: float | None

class logcore.utils.AsyncTimer(logger, operation_name, level='INFO', **kwargs) → None[source]

Bases: object

Parameters:

logger (Any)
operation_name (str)
level (str)
kwargs (Any)

property elapsed: float | None

Configuration

class logcore.config.LogCoreConfig(name, level=LogLevel.INFO, json=False, file=None, correlation_id=None, max_file_size=10485760, backup_count=5, redact_fields=None, sampler=None, propagate=False, console=True, console_stream='stderr', async_logging=False, queue_size=10000) → None[source]

Bases: object

Parameters:

name (str)
level (LogLevel)
json (bool)
file (Optional[str])
correlation_id (Optional[str])
max_file_size (int)
backup_count (int)
redact_fields (Optional[Set[str]])
sampler (Optional[Sampler])
propagate (bool)
console (bool)
console_stream (str)
async_logging (bool)
queue_size (int)

name: str

level: LogLevel = 'INFO'

json: bool = False

file: str | None = None

correlation_id: str | None = None

max_file_size: int = 10485760

backup_count: int = 5

redact_fields: Set[str] | None = None

sampler: Sampler | None = None

propagate: bool = False

console: bool = True

console_stream: str = 'stderr'

async_logging: bool = False

queue_size: int = 10000

class logcore.LogLevel(value)[source]

Bases: Enum

DEBUG = 'DEBUG'

INFO = 'INFO'

WARNING = 'WARNING'

ERROR = 'ERROR'

CRITICAL = 'CRITICAL'

classmethod from_string(level) → LogLevel[source]

Parameters:: level (str)
Return type:: LogLevel

logcore.config.create_config(name, level=None, json=None, file=None, correlation_id=None, max_file_size=None, backup_count=None, redact_fields=None, sampler=None, sample_rate=None, propagate=None, console=None, console_stream=None, async_logging=None, queue_size=None) → LogCoreConfig[source]

Parameters:

name (str)
level (Optional[str])
json (Optional[bool])
file (Optional[str])
correlation_id (Optional[str])
max_file_size (Optional[int])
backup_count (Optional[int])
redact_fields (Optional[Set[str]])
sampler (Optional[Sampler])
sample_rate (Optional[float])
propagate (Optional[bool])
console (Optional[bool])
console_stream (Optional[str])
async_logging (Optional[bool])
queue_size (Optional[int])

Return type:

LogCoreConfig

Formatters

class logcore.formatters.JSONFormatter(redact_fields=None)[source]

Bases: RedactingFormatter, Formatter

JSON formatter for structured logging.

Parameters:: redact_fields (Optional[Set[str]])

format(record) → str[source]

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

Parameters:: record (LogRecord)
Return type:: str

class logcore.formatters.TextFormatter(redact_fields=None, use_colors=None, stream=None)[source]

Bases: RedactingFormatter, Formatter

Text formatter with colors.

Parameters:

redact_fields (Optional[Set[str]])
use_colors (Optional[bool])
stream (Optional[Any])

COLORS = {'CRITICAL': '', 'DEBUG': '', 'ERROR': '', 'INFO': '', 'WARNING': ''}

format(record) → str[source]

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

Parameters:: record (LogRecord)
Return type:: str

class logcore.formatters.RedactingFormatter(redact_fields=None)[source]

Bases: object

Base formatter with partial masking of sensitive fields.

Redaction is structural: the walk descends into dicts, lists and tuples of the log record’s own fields, so a secret nested inside a payload is masked just like a top-level one. A regex pass over the rendered message catches secrets embedded in message text, which no structural walk can see.

Parameters:: redact_fields (Optional[Set[str]])

Handlers

class logcore.handlers.ConsoleHandler(config)[source]

Bases: object

Parameters:: config (LogCoreConfig)

get_handler() → Handler[source]

Return type:: Handler

class logcore.handlers.FileHandler(config, file_path)[source]

Bases: object

Parameters:

config (LogCoreConfig)
file_path (str)

get_handler() → Handler[source]

Return type:: Handler

logcore.handlers.create_handlers(config) → List[Handler][source]

Build the handler list for config.

With async_logging enabled the real handlers are moved onto a background listener thread and the logger gets a single QueueHandler, so log calls no longer block the caller on I/O.

Parameters:: config (LogCoreConfig)
Return type:: List[Handler]

Lifecycle

Only relevant when async_logging=True moves handler I/O onto a background thread. shutdown is registered with atexit; call it explicitly before a hard exit that skips atexit handlers.

logcore.handlers.flush(timeout=5.0) → None[source]

Drain pending records on every active queue listener.

Waits up to timeout seconds per listener for its queue to empty, then flushes the underlying handlers. Bounded so a wedged handler cannot hang process shutdown.

Parameters:: timeout (float)
Return type:: None

logcore.handlers.shutdown() → None[source]

Stop all queue listeners after draining them.

Registered with atexit, and safe to call explicitly before a hard exit (os._exit, a container SIGKILL grace period) where atexit will not run.

Return type:: None

logcore.handlers.dropped_record_count() → int[source]

Total records shed by full async queues since process start.

Return type:: int

Standard library interop

logcore.interop.configure_stdlib(level='INFO', json=True, file=None, replace_existing=True, redact_fields=None, quiet=None, async_logging=False) → Logger[source]

Route the stdlib root logger through LogCore’s formatters.

Call this once during application startup, before the libraries you want to capture emit anything.

Parameters:

level (Union[str, LogLevel]) – Root log level.
json (bool) – Emit JSON (the usual choice when this is worth doing) or human-readable text.
file (Optional[str]) – Optional path to also write a rotating log file.
replace_existing (bool) – Remove handlers already on the root logger. Leave this True to override an earlier logging.basicConfig(), which would otherwise double every line.
redact_fields (Optional[Set[str]]) – Field names to mask. Defaults to LogCore’s built-in set.
quiet (Optional[Iterable[Union[str, Tuple[str, str]]]]) – Loggers to turn down. Either a name (raised to WARNING) or a (name, level) pair.
async_logging (bool) – Move handler I/O onto a background thread.

Return type:

Logger

Returns:

The configured root logger.

Example

>>> import logcore
>>> logcore.configure_stdlib(level="INFO", json=True,
...                          quiet=["urllib3", ("botocore", "ERROR")])

logcore.interop.reset_stdlib() → None[source]

Remove the handlers configure_stdlib() installed on the root logger.

Intended for tests and for applications that reconfigure logging at runtime. Handlers installed by anything else are left alone.

Return type:: None

logcore.interop.dict_config_formatter(json=True, redact_fields=None) → Dict[str, object][source]

Return a logging.config.dictConfig formatter entry for LogCore.

For applications that configure logging declaratively rather than by calling configure_stdlib().

Example

>>> import logging.config, logcore
>>> logging.config.dictConfig({
...     "version": 1,
...     "formatters": {"logcore": logcore.dict_config_formatter()},
...     "handlers": {"console": {"class": "logging.StreamHandler",
...                              "formatter": "logcore"}},
...     "root": {"handlers": ["console"], "level": "INFO"},
... })

Parameters:

json (bool)
redact_fields (Optional[Set[str]])

Return type:

Dict[str, object]

Web middleware

class logcore.middleware.CorrelationIdMiddleware(app, header='X-Request-ID', logger=None, echo=True) → None[source]

Bases: object

ASGI middleware binding a correlation ID for each request.

Works with any ASGI framework (FastAPI, Starlette, Litestar, Quart):

app.add_middleware(CorrelationIdMiddleware)

or by wrapping directly:

app = CorrelationIdMiddleware(app)

Parameters:

app (Callable[..., Awaitable[None]]) – The ASGI application to wrap.
header (str) – Request/response header carrying the ID.
logger (Optional[LogCoreLogger]) – Optional LogCore logger whose tail-sampling buffer should be released when the request ends.
echo (bool) – Whether to add the ID to the response headers.

class logcore.middleware.WSGICorrelationIdMiddleware(app, header='X-Request-ID', logger=None, echo=True) → None[source]

Bases: object

WSGI counterpart of CorrelationIdMiddleware.

Works with Flask, Django and any other WSGI application:

app.wsgi_app = WSGICorrelationIdMiddleware(app.wsgi_app)

Parameters:

app (Callable[..., Iterable[bytes]])
header (str)
logger (Optional[LogCoreLogger])
echo (bool)