I need to build a distributed rate limiter to ensure we manage QPS to another API across all of our Pods / Backends. I'm using Redis to do this.
I came across this package: https://pypi.org/project/python-redis-rate-limit/ which uses a connection pool, and is implemented as a decorator. Looks simple enough. On top of this, I also want retry logic, so I'm looking at using https://pypi.org/project/backoff/ to do exponential backoff.
You can imagine some code like:
@backoff.on_exception(... redis_rate_limit.TooManyRequests, )
@redis_rate_limit.RateLimit(...redis_pool=redis.ConnectionPool(...))
def call_api(...):
# calls the API.
How do I unittest such a thing? Do I monkeypatch the RateLimit and on_exception functions? If so, then I kinda need to know a bit about their implementations and how they are called, right? Any input on the testing strategy, or even the implementation itself is much appreciated.
Thanks!