CallGate - Awesome Rate Limiter

---

⚠️ IMPORTANT: v2.0.0 Breaking Changes

If you're upgrading from v1.x with Redis storage, you MUST migrate your data.
Redis keys format has changed and old data will be inaccessible without migration.

👉 See Migration Guide for step-by-step instructions.

---

Overview

This project implements a sliding window time-bound rate limiter, which allows tracking events over a configurable time window divided into equal frames. Each frame tracks increments and decrements within a specific time period defined by the frame step.

The CallGate maintains only the values within the set bounds, automatically removing outdated frames as new periods start.

Features

• Thread/Process/Coroutine safe
• Distributable
• Persistable and recoverable
• Easy to use
• Provides various data storage options, including in-memory, shared memory, and Redis
• Includes error handling for common scenarios, with specific exceptions derived from base errors within the library
• A lot of sugar (very sweet):
  • Supports asynchronous and synchronous calls
  • Works as asynchronous and synchronous context manager
  • Works as decorator for functions and coroutines

Installation

You can install CallGate using pip:

pip install call_gate

You may also optionally install redis along with call_gate:

pip install call_gate[redis]

Or you may install them separately:

pip install call_gate
pip install redis  # >=5.0.0

How to?

Create

Use the CallGate class to create a new named rate limiter with gate size and a frame step:

from call_gate import CallGate

gate = CallGate("my_gate", 10, 1)
# what is equivalent to
# gate = CallGate("my_gate", timedelta(seconds=10), timedelta(seconds=1))

This creates a gate with a size of 10 seconds and a frame step of 1 second. Name is mandatory and important: it is used to identify the gate when using shared storage, especially Redis.

Using timedelta allows to set these parameters more precisely and flexible:

from datetime import timedelta

from call_gate import CallGate

gate = CallGate(
    name="my_gate",
    gate_size=timedelta(seconds=1),
    frame_step=timedelta(milliseconds=1)
)

Set Limits

Basically, the gate has two limits:

• gate_limit: how many values can be in the whole gate
• frame_limit: granular limit for each frame in the gate.

Both are set to zero by default. You can keep them zero (what is useless) or reset any of them (or both of them) as follows:

from datetime import timedelta

from call_gate import CallGate

gate = CallGate(
    name="my_gate",
    gate_size=timedelta(seconds=1),
    frame_step=timedelta(milliseconds=1),
    gate_limit=600,
    frame_limit=2
)

What does it mean? This gate has a total scope of 1 second divided by 1 millisecond, what makes this gate rather large: 1000 frames. And the defined limits tell us that within each millisecond we can perform no more than 2 actions.

f the limit is exceeded, we will have to wait until the next millisecond. But the gate limit will reduce us to 600 total actions during 1 second.

You can easily calculate, that during 1 second we shall consume the major limit in the first 300 milliseconds and the rest of the time our code will be waiting until the total gate.sum is reduced.

It will be reduced frame-by-frame. Each time, when the sliding window slides by one frame, a sum is recalculated. Thus, we will do 600 calls more or less quickly and after it we'll start doing slowly and peacefully, frame-by-frame: 2 calls per 1 millisecond + waiting until the gate sum will be lower than 600.

The best practice is to follow the rate-limit documentation of the service which you are using.

For example, at the edge of 2024-2025 Gmail API has the following rate-limits for mail sending via 1 account (mailbox):

• 2 emails per second, but no more than 1200 emails within last 10 minutes;
• 2000 emails within last 24 hours.

This leads us to the following:

gate10m = CallGate(name="gmail10m",
   gate_size=timedelta(minutes=10),
   frame_step=timedelta(seconds=1),
   gate_limit=1200,
   frame_limit=2
)

gate24h = CallGate(name="gmail24h",
   gate_size=timedelta(days=1),
   frame_step=timedelta(minutes=1),
   gate_limit=2000,
)

Both of these windows should be used simultaneously in a sending script on each API call.

