Logging

grelmicro.log

Logging.

DuplicateFilter

DuplicateFilter(
    *,
    allowed_repetitions: PositiveInt | None = None,
    cache_size: PositiveInt | None = None,
    key_mode: KeyMode | None = None,
    ttl_seconds: PositiveFloat | None = None,
    key: Callable[[LogRecord], Hashable] | None = None,
    env_prefix: str | None = None,
    read_env: bool = True,
)

Bases: Filter

Drop log records that repeat beyond allowed_repetitions.

Keys are tracked in an LRU cache with at most cache_size entries. Use key_mode to pick the default key, or pass key to override it.

Set ttl_seconds to emit a new burst of allowed_repetitions records every ttl_seconds during a long flood. This gives operators regular reminders that the issue is still active.

key_mode="template" is about 3 times faster than "rendered" because it skips message formatting.

Thread-safe.

State is kept in the current process only. There is no reset method. Create a new filter to clear all counters.

Initialize the filter.

PARAMETER	DESCRIPTION
allowed_repetitions	Maximum number of records per key that pass the filter before subsequent records are dropped. Default: 5. When unset, resolves from the environment variable GREL_DUPLICATE_FILTER_ALLOWED_REPETITIONS if present, otherwise falls back to the DuplicateFilterConfig default. TYPE: PositiveInt | None DEFAULT: None
cache_size	Maximum number of distinct keys tracked. When exceeded, the least-recently-seen key is evicted. Default: 100. TYPE: PositiveInt | None DEFAULT: None
key_mode	Default key strategy: "template" (default) uses str(record.msg); "rendered" uses record.getMessage(). Ignored when key is set. TYPE: KeyMode | None DEFAULT: None
ttl_seconds	Silence window for automatic counter reset. None (default) disables time-based expiry. TYPE: PositiveFloat | None DEFAULT: None
key	Override the default key function. Receives the record and returns any hashable value. TYPE: Callable[[LogRecord], Hashable] | None DEFAULT: None
env_prefix	Override the auto-derived environment variable prefix. Default: GREL_DUPLICATE_FILTER_. TYPE: str | None DEFAULT: None
read_env	Whether to read environment variables. Default: True. TYPE: bool DEFAULT: True

config property

config: DuplicateFilterConfig

Return the config.

from_config classmethod

from_config(
    config: DuplicateFilterConfig,
    *,
    key: Callable[[LogRecord], Hashable] | None = None,
) -> Self

Construct a DuplicateFilter from a pre-built DuplicateFilterConfig.

PARAMETER	DESCRIPTION
config	The pre-built duplicate filter configuration. Use this path when the configuration is assembled at startup from a settings tree. The environment path is bypassed and the config is used as-is. TYPE: DuplicateFilterConfig
key	Override the default key function. Receives the record and returns any hashable value. TYPE: Callable[[LogRecord], Hashable] | None DEFAULT: None

filter

filter(record: LogRecord) -> bool

Return True if the record should pass, False to drop.

DuplicateFilterConfig

Bases: BaseModel

Duplicate Filter Config.

allowed_repetitions class-attribute instance-attribute

allowed_repetitions: PositiveInt = 5

Maximum number of records per key that pass the filter before subsequent records are dropped.

cache_size class-attribute instance-attribute

cache_size: PositiveInt = 100

Maximum number of distinct keys tracked. When exceeded, the least-recently-seen key is evicted.

key_mode class-attribute instance-attribute

key_mode: KeyMode = 'template'

Default key strategy. "template" (default) uses str(record.msg) and is ~3x faster; %-style parameterized calls collapse across argument values. "rendered" uses record.getMessage() to distinguish per-subject.

ttl_seconds class-attribute instance-attribute

ttl_seconds: PositiveFloat | None = None

Silence window for automatic counter reset. If a key has not been seen for ttl_seconds, its counter resets the next time it appears. None (default) disables time-based expiry, so only LRU eviction can reset a counter.

ErrorDict

Bases: TypedDict

Structured error representation.

type instance-attribute

type: str

message instance-attribute

message: str

stack instance-attribute

stack: NotRequired[str]

JSONRecordDict

Bases: TypedDict

Structured JSON log record after serialization.

Describes the shape of each log line after it is serialized to JSON. Internally, formatters use datetime objects for time. The JSON serializer converts them to ISO 8601.

The core fields follow common structured-logging conventions. Extra context fields are added flat at the top level.

Example::

{"time": "2026-03-30T14:00:00+00:00", "level": "INFO",
 "msg": "request handled", "logger": "myapp.api",
 "caller": "handle:45",
 "method": "GET", "status": 200}

