Rate Limiting Utilities

Warnings

• breaking Version 5.0.0 introduced several backward-incompatible changes, including dropping support for the `Fixed Window with Elastic Expiry` strategy and the `etcd` storage backend. Additionally, the default implementation for `async+memcached` was changed from `emcache` to `memcachio`.
  Fix: Review your application's rate limiting strategies and storage configurations. If using `Fixed Window with Elastic Expiry` or `etcd`, migrate to an alternative strategy/storage. If using `async+memcached`, ensure `memcachio` is compatible or manually configure `emcache` if preferred and still supported via an extra.
• gotcha Using `MemoryStorage` (the default in basic examples) in a multi-process or distributed environment will not provide a global rate limit. Each process will have its own independent rate counter, which can lead to limits being exceeded across the entire system.
  Fix: For production or distributed applications, configure a shared storage backend such as Redis, Memcached, or MongoDB. These backends enable centralized rate limit tracking across multiple application instances.
• gotcha When migrating from `limits` versions prior to 5.6.0, note that project metadata moved to `pyproject.toml` and the project now uses `hatch` for package builds. While this primarily affects contributors, it's worth noting for build system interactions.
  Fix: For development or build automation, ensure your tooling is compatible with `pyproject.toml` and `hatch`. Direct users are generally unaffected unless they interact with the build system.
• deprecated Older versions of `limits` (e.g., prior to v2.7.0 and v2.4.0) had specific requirements for `redis` and `coredis` versions. While current versions relax these, always verify compatibility.
  Fix: Always check the `limits` library's `requires` and `Provides-Extra` on PyPI or the `pyproject.toml` for the most up-to-date dependency compatibility, especially when upgrading or installing specific optional backends.

Install

• pip install limits Core library
• pip install 'limits[redis]' # For Redis storage With Redis support
• pip install 'limits[memcached]' # For Memcached storage With Memcached support
• pip install 'limits[mongodb]' # For MongoDB storage With MongoDB support
• pip install 'limits[async-redis]' # For Async Redis storage With Async Redis support
• pip install 'limits[all]' # Installs all optional dependencies All optional dependencies

Imports

• strategies

  from limits import strategies

• storage

  from limits import storage

• RateLimitItem

  from limits import RateLimitItem

• RateLimitExceeded

  from limits.errors import RateLimitExceeded

  `LimitedError` is from `throttled-py` library, `RateLimitExceeded` is for `limits`.
• parse

  from limits import parse

• parse_many

  from limits import parse_many

Quickstart

This quickstart demonstrates how to set up a simple rate limit using in-memory storage. It defines a rate limit, initializes a fixed-window rate limiter, and then simulates multiple actions for a user, demonstrating how the `test` and `hit` methods interact with `RateLimitExceeded` exceptions. It also shows how to get remaining quota and reset time.

from limits import parse, strategies, storage, RateLimitExceeded
import time
import os

# 1. Initialize a storage backend
# For simplicity, using in-memory storage. For production, use Redis/Memcached.
# A Redis example:
# REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379')
# store = storage.RedisStorage(REDIS_URL)

store = storage.MemoryStorage()

# 2. Define a rate limit (e.g., 5 requests per minute)
# parse() converts a string like '5/minute' into a RateLimitItem
rate_limit_string = '5/minute'
rate_limit = parse(rate_limit_string)

# 3. Initialize a rate limiter strategy
# FixedWindowRateLimiter is a common strategy
limiter = strategies.FixedWindowRateLimiter(store)

def do_limited_action(user_id):
    try:
        # 4. Test the limit for a specific identifier (e.g., user_id)
        if limiter.test(rate_limit, user_id):
            # 5. Consume the limit (i.e., record a hit)
            limiter.hit(rate_limit, user_id)
            print(f"[{user_id}] Action performed at {time.strftime('%H:%M:%S')}. Remaining: {limiter.get_remaining(rate_limit, user_id)}")
        else:
            raise RateLimitExceeded(rate_limit)
    except RateLimitExceeded as e:
        # Query available capacity and reset time
        remaining = limiter.get_remaining(e.limit, user_id)
        reset_at = limiter.get_reset_time(e.limit, user_id)
        print(f"[{user_id}] Rate limit exceeded for {e.limit}. Try again in {reset_at - time.time():.1f} seconds. Remaining: {remaining}")

# Simulate some requests
user = "test_user_1"
print(f"--- Simulating requests for {user} ({rate_limit_string}) ---")
for i in range(7):
    do_limited_action(user)
    time.sleep(1) # Simulate some delay

print("\n--- Waiting for reset time ---")
time.sleep(60)

print("\n--- Simulating requests after reset ---")
for i in range(3):
    do_limited_action(user)
    time.sleep(1)