While timedelta allows you to set even microseconds, you shall be a realist and remember that Python is not that fast. Some operations may definitely take some microseconds but usually your code needs some milliseconds or longer to switch context, perform a loop, etc. You should also consider network latency if you use remote Redis or make calls to other remote services.

Choose Storage

The library provides three storage options:

• simple: (default) simple storage with a collections.deque;
• shared: shared memory storage using multiprocessing SyncManager list and Value for sum;
• redis: Redis storage (requires redis package and a running Redis-server).

You can specify the storage option when creating the gate either as a string or as one of the GateStorageType keys:

from call_gate import GateStorageType

gate = CallGate(
    "my_gate", 
    timedelta(seconds=10), 
    timedelta(seconds=1), 
    storage=GateStorageType.shared  # <------ or "shared"
)

The simple (default) storage is a thread-safe and pretend to be a process-safe as well. But using it in multiple processes may be un-safe and may result in unexpected behaviour, so don't rely on it in multiprocessing or in WSGI/ASGI workers-forking applications.

The shared storage is a thread-safe and process-safe. You can use it safely in multiple processes and in WSGI/ASGI applications started from one parent process. Note: For Hypercorn >= 0.18.0, you must disable daemon mode (daemon = false in config) to use shared memory storage with multiple workers.

The main disadvantage of these two storages - they are in-memory and do not persist their state between restarts.

The solution is redis storage, which is not just thread-safe and process-safe as well, but also distributable. You can easily use the same gate in multiple processes, even in separated Docker-containers connected to the same Redis-server, Redis-sentinel or Redis-cluster.

Coroutine safety is ensured for all of them by the main class: CallGate.

ASGI Server Compatibility

Uvicorn & Gunicorn: Work out of the box with all storage types.

Hypercorn:

• < 0.18.0: Only redis storage works with multiple workers
• >= 0.18.0: All storage types work when daemon mode is disabled:

  # Using stdin config
  echo 'daemon = false' | hypercorn myapp:app --config /dev/stdin --workers 4
  
  # Using config file
  hypercorn myapp:app --config hypercorn.toml --workers 4

Redis Configuration

Use pre-initialized Redis client:

from redis import Redis

client = Redis(
    host="10.0.0.1",
    port=16379,
    db=0,
    password="secret",
    decode_responses=True,  # Required
    socket_timeout=5,
    socket_connect_timeout=5
)

gate = CallGate(
    "my_gate", 
    timedelta(seconds=10), 
    timedelta(seconds=1), 
    storage=GateStorageType.redis,
    redis_client=client
)

Redis Cluster support:

from redis import RedisCluster
from redis.cluster import ClusterNode

cluster_client = RedisCluster(
    startup_nodes=[
        ClusterNode("node1", 7001),
        ClusterNode("node2", 7002), 
        ClusterNode("node3", 7003)
    ],
    decode_responses=True,  # Required
    skip_full_coverage_check=True,
    socket_timeout=5,
    socket_connect_timeout=5
)

gate = CallGate(
    "my_gate", 
    timedelta(seconds=10), 
    timedelta(seconds=1), 
    storage=GateStorageType.redis,
    redis_client=cluster_client
)

Important notes:

• decode_responses=True is highly recommended for proper operation
• Connection timeouts are recommended to prevent hanging operations
• Redis client validation (ping) is performed during CallGate initialization

---

⚠️ MIGRATION GUIDE v1.x → v2.x

BREAKING CHANGES SUMMARY:

1. Due to Redis Cluster support, Redis keys format changed - old v1.x data is incompatible with v2.x
2. Redis storage requires pre-initialized redis_client parameter (removed **kwargs support)
3. CallGate.from_file() requires redis_client for Redis storage

---

Data Migration for Redis Storage

Redis keys format has changed - old v1.x data will NOT be accessible in v2.x

Step 1: Export data using v1.x and Python REPL

# Using CallGate v1.x
>>> from call_gate import CallGate
>>> redis_kwargs = {"host": "localhost", "port": 6379, "db": 15}
>>> gate = CallGate("my_gate", 60, 1, storage="redis", **redis_kwargs)
>>> gate.to_file("gate_backup.json")

