Freiner Documentation

Freiner provides utilities to implement rate limiting using various strategies and storage backends such as redis & memcached.

Quickstart

Initialize the storage backend:

from freiner import MemoryStorage()
memory_storage = MemoryStorage()

Initialize a rate limiter with the Moving Window strategy:

from freiner import MovingWindowRateLimiter
moving_window = MovingWindowRateLimiter(memory_storage)

Initialize a rate limit using the Rate Limit String Notation:

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

Initialize a rate limit explicitly using a subclass of freiner.limits.RateLimitItem:

from freiner import RateLimitItemPerSecond, RateLimitPerMinute
one_per_second = RateLimitItemPerSecond(1, 1)
ten_per_five_minutes = RateLimitPerMinute(10, 5)

Test the limits:

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

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

Check specific limits without hitting them:

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

Clear a limit:

assert moving_window.hit(one_per_minute, ns, "foo") == True
assert moving_window.hit(one_per_minute, ns, "foo") == False
moving_window.clear(one_per_minute", ns, "foo")
assert moving_window.hit(one_per_minute, ns, "foo") == True

Development

Since Freiner integrates with various backend storages, local development and running tests can require some setup. These are all scaffolded using docker and docker-compose. Everything should be started and stopped automatically when running tests:

invoke test