# LowMQ Python Client
[](https://pypi.org/project/lowmq-client/)
[](https://pypi.org/project/lowmq-client/)
[](./LICENSE)
An ergonomic, type-annotated, asyncio-based client for LowMQ — a lightweight message queue.
- LowMQ server repo: https://github.com/farawayCC/lowmq
- This package: https://pypi.org/project/lowmq-client/
## Highlights
- Async-first API built on top of aiohttp
- Fully type-annotated, ships py.typed for type checkers
- Safe JSON handling and helpful exceptions
- Pluggable aiohttp session and configurable timeouts
- Tiny footprint, minimal required deps
## Install
Using pip:
```bash
pip install lowmq-client
```
For development (linters, tests, etc.):
```bash
pip install -e .[dev]
pre-commit install
```
Using uv (fast Python package manager):
```bash
# Install uv with pipx (recommended)
pipx install uv
# Sync the project (installs core + dev, as configured)
uv sync
# Activate the virtual environment that uv manages
# Windows PowerShell:
. .venv\Scripts\activate
# Linux/macOS:
# source .venv/bin/activate
pre-commit install
```
## Quickstart
```python
import asyncio
from lowmq_client import LowMqClient
async def main() -> None:
base_url = "https://your-lowmq-server.com"
auth_key = "your-auth-key"
async with LowMqClient(auth_key, base_url) as client:
# Add a message to a queue
add_res = await client.add_packet("payments", {"amount": 100}, freeze_time_min=5)
print("added:", add_res)
# Get a message from a queue (and keep it in the queue)
msg = await client.get_packet("payments", delete=False)
print("fetched:", msg)
# Delete a message by id
if msg and "_id" in msg:
ok = await client.delete_packet("payments", msg["_id"])
print("deleted:", ok)
asyncio.run(main())
```
## API
- LowMqClient(auth_key: str, lowmq_url: str | pydantic.AnyUrl, session: Optional[aiohttp.ClientSession] = None, timeout: Optional[aiohttp.ClientTimeout] = None)
- Asynchronous client. Can re-use an external aiohttp session.
- await set_auth_key(auth_key: str) -> None
- await set_lowmq_url(lowmq_url: str | pydantic.AnyUrl) -> None
- await add_packet(queue_name: str, payload: Any, freeze_time_min: int = 5) -> dict
- POST /msg?freezeTimeMin=...
- await get_packet(queue_name: str, delete: bool = False) -> dict
- GET /msg?key=...&delete=true|false
- await delete_packet(queue_name: str, packet_id: str) -> bool
- DELETE /msg?key=...&_id=...
### Exceptions
- LowMqError — base class for client exceptions
- InvalidUrlError — invalid LowMQ base URL
- ClientClosedError — client used without an active session
- ApiError — server returned non-2xx, includes status, reason, and parsed body when possible
## Typing
This package is fully typed and includes a py.typed marker. You can rely on type
checkers (mypy/pyright) to validate your usage. The public API returns standard
Python types (dict) for server responses; you can define your own Pydantic
models if you prefer stronger typing for message payloads.
## Recipes
- Reusing your aiohttp session:
```python
import aiohttp
from lowmq_client import LowMqClient
session = aiohttp.ClientSession()
client = LowMqClient("key", "https://lowmq.example", session=session)
# ... use client inside an async context as usual ...
```
- Custom timeout:
```python
import aiohttp
from lowmq_client import LowMqClient
client = LowMqClient(
"key",
"https://lowmq.example",
timeout=aiohttp.ClientTimeout(total=10),
)
```
## Development
- Lint and format (Ruff):
```bash
ruff check .
ruff format .
```
- Tests:
```bash
python -m unittest -v
```
- Pre-commit hooks:
```bash
pre-commit install
pre-commit run --all-files
```
## Links
- LowMQ server: https://github.com/farawayCC/lowmq
- PyPI: https://pypi.org/project/lowmq-client/
- Issues: https://github.com/AI-Stratov/lowmq-python-client/issues
## License
MIT
Raw data
{
"_id": null,
"home_page": null,
"name": "lowmq-client",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "lowmq, message-queue, asyncio, aiohttp, client, pydantic",
"author": "AI-Stratov",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/d6/84/063b2c324f5ffb17464b135d44a2bab2c9b4213e7dc6c82ff50a52d0cc55/lowmq_client-1.0.0.tar.gz",
"platform": null,
"description": "# LowMQ Python Client\n\n[](https://pypi.org/project/lowmq-client/)\n[](https://pypi.org/project/lowmq-client/)\n[](./LICENSE)\n\nAn ergonomic, type-annotated, asyncio-based client for LowMQ \u2014 a lightweight message queue.\n\n- LowMQ server repo: https://github.com/farawayCC/lowmq\n- This package: https://pypi.org/project/lowmq-client/\n\n## Highlights\n\n- Async-first API built on top of aiohttp\n- Fully type-annotated, ships py.typed for type checkers\n- Safe JSON handling and helpful exceptions\n- Pluggable aiohttp session and configurable timeouts\n- Tiny footprint, minimal required deps\n\n## Install\n\nUsing pip:\n\n```bash\npip install lowmq-client\n```\n\nFor development (linters, tests, etc.):\n\n```bash\npip install -e .[dev]\npre-commit install\n```\n\nUsing uv (fast Python package manager):\n\n```bash\n# Install uv with pipx (recommended)\npipx install uv\n\n# Sync the project (installs core + dev, as configured)\nuv sync\n\n# Activate the virtual environment that uv manages\n# Windows PowerShell:\n. .venv\\Scripts\\activate\n# Linux/macOS:\n# source .venv/bin/activate\n\npre-commit install\n```\n\n## Quickstart\n\n```python\nimport asyncio\nfrom lowmq_client import LowMqClient\n\nasync def main() -> None:\n base_url = \"https://your-lowmq-server.com\"\n auth_key = \"your-auth-key\"\n\n async with LowMqClient(auth_key, base_url) as client:\n # Add a message to a queue\n add_res = await client.add_packet(\"payments\", {\"amount\": 100}, freeze_time_min=5)\n print(\"added:\", add_res)\n\n # Get a message from a queue (and keep it in the queue)\n msg = await client.get_packet(\"payments\", delete=False)\n print(\"fetched:\", msg)\n\n # Delete a message by id\n if msg and \"_id\" in msg:\n ok = await client.delete_packet(\"payments\", msg[\"_id\"])\n print(\"deleted:\", ok)\n\nasyncio.run(main())\n```\n\n## API\n\n- LowMqClient(auth_key: str, lowmq_url: str | pydantic.AnyUrl, session: Optional[aiohttp.ClientSession] = None, timeout: Optional[aiohttp.ClientTimeout] = None)\n - Asynchronous client. Can re-use an external aiohttp session.\n- await set_auth_key(auth_key: str) -> None\n- await set_lowmq_url(lowmq_url: str | pydantic.AnyUrl) -> None\n- await add_packet(queue_name: str, payload: Any, freeze_time_min: int = 5) -> dict\n - POST /msg?freezeTimeMin=...\n- await get_packet(queue_name: str, delete: bool = False) -> dict\n - GET /msg?key=...&delete=true|false\n- await delete_packet(queue_name: str, packet_id: str) -> bool\n - DELETE /msg?key=...&_id=...\n\n### Exceptions\n\n- LowMqError \u2014 base class for client exceptions\n- InvalidUrlError \u2014 invalid LowMQ base URL\n- ClientClosedError \u2014 client used without an active session\n- ApiError \u2014 server returned non-2xx, includes status, reason, and parsed body when possible\n\n## Typing\n\nThis package is fully typed and includes a py.typed marker. You can rely on type\ncheckers (mypy/pyright) to validate your usage. The public API returns standard\nPython types (dict) for server responses; you can define your own Pydantic\nmodels if you prefer stronger typing for message payloads.\n\n## Recipes\n\n- Reusing your aiohttp session:\n\n```python\nimport aiohttp\nfrom lowmq_client import LowMqClient\n\nsession = aiohttp.ClientSession()\nclient = LowMqClient(\"key\", \"https://lowmq.example\", session=session)\n# ... use client inside an async context as usual ...\n```\n\n- Custom timeout:\n\n```python\nimport aiohttp\nfrom lowmq_client import LowMqClient\n\nclient = LowMqClient(\n \"key\",\n \"https://lowmq.example\",\n timeout=aiohttp.ClientTimeout(total=10),\n)\n```\n\n## Development\n\n- Lint and format (Ruff):\n\n```bash\nruff check .\nruff format .\n```\n\n- Tests:\n\n```bash\npython -m unittest -v\n```\n\n- Pre-commit hooks:\n\n```bash\npre-commit install\npre-commit run --all-files\n```\n\n## Links\n\n- LowMQ server: https://github.com/farawayCC/lowmq\n- PyPI: https://pypi.org/project/lowmq-client/\n- Issues: https://github.com/AI-Stratov/lowmq-python-client/issues\n\n## License\n\nMIT\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python asynchronous client for interacting with LowMQ",
"version": "1.0.0",
"project_urls": {
"Homepage": "https://github.com/AI-Stratov/lowmq-python-client",
"Issues": "https://github.com/AI-Stratov/lowmq-python-client/issues",
"LowMQ Server": "https://github.com/farawayCC/lowmq",
"Repository": "https://github.com/AI-Stratov/lowmq-python-client"
},
"split_keywords": [
"lowmq",
" message-queue",
" asyncio",
" aiohttp",
" client",
" pydantic"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "7ffe6a5ff0c9f5cf6edbf4b65ed0e1c13f5272ce2682430c324f7fd838e45d8e",
"md5": "93d14b672ca6aa344772c70643e16973",
"sha256": "d0e764ef69813d3eda166bfe38871da245cabf718691aa77a10db4a9e8cbe28c"
},
"downloads": -1,
"filename": "lowmq_client-1.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "93d14b672ca6aa344772c70643e16973",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 7340,
"upload_time": "2025-08-26T20:25:51",
"upload_time_iso_8601": "2025-08-26T20:25:51.935004Z",
"url": "https://files.pythonhosted.org/packages/7f/fe/6a5ff0c9f5cf6edbf4b65ed0e1c13f5272ce2682430c324f7fd838e45d8e/lowmq_client-1.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d684063b2c324f5ffb17464b135d44a2bab2c9b4213e7dc6c82ff50a52d0cc55",
"md5": "7cdcaa0f5873b2b6762bcf7ba56d48f7",
"sha256": "384d9384f7188b518aac6c6c72dda8bbbdd26a7c0ab8c286c845036b785bd245"
},
"downloads": -1,
"filename": "lowmq_client-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "7cdcaa0f5873b2b6762bcf7ba56d48f7",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 7638,
"upload_time": "2025-08-26T20:25:53",
"upload_time_iso_8601": "2025-08-26T20:25:53.083903Z",
"url": "https://files.pythonhosted.org/packages/d6/84/063b2c324f5ffb17464b135d44a2bab2c9b4213e7dc6c82ff50a52d0cc55/lowmq_client-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-26 20:25:53",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "AI-Stratov",
"github_project": "lowmq-python-client",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "lowmq-client"
}
JSON