time instance-attribute

time: str

level instance-attribute

level: str

msg instance-attribute

msg: str

logger instance-attribute

logger: str

caller instance-attribute

caller: NotRequired[str]

trace_id instance-attribute

trace_id: NotRequired[str]

span_id instance-attribute

span_id: NotRequired[str]

error instance-attribute

error: NotRequired[ErrorDict]

LoggingConfig

Bases: BaseModel

Logging Config.

backend class-attribute instance-attribute

backend: LoggingBackendType = STDLIB

Logging backend implementation.

level class-attribute instance-attribute

level: LoggingLevelType = INFO

Log level threshold.

format class-attribute instance-attribute

format: LoggingFormatType | str = AUTO

Log format. Built-in option or a custom template string.

timezone class-attribute instance-attribute

timezone: LoggingTimeZoneType = LoggingTimeZoneType('UTC')

IANA timezone for timestamps.

json_serializer class-attribute instance-attribute

json_serializer: LoggingSerializerType = STDLIB

JSON serializer used for structured output.

caller_enabled class-attribute instance-attribute

caller_enabled: bool = False

Include caller (function and line) in log records.

otel_enabled class-attribute instance-attribute

otel_enabled: bool = opentelemetry is not None

Extract OpenTelemetry trace context into log records.

LoggingError

Bases: GrelmicroError

Base logging error.

RateLimitFilter

RateLimitFilter(
    *,
    capacity: PositiveInt | None = None,
    refill_rate: PositiveFloat | None = None,
    key_mode: KeyMode | None = None,
    cost: PositiveFloat | None = None,
    key: Callable[[LogRecord], str] | None = None,
    env_prefix: str | None = None,
    read_env: bool = True,
)

Bases: Filter

Rate-limit log records with a token bucket.

Drops records once the per-key bucket is empty. Each key refills at refill_rate tokens per second, capped at capacity.

Pick the default key strategy via key_mode (per-logger, per-level, global, template, rendered) or supply a custom key callable. Tune capacity and refill_rate to match your acceptable burst and sustained log rate.

Thread-safe. State lives in a MemoryTokenBucket.

State is kept in the current process only. It is not shared between processes. Create a new filter to clear all counters, or call reset for a single key.

Example:

import logging

from grelmicro.log import RateLimitFilter

logger = logging.getLogger("grelmicro.ingest")
logger.addFilter(RateLimitFilter(capacity=10, refill_rate=1))

Read more in the Logging docs.

Initialize the filter.

PARAMETER	DESCRIPTION
capacity	Maximum burst size. Default: 5. When unset, resolves from the environment variable GREL_RATE_LIMIT_FILTER_CAPACITY if present, otherwise falls back to the RateLimitFilterConfig default. TYPE: PositiveInt | None DEFAULT: None
refill_rate	Tokens replenished per second. Default: 1. TYPE: PositiveFloat | None DEFAULT: None
key_mode	Default key strategy: "logger" (default), "level", "global", "template" or "rendered". Ignored when key is set. TYPE: KeyMode | None DEFAULT: None
cost	Tokens consumed per record. Default: 1.0. TYPE: PositiveFloat | None DEFAULT: None
key	Override the default key function. It receives the record and returns a string key. Returned keys share the same namespace as the default key function. Add a unique prefix if you need to avoid collisions. TYPE: Callable[[LogRecord], str] | None DEFAULT: None
env_prefix	Override the auto-derived environment variable prefix. Default: GREL_RATE_LIMIT_FILTER_. TYPE: str | None DEFAULT: None
read_env	Whether to read environment variables. Default: True. TYPE: bool DEFAULT: True

config property

config: RateLimitFilterConfig

Return the config.

from_config classmethod

from_config(
    config: RateLimitFilterConfig,
    *,
    key: Callable[[LogRecord], str] | None = None,
) -> Self

Construct a RateLimitFilter from a pre-built RateLimitFilterConfig.

PARAMETER	DESCRIPTION
config	The pre-built rate limit filter configuration. Use this path when the configuration is assembled at startup from a settings tree. The environment path is bypassed and the config is used as-is. TYPE: RateLimitFilterConfig
key	Override the default key function. It receives the record and returns a string key. TYPE: Callable[[LogRecord], str] | None DEFAULT: None

filter

filter(record: LogRecord) -> bool

Return True to keep the record, False to drop it.

reset

reset(key: str = '') -> None

Restore key's bucket to full capacity.

