Log Rate Limit

Common errors

• Logs are not being rate-limited as expected, even though the same log message appears repeatedly in the code.

  cause The default behavior assigns a unique stream to each *exact* log message. If there are slight variations in the message string due to variables, timestamps, or other dynamic content, the filter treats them as different log streams, and thus does not rate-limit them together.
  fix Identify the static part of your log messages and manually assign a consistent `stream_id` via the `extra=RateLimit()` dictionary for all related log calls. For example, `logger.error("Device %s failed!", device_id, extra=RateLimit(stream_id="device_failure"))`.

• My application logs are being suppressed too aggressively, and important messages are being dropped.

  cause The `StreamRateLimitFilter` parameters (`period_sec`, `max_logs`) might be too restrictive for your logging volume, or the filter is applied to a logger that handles critical messages that should never be rate-limited.
  fix Adjust the `period_sec` (time window) or `max_logs` (number of allowed logs) parameters when initializing `StreamRateLimitFilter`. Alternatively, for individual critical log messages, explicitly disable rate-limiting by setting `stream_id=None` in the `extra` dictionary: `logger.critical("Emergency!", extra=RateLimit(stream_id=None))`.

Warnings

• gotcha The `StreamRateLimitFilter` operates on the raw log message before `logging.Formatter` applies any formatting (like adding timestamps). If you want to rate-limit messages that include dynamic content but are logically 'the same' (e.g., 'Error on device X at Y time'), you must manually define a `stream_id` based on the static parts of the message.
  Fix: For dynamic log messages that should be rate-limited together, pass a custom `stream_id` via the `extra` dictionary: `logger.warning('Error: %s', var, extra=RateLimit(stream_id='my_static_error_type'))`.
• gotcha By default, the filter assigns a unique stream ID to every distinct log message string. This means that if your log messages differ even slightly (e.g., 'Attempt 1 failed' vs. 'Attempt 2 failed'), they will be treated as separate streams and not rate-limited together. Only identical messages will be suppressed.
  Fix: If similar but not identical messages should share a rate limit, provide a consistent `stream_id` for them in the `extra` dictionary: `logger.info('Status update: %s', status, extra=RateLimit(stream_id='status_updates'))`.
• gotcha The library explicitly does not support Python 4.0.0 or later. While Python 4.x is not yet released, this indicates a potential future breaking change if the library is not updated for compatibility.
  Fix: Ensure your project's Python interpreter is within the `<4.0.0,>=3.8.1` range. Monitor future releases of `log-rate-limit` for Python 4.x compatibility updates if planning to upgrade Python versions.

Install

• pip install log-rate-limit Install latest version

Imports

• RateLimit

  from log_rate_limit import RateLimit

  Used to configure rate-limiting per log record via the 'extra' dictionary.
• StreamRateLimitFilter

  from log_rate_limit import StreamRateLimitFilter

  The main logging filter class to apply to a logger.

Quickstart

This quickstart demonstrates how to apply `StreamRateLimitFilter` to a logger, allowing 3 messages per 10 seconds for each unique stream. It shows how the default behavior handles identical messages, how to use a custom `stream_id` to group dynamic messages, and how to explicitly disable rate-limiting for critical logs using `extra=RateLimit(stream_id=None)`.

import logging
import time
from log_rate_limit import RateLimit, StreamRateLimitFilter

# Configure basic logging
logging.basicConfig(level=logging.INFO, format='%(levelname)s:%(name)s:%(message)s')
logger = logging.getLogger(__name__)

# Apply the rate limit filter to the logger
# This example allows 3 logs per 10 seconds for each unique stream_id
logger.addFilter(StreamRateLimitFilter(period_sec=10, max_logs=3))

print("--- Demonstrating default rate limiting (same messages) ---")
for i in range(5):
    logger.info("Repeating message!")
    time.sleep(0.5)

print("\n--- Demonstrating custom stream_id for dynamic messages ---")
# Messages with dynamic content can be grouped by a custom stream_id
for i in range(5):
    device_id = "device_A"
    logger.warning("Error on %s: process %d failed!", device_id, i, extra=RateLimit(stream_id=f"error_on_{device_id}"))
    time.sleep(0.5)

print("\n--- Demonstrating disabling rate limiting for a specific log ---")
logger.error("CRITICAL: Something went terribly wrong, do NOT rate-limit this!", extra=RateLimit(stream_id=None))
logger.info("This info log will be rate-limited by default if repeated.")
logger.info("This info log will be rate-limited by default if repeated.")