Step 2: Upgrade call-gate version

pip install call-gate --upgrade

Step 3: Import data using v2.x and Python REPL

# Using CallGate v2.x
>>> from call_gate import CallGate
>>> from redis import Redis
>>> redis_kwargs = {"host": "localhost", "port": 6379, "db": 15}
>>> client = Redis(**redis_kwargs, decode_responses=True)
>>> gate = CallGate.from_file("gate_backup.json", storage="redis", redis_client=client) # Data is automatically written to Redis with new key format
>>> gate.state

It's not recommended to insert step 3 into your business logic as it will rewrite your actual data from the file contents on each restart.

Why keys changed:

• v1.x keys: gate_name, gate_name:sum, gate_name:timestamp
• v2.x keys: {gate_name}, {gate_name}:sum, {gate_name}:timestamp
• Hash tags {...} ensure all keys for one gate are in the same Redis Cluster slot

---

Use Directly

Actually, the only method you need is the update method:

# try to increment the current frame value by 1,
# wait while any limit is exceeded
# commit an increment when the "gate is open"
gate.update()

await gate.update(
          5,          # try to increment the current frame value by 5
          throw=True  # throw an error if any limit is exceeded
      )

Use as a Decorator

You can also use the gate as a decorator for functions and coroutines:

@gate(5, throw=True)
def my_function():
    # code here

@gate()
async def my_coroutine():
    # code here

Use as a Context Manager

You can also use the gate as a context manager with functions and coroutines:

def my_function(gate):
    with gate(5, throw=True):
        # code here

async def my_coroutine(gate):
    async with gate():
        # code here

Use Asynchronously

As you could have already understood, CallGate can also be used asynchronously.

There are 3 public methods that can be used interchangeably:

import asyncio

async def main(gate):
    await gate.update()
    await gate.check_limits()
    await gate.clear()

if __name__ == "__main__":
    gate = CallGate("my_async_gate", timedelta(seconds=10), timedelta(seconds=1))
    asyncio.run(main(gate))

Handle Errors

The package provides a pack of custom exceptions. Basically, you may be interested in the following:

• ThrottlingError - a base limit error, raised when rate limits are reached or violated.
• FrameLimitError - (derives from ThrottlingError) a limit error, raised when frame limit is reached or violated.
• GateLimitError - (derives from ThrottlingError) a limit error, raised when gate limit is reached or violated.
• CallGateRedisConfigurationError - raised when Redis client configuration is invalid.

These errors are handled automatically by the library, but you may also choose to throw them explicitly by switching the throw parameter to True

from call_gate import FrameLimitError, GateLimitError, ThrottlingError, CallGateRedisConfigurationError

while True:
    try:
        gate.update(5, throw=True)
    except FrameLimitError as e:
        print(f"Frame limit exceeded! {e}")
    except GateLimitError as e:
        print(f"Gate limit exceeded! {e}")
    except CallGateRedisConfigurationError as e:
        print(f"Redis configuration error! {e}")
        
    # or
    
    # except ThrottlingError as e:
    #    print(f"Throttling Error! {e}")

The others may be found in call_gate.errors module.

Persist and Restore

If you need to persist the state of the gate between restarts, you can use the gate.to_file({file_path}) method.

To restore the state you can use the restored = CallGate.from_file({file_path}) method.

For Redis storage, you must provide redis_client parameter:

from redis import Redis

client = Redis(host="localhost", port=6379, db=15, decode_responses=True)
restored = CallGate.from_file("gate_backup.json", storage="redis", redis_client=client)

If you wish to restore the state using another storage type, you can pass the desired type as a keyword parameter:

restored = CallGate.from_file("gate_backup.json", storage="simple")  # No redis_client needed

Redis persists the gate's state automatically until you restart its container without having shared volumes or clear the Redis database. But still you can save its state to the file and to restore it as well.

You may also use the gate.as_dict() method to get the state of the gate as a dictionary.

Explore the Properties

The CallGate has a lot of useful properties:

gate.name           # get the name of the gate
gate.gate_size      # get the gate size
gate.frame_step     # get the frame step
gate.gate_limit     # get the maximum limit of the gate
gate.frame_limit    # get the maximum limit of the frame
gate.storage        # get the storage type
gate.timezone       # get the gate timezone
gate.frames         # get the number of frames
gate.current_dt     # get the current frame datetime
gate.current_frame  # get the current frame datetime and value
gate.last_frame     # get the last frame datetime and value
gate.limits         # get the gate and frame limits
gate.sum            # get the sum of all values in the gate
gate.data           # get the values of the gate
gate.state          # get the sum and data of the gate atomically

Example

To understand how it works, run this code in your favourite IDE:

import asyncio
from datetime import datetime, timedelta
from call_gate import CallGate

def dummy_func(gate: CallGate):
    requests = 0
    while requests < 30:
        with gate(throw=False):
            requests += 1
            print(f"\r{gate.data = }, {gate.sum = }, {requests = }", end="", flush=True)
    data, sum_ = gate.state
    print(f"\rData: {data}, gate sum: {sum_}, Requests made:, {requests}, {datetime.now()},", flush=True)

async def async_dummy(gate: CallGate):
    requests = 0
    while requests < 30:
        await gate.update()
        requests += 1
        print(f"\r{gate.data = }, {gate.sum = }, {requests = }", end="", flush=True)
    data, sum_ = gate.state
    print(f"\rData: {data}, gate sum: {sum_}, Requests made:, {requests}, {datetime.now()},", flush=True)

if __name__ == "__main__":
    gate = CallGate("my_gate", timedelta(seconds=3), frame_step=timedelta(milliseconds=300), gate_limit=10, frame_limit=2)
    print("Starting sync", datetime.now())
    dummy_func(gate)
    print("Starting async", datetime.now())
    asyncio.run(async_dummy(gate))

More minimal samples live in the examples/ directory.

Remarkable Notes

• The package is compatible with Python 3.9+.
• Under WSGI/ASGI applications I mean the applications such as gunicorn or uvicorn.
• Hypercorn compatibility:
  • Hypercorn < 0.18.0: CallGate cannot be used with multiple workers as Hypercorn spawns each worker as a daemon process, which do not allow child processes.
  • Hypercorn >= 0.18.0: CallGate can be used by disabling daemon mode via configuration file:

    echo 'daemon = false' | hypercorn app:app --config /dev/stdin --workers 4

    Or using a TOML config file:

    # hypercorn.toml
    daemon = false

    hypercorn app:app --config hypercorn.toml --workers 4

  • There are special tests for both cases: "test_hypercorn_server_daemon_behavior" and "test_hypercorn_no_daemon_rate_limit".
• All the updates are atomic, so no race conditions shall occur.
• The majority of Redis calls is performed via Lua-scripts, what makes them run on the Redis-server side.
• Redis Support: CallGate supports Redis standalone, sentinel, and cluster storages.
• Connection Validation: Redis clients are validated with ping() during CallGate initialization to ensure connectivity.
• The maximal value guaranteed for in-memory storages is 2**64 - 1, but for Redis it is 2**53 - 1 only because Redis uses Lua 5.1.
  Lua 5.1 works with numbers as double64 bit floating point numbers in IEEE 754 standard. Starting from 2**53 Lua loses precision.
  But for the purposes of this package even 2**53 - 1 is still big enough.
• If the timezone of your gate is important for any reason, it may be set using the timezone parameter in the CallGate constructor in the string format: "UTC", "Europe/London", "America/New_York", etc. By default, it is None.
• If you need to control the gate's data and sum between the update calls, it's better to use state property instead of calling sum and data. Gate's state collects both values at once. And when you are calling sum and data one-by-one, the frame time may pass and the values may be out of sync.

Testing

The code is covered with 1.5K test cases.

pytest tests/

License

This project is licensed under the MIT License. See the LICENSE file for details.

Contributing

Contributions are welcome! If you have any ideas or bug reports, please open an issue or submit a pull request.