# Asyncio Tokenized Lock
[](https://codecov.io/gh/lucascicco/asyncio-tokenized-lock)
[](https://pypi.org/project/asyncio-tokenized-lock/)
Asyncio Tokenized Lock is a Python library that provides a token-based locking mechanism for managing asynchronous locks in an asyncio environment. It introduces two main classes, `LockManager` and `TokenizedLock`, to facilitate the coordination of asynchronous tasks with associated tokens.
## Features
- **Tokenized Locking:** Locks are associated with tokens, allowing for fine-grained control over asynchronous operations.
- **Asynchronous Support:** Built on asyncio, enabling efficient coordination of asynchronous tasks.
- **Context Manager Interface:** `TokenizedLock` can be used as an asynchronous context manager for clean and concise lock management.
- **Timeout Support:** The library supports timeouts for lock acquisition.
## Installation
```shell
pip install asyncio-tokenized-lock
```
## Usage (Examples)
### Basic Usage
```python
from asyncio_tokenized_lock.lock import LockManager, TokenizedLock
manager = LockManager[str]()
lock = manager.register("example_token")
# Acquire and release the lock
async with lock:
# Perform operations while holding the lock
pass
```
### Timeout
#### Context Manager
```python
from asyncio_tokenized_lock.lock import LockManager, TokenizedLock
manager = LockManager[str]()
token = "example_token"
lock = manager.register(token)
lock.ctx_timeout = 1.0 # Set the timeout to 1 second
try:
# Acquire the lock with a timeout using context manager
async with lock:
# Perform operations while holding the lock
pass
except asyncio.TimeoutError:
...
```
#### Inline
```python
try:
await lock.acquire(timeout=1.0)
# Perform operations while holding the lock
except asyncio.TimeoutError:
print("Lock acquisition timed out")
```
### Queue Concurrency
```python
import asyncio
import uuid
import logging
from asyncio_tokenized_lock.lock import LockManager, TokenizedLock
log = logging.getLogger("my-module")
async def consume_queue_safely(concurrency: int = 5, queue_size: int = 100):
manager = LockManager[str]()
queue = asyncio.Queue()
put_tasks = [
asyncio.ensure_future(queue.put(item=uuid.uuid4()))
for _ in range(queue_size)
]
async def safe_consume(queue: asyncio.Queue):
while not queue.empty():
item = await queue.get()
lock = manager.register(item)
if lock.locked:
continue
async with lock:
# Perform operations with the locked item
yield item
# Worker class to consume from the queue safely
@dataclass
class Worker:
id: str
queue: asyncio.Queue
async def consume(self):
while True:
async for item in safe_consume(self.queue):
# Perform operations with the item
log.info(f"[WORKER-{self.id}] Item {item} processed")
break
workers = [Worker(id=str(i), queue=queue) for i in range(concurrency)]
consume_tasks = [asyncio.ensure_future(w.consume()) for w in workers]
# Wait for tasks to complete
await asyncio.wait(put_tasks)
await asyncio.wait(consume_tasks)
```
## Testing
```shell
poetry run pytest
```
Run the provided test suite to ensure the correct behavior of the `LockManager` and `TokenizedLock` classes in different scenarios.
## Contributing
Contributions are welcome! Feel free to open issues, submit pull requests, or suggest improvements.
## License
This library is licensed under the MIT License - see the [LICENSE](./LICENSE) file for details.
Raw data
{
"_id": null,
"home_page": "https://github.com/lucascicco/asyncio-tokenized-lock",
"name": "asyncio-tokenized-lock",
"maintainer": "Lucas Cicco",
"docs_url": null,
"requires_python": ">=3.9,<4.0",
"maintainer_email": "lucasciccomy@gmail.com",
"keywords": "asyncio,lock,token,concurrency",
"author": "Lucas Cicco",
"author_email": "lucasciccomy@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/ec/a6/dff2244b8aa82101480833eaa066420e7ddcd06c70b95bd388541e668799/asyncio_tokenized_lock-0.1.3.tar.gz",
"platform": null,
"description": "# Asyncio Tokenized Lock\n\n[](https://codecov.io/gh/lucascicco/asyncio-tokenized-lock)\n[](https://pypi.org/project/asyncio-tokenized-lock/)\n\n\nAsyncio Tokenized Lock is a Python library that provides a token-based locking mechanism for managing asynchronous locks in an asyncio environment. It introduces two main classes, `LockManager` and `TokenizedLock`, to facilitate the coordination of asynchronous tasks with associated tokens.\n\n## Features\n\n- **Tokenized Locking:** Locks are associated with tokens, allowing for fine-grained control over asynchronous operations.\n- **Asynchronous Support:** Built on asyncio, enabling efficient coordination of asynchronous tasks.\n- **Context Manager Interface:** `TokenizedLock` can be used as an asynchronous context manager for clean and concise lock management.\n- **Timeout Support:** The library supports timeouts for lock acquisition.\n\n## Installation\n\n```shell\npip install asyncio-tokenized-lock\n```\n\n## Usage (Examples)\n\n### Basic Usage\n\n```python\nfrom asyncio_tokenized_lock.lock import LockManager, TokenizedLock\n\nmanager = LockManager[str]()\nlock = manager.register(\"example_token\")\n\n# Acquire and release the lock\nasync with lock:\n # Perform operations while holding the lock\n pass\n```\n\n### Timeout\n\n#### Context Manager\n\n```python\nfrom asyncio_tokenized_lock.lock import LockManager, TokenizedLock\n\nmanager = LockManager[str]()\ntoken = \"example_token\"\nlock = manager.register(token)\nlock.ctx_timeout = 1.0 # Set the timeout to 1 second\ntry:\n # Acquire the lock with a timeout using context manager\n async with lock:\n # Perform operations while holding the lock\n pass\nexcept asyncio.TimeoutError:\n ...\n```\n\n#### Inline\n\n```python\ntry:\n await lock.acquire(timeout=1.0)\n # Perform operations while holding the lock\nexcept asyncio.TimeoutError:\n print(\"Lock acquisition timed out\")\n```\n\n### Queue Concurrency\n\n```python\nimport asyncio\nimport uuid\nimport logging\nfrom asyncio_tokenized_lock.lock import LockManager, TokenizedLock\n\nlog = logging.getLogger(\"my-module\")\n\nasync def consume_queue_safely(concurrency: int = 5, queue_size: int = 100):\n manager = LockManager[str]()\n queue = asyncio.Queue()\n put_tasks = [\n asyncio.ensure_future(queue.put(item=uuid.uuid4()))\n for _ in range(queue_size)\n ]\n\n async def safe_consume(queue: asyncio.Queue):\n while not queue.empty():\n item = await queue.get()\n lock = manager.register(item)\n\n if lock.locked:\n continue\n\n async with lock:\n # Perform operations with the locked item\n yield item\n\n # Worker class to consume from the queue safely\n @dataclass\n class Worker:\n id: str\n queue: asyncio.Queue\n\n async def consume(self):\n while True:\n async for item in safe_consume(self.queue):\n # Perform operations with the item\n log.info(f\"[WORKER-{self.id}] Item {item} processed\")\n break\n\n workers = [Worker(id=str(i), queue=queue) for i in range(concurrency)]\n consume_tasks = [asyncio.ensure_future(w.consume()) for w in workers]\n\n # Wait for tasks to complete\n await asyncio.wait(put_tasks)\n await asyncio.wait(consume_tasks)\n```\n\n## Testing\n\n```shell\npoetry run pytest\n```\n\nRun the provided test suite to ensure the correct behavior of the `LockManager` and `TokenizedLock` classes in different scenarios.\n\n## Contributing\n\nContributions are welcome! Feel free to open issues, submit pull requests, or suggest improvements.\n\n## License\n\nThis library is licensed under the MIT License - see the [LICENSE](./LICENSE) file for details.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A Python library providing token-based locking for efficient coordination of asynchronous tasks in asyncio.",
"version": "0.1.3",
"project_urls": {
"Homepage": "https://github.com/lucascicco/asyncio-tokenized-lock",
"Repository": "https://github.com/lucascicco/asyncio-tokenized-lock"
},
"split_keywords": [
"asyncio",
"lock",
"token",
"concurrency"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a00cd4cb17e3560863d3e1a5abe20b6eef4c2f58b3cc82d73e9737b80f25cd95",
"md5": "10254121a9764383db732876b4e7a869",
"sha256": "a3c36a6a60f73bcc54282c39b44e4a4b12348693e3a03253281bc70362494265"
},
"downloads": -1,
"filename": "asyncio_tokenized_lock-0.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "10254121a9764383db732876b4e7a869",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9,<4.0",
"size": 5461,
"upload_time": "2023-12-04T20:44:51",
"upload_time_iso_8601": "2023-12-04T20:44:51.578352Z",
"url": "https://files.pythonhosted.org/packages/a0/0c/d4cb17e3560863d3e1a5abe20b6eef4c2f58b3cc82d73e9737b80f25cd95/asyncio_tokenized_lock-0.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "eca6dff2244b8aa82101480833eaa066420e7ddcd06c70b95bd388541e668799",
"md5": "88ae8925573fc31804a405428a413c01",
"sha256": "7649cbf304551dafa5069b5a172995901ec439de1479d3842c7a316c87f97b75"
},
"downloads": -1,
"filename": "asyncio_tokenized_lock-0.1.3.tar.gz",
"has_sig": false,
"md5_digest": "88ae8925573fc31804a405428a413c01",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9,<4.0",
"size": 5581,
"upload_time": "2023-12-04T20:44:53",
"upload_time_iso_8601": "2023-12-04T20:44:53.585561Z",
"url": "https://files.pythonhosted.org/packages/ec/a6/dff2244b8aa82101480833eaa066420e7ddcd06c70b95bd388541e668799/asyncio_tokenized_lock-0.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-12-04 20:44:53",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "lucascicco",
"github_project": "asyncio-tokenized-lock",
"travis_ci": false,
"coveralls": true,
"github_actions": true,
"lcname": "asyncio-tokenized-lock"
}
JSON
