limits
limits is a python library to perform rate limiting with commonly used storage backends (Redis, Memcached & MongoDB).
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/minutethe strategy will:- Allow 10 requests between
00:01:00and00:02:00 - Allow 10 requests at
00:00:59and 10 more requests at00:01:00
- Allow 10 requests between
- 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
- Allow 9 requests that arrive at
Storage backends
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 leave it to fate
some_storage = storage.storage.from_string(fate)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")