Store Backends

Store Backends#

1) In-Memory#

MemoryStore is essentially a memory-based LRU Cache with expiration time, it is thread-safe and can be used for rate limiting in a single process.

By default, Throttled will initialize a global MemoryStore instance with maximum capacity of 1024, so you don’t usually need to create it manually.

Sync

from throttled import Throttled


# 🌟 Use the global MemoryStore as the storage backend.
@Throttled(key="/api/products", quota="1/m")
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()

Async

import asyncio

from throttled.asyncio import Throttled


# 🌟 Use the global MemoryStore as the storage backend.
@Throttled(key="/api/products", quota="1/m")
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())

Also note that throttled.store.MemoryStore and throttled.asyncio.store.MemoryStore are implemented based on threading.RLock and asyncio.Lock respectively, so the global instance is also independent for synchronous and asynchronous usage.

Different instances mean different storage spaces, if you want to limit the same key in different places in your program, make sure that Throttled receives the same MemoryStore instance and uses the same Quota configuration.

The following example uses MemoryStore as the storage backend and throttles the same Key on ping and pong:

Sync

from throttled import Throttled, store

# 🌟 Use MemoryStore as the storage backend.
mem_store = store.MemoryStore()


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
def ping() -> str:
    return "ping"


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
def pong() -> str:
    return "pong"


def demo() -> None:
    # >> ping
    ping()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    pong()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()

Async

import asyncio

from throttled.asyncio import Throttled, store

# 🌟 Use MemoryStore as the storage backend.
mem_store = store.MemoryStore()


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
async def ping() -> str:
    return "ping"


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
async def pong() -> str:
    return "pong"


async def demo() -> None:
    # >> ping
    await ping()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await pong()


if __name__ == "__main__":
    asyncio.run(demo())

2) Redis#

RedisStore is implemented based on redis-py, you can use it for rate limiting in a distributed environment.

It supports the following arguments:

server: Standard Redis URL.
options: Redis connection configuration, supports all configuration items of redis-py, see RedisStore Options.

The following examples demonstrate how to use RedisStore with different Redis deployment types.

2.1) Standalone#

Three server formats are supported for standalone Redis:

redis:// creates a TCP socket connection. See more at: Redis URI Schemes.
rediss:// creates a SSL wrapped TCP socket connection. See more at: Redis SSL URI Schemes.
unix:// creates a Unix Domain Socket connection.

For example:

redis://[[username]:[password]@]localhost:6379/0
rediss://[[username]:[password]@]localhost:6379/0
unix://[username@]/path/to/socket.sock?db=0[&password=password]

Sync

from throttled import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis://127.0.0.1:6379/0",
        # 🌟 Pass any extra kwargs for redis-py client.
        options={"REDIS_CLIENT_KWARGS": {}, "CONNECTION_POOL_KWARGS": {}},
    ),
)
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()

Async

import asyncio

from throttled.asyncio import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis://127.0.0.1:6379/0",
        # 🌟 Pass any extra kwargs for redis-py client.
        options={"REDIS_CLIENT_KWARGS": {}, "CONNECTION_POOL_KWARGS": {}},
    ),
)
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())

2.2) Sentinel#

It is also easy to use RedisStore with Redis Sentinel.

server format for Redis Sentinel is as follows:

redis+sentinel://[[username]:[password]@]host1[:port1][,hostN][:portN][/service_name]

username: [Optional] Authentication username
password: [Optional] Authentication password
host: [Required] Sentinel node hostname or IP address.
port: [Optional] Sentinel node port, default is 26379.
service_name: [Optional] Name of the master service monitored by Sentinel, default is mymaster.

Sync

from throttled import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+sentinel://:yourpassword@host1:26379,host2:26379/mymaster",
        # 🌟 Pass any extra kwargs for redis-py Sentinel client.
        options={
            "SENTINEL_KWARGS": {},
            "REDIS_CLIENT_KWARGS": {},
            "CONNECTION_POOL_KWARGS": {},
        },
    ),
)
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()

Async

import asyncio

from throttled.asyncio import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+sentinel://:yourpassword@host1:26379,host2:26379/mymaster",
        # 🌟 Pass any extra kwargs for redis-py Sentinel client.
        options={
            "SENTINEL_KWARGS": {},
            "REDIS_CLIENT_KWARGS": {},
            "CONNECTION_POOL_KWARGS": {},
        },
    ),
)
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())

2.3) Cluster#

RedisStore also supports Redis Cluster.

server format for Redis Cluster is as follows:

redis+cluster://[[username]:[password]@]host1[:port1][,hostN][:portN]

Additional options can be passed to the RedisCluster via the options.REDIS_CLIENT_KWARGS parameter.

Sync

from throttled import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+cluster://:yourpassword@host1:6379,host2:6379",
        # 🌟 Pass any extra kwargs for redis-py Cluster client
        options={"REDIS_CLIENT_KWARGS": {}},
    ),
)
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()

Async

import asyncio

from throttled.asyncio import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+cluster://:yourpassword@host1:6379,host2:6379",
        # 🌟 Pass any extra kwargs for redis-py Cluster client
        options={"REDIS_CLIENT_KWARGS": {}},
    ),
)
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())

username: [Optional] Redis ACL username used for authentication.
password: [Optional] Password used for authentication.
host1, …, hostN: [Required] One or more Redis Cluster node hostnames or IP addresses.
port1, …, portN: [Optional] Port for each host; defaults to 6379 when omitted.

3) References#

redis-py Connecting to Redis

Store Backends

Contents

Store Backends#

1) In-Memory#

2) Redis#

2.1) Standalone#

2.2) Sentinel#

2.3) Cluster#

3) References#