PARAMETER	DESCRIPTION
key	Identifier to reset. Use the same value the key function produces for the records you want to clear. Pass "" for the "global" key mode. TYPE: str DEFAULT: ''

RateLimitFilterConfig

Bases: BaseModel

Rate Limit Filter Config.

capacity class-attribute instance-attribute

capacity: PositiveInt = 5

Maximum burst size: the bucket holds at most this many tokens. A burst of up to capacity records is allowed before any are dropped.

refill_rate class-attribute instance-attribute

refill_rate: PositiveFloat = 1

Tokens (records) replenished per second. Sets the sustained rate after the initial burst is exhausted.

key_mode class-attribute instance-attribute

key_mode: KeyMode = 'logger'

Default key strategy. "logger" (default) buckets per logger name (each logger gets its own burst budget). "level" buckets per log level. "global" uses a single shared bucket for all records. "template" buckets per (logger, level, str(record.msg)): collapses across arg values of the same template. "rendered" buckets per (logger, level, record.getMessage()): distinguishes fully-rendered messages. Ignored when key is set.

cost class-attribute instance-attribute

cost: PositiveFloat = 1.0

Tokens used per record. Increase this to make some filters drain their bucket faster than others. For example, a filter on a verbose-level handler.

configure

configure(
    *,
    backend: LoggingBackendType | None = None,
    level: LoggingLevelType | None = None,
    format: LoggingFormatType | str | None = None,
    timezone: LoggingTimeZoneType | None = None,
    json_serializer: LoggingSerializerType | None = None,
    caller_enabled: bool | None = None,
    otel_enabled: bool | None = None,
    read_env: bool = True,
) -> LoggingConfig

Configure logging with the selected backend.

Two paths:

• Programmatic: pass any of the per-field kwargs. Unset fields resolve from GREL_LOG_* env vars (when read_env=True), then from LoggingConfig defaults.
• Environmental: omit all kwargs. GREL_LOG_* env vars populate every field.

For the declarative path, use configure_with.

PARAMETER	DESCRIPTION
backend	Logging backend (stdlib, loguru, structlog). Default: stdlib. TYPE: LoggingBackendType | None DEFAULT: None
level	Log level threshold. Default: INFO. TYPE: LoggingLevelType | None DEFAULT: None
format	Log format. Default: AUTO. TYPE: LoggingFormatType | str | None DEFAULT: None
timezone	IANA timezone for timestamps. Default: UTC. TYPE: LoggingTimeZoneType | None DEFAULT: None
json_serializer	JSON serializer. Default: stdlib. TYPE: LoggingSerializerType | None DEFAULT: None
caller_enabled	Include caller (function and line) in log records. Default: False. TYPE: bool | None DEFAULT: None
otel_enabled	Extract OpenTelemetry trace context. Default: True if OpenTelemetry is installed, else False. TYPE: bool | None DEFAULT: None
read_env	Whether to read GREL_LOG_* environment variables. Default: True. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
LoggingConfig	The applied LoggingConfig. Snapshot of what was resolved.

RAISES	DESCRIPTION
DependencyNotFoundError	If the selected backend module is not installed.
ValidationError	If configuration is invalid.

configure_with

configure_with(config: LoggingConfig) -> LoggingConfig

Configure logging from a pre-built LoggingConfig.

PARAMETER	DESCRIPTION
config	Pre-built logging configuration. Use this path when the configuration is assembled at startup from a settings tree. The environment path is bypassed and the config is used as-is. TYPE: LoggingConfig

RETURNS	DESCRIPTION
LoggingConfig	The same LoggingConfig, for symmetry with configure.

grelmicro.log.uvicorn

Uvicorn-friendly formatters for dictConfig usage.

UvicornFormatter

UvicornFormatter()

Bases: _UvicornBaseFormatter

Format-aware uvicorn formatter compatible with logging.config.dictConfig.

Reads LOG_FORMAT and produces the matching output (AUTO, JSON, LOGFMT, TEXT, PRETTY). No constructor arguments required.

Initialize with settings from environment variables.

format

format(record: LogRecord) -> str

Format the log record.

caller is always disabled (caller_enabled=False) because uvicorn's caller info points to uvicorn internals, which is not useful. The logger field (e.g., uvicorn.error, uvicorn.access) already identifies the source.

UvicornAccessFormatter

UvicornAccessFormatter()

Bases: UvicornFormatter

Format-aware uvicorn access log formatter.

Parses uvicorn's access log tuple arguments into structured fields (client_addr, method, full_path, http_version, status_code).

format

format(record: LogRecord) -> str

Format access records with split request fields.