# Leak Snek Rate Limiting
Leak Snek is a Python library that provides a flexible and extensible implementation of rate limiting for your applications
## Table of Contents
- [Features](#features)
- [Getting Started](#getting-started)
- [License](#license)
## Features
- **Asynchronous and Synchronous Support**: The library provides support for both asynchronous and synchronous rate.
- **Flexible Rate Limiting Algorithms**: Define your custom rate limiting algorithms by implementing the `RateLimiter` and `AsyncRateLimiter` interfaces.
- **Rate Storage**: Manage rate information with the `RateStorage` and `AsyncRateStorage` interfaces.
- **Leaky Bucket Algorithm**: Included implementations of the Leaky Bucket Algorithm for both asynchronous and synchronous use cases.
## Getting Started
1. **Installation**:
Install Leak Snek using pip:
```bash
pip install leak_snek
```
2. **Usage**:
You can use the provided classes and interfaces to implement rate limiting in your Python applications. Additionally, if you need the Leaky Bucket Algorithm, you can use the provided implementations.
- Synchronous Leaky Bucket Algorithm:
```python
from threading import Lock
from leak_snek.limiters.leaky_bucket import LeakyBucketLimiter
from leak_snek.mutexes.memory_mutex import MemoryMutex
from leak_snek.shortcuts.rate_limit import rl
from leak_snek.storages.memory_storage import MemoryStorage
limiter = LeakyBucketLimiter[str](
rate_limit=rl("10/m"),
rate_storage=MemoryStorage(),
key_mutex=MemoryMutex(
local_lock=Lock(),
lock_factory=lambda: Lock(),
),
)
if not limiter.limit_exceeded("my_key"):
... # Perform the operation
```
- Asynchronous Leaky Bucket Algorithm:
```python
from collections.abc import AsyncGenerator, Awaitable
from contextlib import asynccontextmanager
from typing import Any, Self, cast, override
from redis.asyncio import Redis
from leak_snek.interfaces.mutexes.aio.mutex import AsyncMutex
from leak_snek.interfaces.storages.aio.rate_store import AsyncRateStorage
from leak_snek.interfaces.values.rate import Rate
from leak_snek.limiters.aio.leaky_bucket import AsyncLeakyBucketLimiter
from leak_snek.shortcuts.rate_limit import rl
@dataclasses.dataclass
class RedisAsyncRateStorage(AsyncRateStorage[str]):
redis: Redis
async def read(self: Self, key: str) -> Rate:
if not await self.redis.exists(key):
return Rate.default()
rate_dict = await cast(Awaitable[dict[Any, Any]], self.redis.hgetall(name=key))
return Rate(operations=int(rate_dict[b"operations"]), updated_at=float(rate_dict[b"updated_at"]))
async def write(self: Self, key: str, value: Rate) -> None:
await cast(
Awaitable[int],
self.redis.hset(name=key, mapping={"operations": value.operations, "updated_at": value.updated_at}),
)
@dataclasses.dataclass
class RedisAsyncMutex(AsyncMutex[str]):
redis: Redis
@override
@asynccontextmanager
async def lock(self: Self, key: str) -> AsyncGenerator[None, None]:
async with self.redis.lock(f"{key}_lock"):
yield
async def main() -> None:
async_limiter = AsyncLeakyBucketLimiter(
rate_limit=rl("10/m"),
rate_storage=RedisAsyncRateStorage(redis=Redis()),
key_mutex=RedisAsyncMutex(redis=Redis()),
)
if not await async_limiter.limit_exceeded("my_key"):
... # Perform the operation
```
In asynchronous example we implement our own async storage & mutex b/c leak_snek doesn't provide redis integration and it doesn't make much of a sense to implement asynchronous in-memory store.
## License
Leak Snek Rate Limiting is released under the MIT License. See the [LICENSE](LICENSE) file for details.
---
Raw data
{
"_id": null,
"home_page": "https://github.com/WinterCitizen/leak_snek",
"name": "leak-snek",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.12,<4.0",
"maintainer_email": "",
"keywords": "rate,limit,typed,rate_limit,leaky_buket",
"author": "Daniil Fedyaev",
"author_email": "wintercitizen@outlook.com",
"download_url": "https://files.pythonhosted.org/packages/e6/d4/06fc71629d823d0fceb52d10e94ee6e8f70cf85499ba698fb70c9902ff14/leak_snek-0.3.2.tar.gz",
"platform": null,
"description": "# Leak Snek Rate Limiting\n\n\n\n\nLeak Snek is a Python library that provides a flexible and extensible implementation of rate limiting for your applications\n\n## Table of Contents\n\n- [Features](#features)\n- [Getting Started](#getting-started)\n- [License](#license)\n\n## Features\n\n- **Asynchronous and Synchronous Support**: The library provides support for both asynchronous and synchronous rate.\n- **Flexible Rate Limiting Algorithms**: Define your custom rate limiting algorithms by implementing the `RateLimiter` and `AsyncRateLimiter` interfaces.\n- **Rate Storage**: Manage rate information with the `RateStorage` and `AsyncRateStorage` interfaces.\n- **Leaky Bucket Algorithm**: Included implementations of the Leaky Bucket Algorithm for both asynchronous and synchronous use cases.\n\n## Getting Started\n\n1. **Installation**:\n\n Install Leak Snek using pip:\n\n ```bash\n pip install leak_snek\n ```\n\n2. **Usage**:\n\n You can use the provided classes and interfaces to implement rate limiting in your Python applications. Additionally, if you need the Leaky Bucket Algorithm, you can use the provided implementations.\n\n - Synchronous Leaky Bucket Algorithm:\n\n ```python\n from threading import Lock\n\n from leak_snek.limiters.leaky_bucket import LeakyBucketLimiter\n from leak_snek.mutexes.memory_mutex import MemoryMutex\n from leak_snek.shortcuts.rate_limit import rl\n from leak_snek.storages.memory_storage import MemoryStorage\n\n limiter = LeakyBucketLimiter[str](\n rate_limit=rl(\"10/m\"),\n rate_storage=MemoryStorage(),\n key_mutex=MemoryMutex(\n local_lock=Lock(),\n lock_factory=lambda: Lock(),\n ),\n )\n\n if not limiter.limit_exceeded(\"my_key\"):\n ... # Perform the operation\n ```\n\n - Asynchronous Leaky Bucket Algorithm:\n\n ```python\n from collections.abc import AsyncGenerator, Awaitable\n from contextlib import asynccontextmanager\n from typing import Any, Self, cast, override\n\n from redis.asyncio import Redis\n\n from leak_snek.interfaces.mutexes.aio.mutex import AsyncMutex\n from leak_snek.interfaces.storages.aio.rate_store import AsyncRateStorage\n from leak_snek.interfaces.values.rate import Rate\n from leak_snek.limiters.aio.leaky_bucket import AsyncLeakyBucketLimiter\n from leak_snek.shortcuts.rate_limit import rl\n\n\n @dataclasses.dataclass\n class RedisAsyncRateStorage(AsyncRateStorage[str]):\n redis: Redis\n\n async def read(self: Self, key: str) -> Rate:\n if not await self.redis.exists(key):\n return Rate.default()\n\n rate_dict = await cast(Awaitable[dict[Any, Any]], self.redis.hgetall(name=key))\n\n return Rate(operations=int(rate_dict[b\"operations\"]), updated_at=float(rate_dict[b\"updated_at\"]))\n\n async def write(self: Self, key: str, value: Rate) -> None:\n await cast(\n Awaitable[int],\n self.redis.hset(name=key, mapping={\"operations\": value.operations, \"updated_at\": value.updated_at}),\n )\n\n\n @dataclasses.dataclass\n class RedisAsyncMutex(AsyncMutex[str]):\n redis: Redis\n\n @override\n @asynccontextmanager\n async def lock(self: Self, key: str) -> AsyncGenerator[None, None]:\n async with self.redis.lock(f\"{key}_lock\"):\n yield\n\n\n async def main() -> None:\n async_limiter = AsyncLeakyBucketLimiter(\n rate_limit=rl(\"10/m\"),\n rate_storage=RedisAsyncRateStorage(redis=Redis()),\n key_mutex=RedisAsyncMutex(redis=Redis()),\n )\n\n if not await async_limiter.limit_exceeded(\"my_key\"):\n ... # Perform the operation\n ```\n\nIn asynchronous example we implement our own async storage & mutex b/c leak_snek doesn't provide redis integration and it doesn't make much of a sense to implement asynchronous in-memory store.\n\n## License\n\nLeak Snek Rate Limiting is released under the MIT License. See the [LICENSE](LICENSE) file for details.\n\n---\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Leak Snek is a Python library that provides a flexible and extensible implementation of rate limiting for your applications",
"version": "0.3.2",
"project_urls": {
"Homepage": "https://github.com/WinterCitizen/leak_snek",
"Repository": "https://github.com/WinterCitizen/leak_snek"
},
"split_keywords": [
"rate",
"limit",
"typed",
"rate_limit",
"leaky_buket"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b26dd6e6a0a94a687cdb8c223383763849838dcb97c693dc29a6889e180ad828",
"md5": "55b3678b8995611a7bfeae5a9c828484",
"sha256": "f34ca3a35231dfe8cd50b51843de99f53e4a284cd4b606cd9fa33d5fc6a5f238"
},
"downloads": -1,
"filename": "leak_snek-0.3.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "55b3678b8995611a7bfeae5a9c828484",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12,<4.0",
"size": 20114,
"upload_time": "2023-10-25T13:26:01",
"upload_time_iso_8601": "2023-10-25T13:26:01.491460Z",
"url": "https://files.pythonhosted.org/packages/b2/6d/d6e6a0a94a687cdb8c223383763849838dcb97c693dc29a6889e180ad828/leak_snek-0.3.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e6d406fc71629d823d0fceb52d10e94ee6e8f70cf85499ba698fb70c9902ff14",
"md5": "70ac7911122dc1e84c09a606217352e5",
"sha256": "6b4562b4b49dc757f88330cccb317d9206e6f92ade6e5c2bf1ab7af1cf5906ff"
},
"downloads": -1,
"filename": "leak_snek-0.3.2.tar.gz",
"has_sig": false,
"md5_digest": "70ac7911122dc1e84c09a606217352e5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12,<4.0",
"size": 12094,
"upload_time": "2023-10-25T13:26:03",
"upload_time_iso_8601": "2023-10-25T13:26:03.453768Z",
"url": "https://files.pythonhosted.org/packages/e6/d4/06fc71629d823d0fceb52d10e94ee6e8f70cf85499ba698fb70c9902ff14/leak_snek-0.3.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-10-25 13:26:03",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "WinterCitizen",
"github_project": "leak_snek",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "leak-snek"
}
JSON
