limits 3.13.0

Rate limiting utilities

limits is a python library to perform rate limiting with commonly used storage backends (Redis, Memcached, MongoDB & Etcd).

Supported Strategies

Fixed Window

This strategy resets at a fixed interval (start of minute, hour, day etc). For example, given a rate limit of 10/minute the strategy will:

• Allow 10 requests between 00:01:00 and 00:02:00

• Allow 10 requests at 00:00:59 and 10 more requests at 00:01:00

Fixed Window (Elastic)

Identical to Fixed window, except every breach of rate limit results in an extension to the time out. For example a rate limit of 1/minute hit twice within a minute will result in a lock-out for two minutes.

Moving Window

Sliding window strategy enforces a rate limit of N/(m time units) on the last m time units at the second granularity.

For example, with a rate limit of 10/minute:

• Allow 9 requests that arrive at 00:00:59

• Allow another request that arrives at 00:01:00

• Reject the request that arrives at 00:01:01

Storage backends

• Redis

• Memcached

• MongoDB

• Etcd

• In-Memory

Dive right in

Initialize the storage backend

from limits import storage
memory_storage = storage.MemoryStorage()
# or memcached
memcached_storage = storage.MemcachedStorage("memcached://localhost:11211")
# or redis
redis_storage = storage.RedisStorage("redis://localhost:6379")
# or use the factory
storage_uri = "memcached://localhost:11211"
some_storage = storage.storage_from_string(storage_uri)

Initialize a rate limiter with the Moving Window Strategy

from limits import strategies
moving_window = strategies.MovingWindowRateLimiter(memory_storage)

Initialize a rate limit

from limits import parse
one_per_minute = parse("1/minute")

Initialize a rate limit explicitly

from limits import RateLimitItemPerSecond
one_per_second = RateLimitItemPerSecond(1, 1)

Test the limits

assert True == moving_window.hit(one_per_minute, "test_namespace", "foo")
assert False == moving_window.hit(one_per_minute, "test_namespace", "foo")
assert True == moving_window.hit(one_per_minute, "test_namespace", "bar")

assert True == moving_window.hit(one_per_second, "test_namespace", "foo")
assert False == moving_window.hit(one_per_second, "test_namespace", "foo")
time.sleep(1)
assert True == moving_window.hit(one_per_second, "test_namespace", "foo")

Check specific limits without hitting them

assert True == moving_window.hit(one_per_second, "test_namespace", "foo")
while not moving_window.test(one_per_second, "test_namespace", "foo"):
    time.sleep(0.01)
assert True == moving_window.hit(one_per_second, "test_namespace", "foo")

Links

• Documentation

• Changelog