How to implement a rate-limited shared counter in Python

Implements a thread-safe global counter that allows a maximum number of increments per second using a lock and time-based refill.

Medium Python 3.9+ Aug 9, 2026 Reliability & rate limiting 12 views 0 copies

rate-limiting threading global-counter lock

Python code

25 lines

Python 3.9+

import threading
import time
import random

counter = 0
lock = threading.Lock()
MAX_CALLS_PER_SECOND = 3
last_refill = time.time()

def rate_limited_increment():
    global counter, last_refill
    with lock:
        now = time.time()
        if now - last_refill >= 1.0:
            last_refill = now
            counter = 0
        if counter < MAX_CALLS_PER_SECOND:
            counter += 1
            return f"Allowed (call #{counter}/{MAX_CALLS_PER_SECOND})"
        return "Rate limited, try again later"

if __name__ == "__main__":
    for i in range(8):
        print(f"Call {i+1}: {rate_limited_increment()}")
        time.sleep(random.uniform(0.2, 0.5))

Output

stdout

Call 1: Allowed (call #1/3)
Call 2: Allowed (call #2/3)
Call 3: Allowed (call #3/3)
Call 4: Rate limited, try again later
Call 5: Rate limited, try again later
Call 6: Rate limited, try again later
Call 7: Allowed (call #1/3)
Call 8: Allowed (call #2/3)

How it works

The code uses a global counter and a lock to ensure thread safety. Every call acquires the lock, checks if a second has passed since the last refill, and resets the counter if so. It then checks whether the counter is below the maximum allowed calls per second and increments if allowed. Using time.time() provides a simple sliding window based on elapsed seconds. The random sleep simulates realistic call intervals to demonstrate rate limiting behavior.

Common mistakes

Forgetting to acquire the lock around counter and last_refill updates, causing race conditions.
Resetting the counter based on wall-clock time at absolute boundaries instead of elapsed time, leading to burst issues.
Using `time.sleep` too long or too short in tests, making output unpredictable.
Not using `global` declarations inside nested functions, causing UnboundLocalError.

Variations

Use `threading.RLock` if reentrant locking is needed.
Use a token bucket algorithm with a fixed refill rate for smoother rate limiting.

Real-world use cases

Throttling API requests from a shared pool of workers to respect external service limits.
Limiting login attempts per user per minute in a web application to prevent brute force.
Enforcing a maximum number of database writes per second from a background job.

How to implement a rate-limited shared counter in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Reliability & rate limiting

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Reliability & rate limiting

Keep learning

Tutorials

Quizzes