token-throttler

from token_throttler import TokenBucket, TokenThrottler from token_throttler.storage import RuntimeStorage # Initialize a throttler with a default cost of 1 token per consumption # and in-memory storage. throttler: TokenThrottler = TokenThrottler(cost=1, storage=RuntimeStorage()) # Add a token bucket named 'api_limit' # It can hold 10 tokens and replenishes 10 tokens every 60 seconds (1 token per 6s). throttler.add_bucket( identifier="api_limit", bucket=TokenBucket(replenish_time=60, max_tokens=10) ) def make_api_call(): if throttler.consume("api_limit"): print("API call allowed. Remaining tokens for 'api_limit'.") # Simulate actual API call else: print("API call denied. Rate limit exceeded for 'api_limit'.") for i in range(15): print(f"Attempt {i+1}: ", end="") make_api_call() # Example of using a specific cost for consumption throttler.add_bucket("heavy_operation", TokenBucket(replenish_time=30, max_tokens=5)) if throttler.consume("heavy_operation", cost=2): print("Heavy operation allowed (cost 2 tokens).") else: print("Heavy operation denied (cost 2 tokens).")