Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
from functools import wraps
from threading import Timer, Semaphore
import requests
def service_request_rate_limit(per_minute: int):
assert isinstance(per_minute, int), per_minute
assert per_minute > 0, per_minute
lock = Semaphore(per_minute)
def decorator(fn):
nonlocal lock
def wrapper(*a, **k):
nonlocal lock
output = fn(*a, **k)
Timer(60.0, lock.release).start()
return output
return wrapper
return decorator
gitlab_ratelimit_requests_per_minute = 1000
gitlab_request_rate_limit = service_request_rate_limit(per_minute=gitlab_ratelimit_requests_per_minute)
gitlab_get = gitlab_request_rate_limit(requests.get)
gitlab_put = gitlab_request_rate_limit(requests.put)
gitlab_post = gitlab_request_rate_limit(
gitlab_delete = gitlab_request_rate_limit(requests.delete)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment