| Name | litestar-saq JSON |
| Version |
0.4.0
JSON |
| download |
| home_page | None |
| Summary | Litestar integration for SAQ |
| upload_time | 2025-01-22 05:07:35 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.9 |
| license | MIT |
| keywords |
litestar
saq
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Litestar SAQ
## Installation
```shell
pip install litestar-saq
```
## Usage
Here is a basic application that demonstrates how to use the plugin.
```python
from __future__ import annotations
from litestar import Litestar
from litestar_saq import QueueConfig, SAQConfig, SAQPlugin
saq = SAQPlugin(config=SAQConfig(dsn="redis://localhost:6397/0", queue_configs=[QueueConfig(name="samples")]))
app = Litestar(plugins=[saq])
```
You can start a background worker with the following command now:
```shell
litestar --app-dir=examples/ --app basic:app workers run
Using Litestar app from env: 'basic:app'
Starting SAQ Workers ──────────────────────────────────────────────────────────────────
INFO - 2023-10-04 17:39:03,255 - saq - worker - Worker starting: Queue<redis=Redis<ConnectionPool<Connection<host=localhost,port=6397,db=0>>>, name='samples'>
INFO - 2023-10-04 17:39:06,545 - saq - worker - Worker shutting down
```
You can also start the process for only specific queues. This is helpful if you want separated processes working on different queues instead of combining them.
```shell
litestar --app-dir=examples/ --app basic:app workers run --queues sample
Using Litestar app from env: 'basic:app'
Starting SAQ Workers ──────────────────────────────────────────────────────────────────
INFO - 2023-10-04 17:39:03,255 - saq - worker - Worker starting: Queue<redis=Redis<ConnectionPool<Connection<host=localhost,port=6397,db=0>>>, name='samples'>
INFO - 2023-10-04 17:39:06,545 - saq - worker - Worker shutting down
```
If you are starting the process for only specific queues and still want to read from the other queues or enqueue a task into another queue that was not initialized in your worker or is found somewhere else, you can do so like here
```python
import os
from saq import Queue
def get_queue_directly(queue_name: str, redis_url: str) -> Queue:
return Queue.from_url(redis_url, name=queue_name)
redis_url = os.getenv("REDIS_URL")
queue = get_queue_directly("queue-in-other-process", redis_url)
# Get queue info
info = await queue.info(jobs=True)
# Enqueue new task
queue.enqueue(
....
)
```
Raw data
{
"_id": null,
"home_page": null,
"name": "litestar-saq",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "litestar, saq",
"author": null,
"author_email": "Cody Fincher <cody.fincher@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/ed/31/edb3935fba108af0ec0d759419c70479864f0e19d471c8db7dac7e751e90/litestar_saq-0.4.0.tar.gz",
"platform": null,
"description": "# Litestar SAQ\n\n## Installation\n\n```shell\npip install litestar-saq\n```\n\n## Usage\n\nHere is a basic application that demonstrates how to use the plugin.\n\n```python\nfrom __future__ import annotations\n\nfrom litestar import Litestar\n\nfrom litestar_saq import QueueConfig, SAQConfig, SAQPlugin\n\nsaq = SAQPlugin(config=SAQConfig(dsn=\"redis://localhost:6397/0\", queue_configs=[QueueConfig(name=\"samples\")]))\napp = Litestar(plugins=[saq])\n\n\n```\n\nYou can start a background worker with the following command now:\n\n```shell\nlitestar --app-dir=examples/ --app basic:app workers run\nUsing Litestar app from env: 'basic:app'\nStarting SAQ Workers \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\nINFO - 2023-10-04 17:39:03,255 - saq - worker - Worker starting: Queue<redis=Redis<ConnectionPool<Connection<host=localhost,port=6397,db=0>>>, name='samples'>\nINFO - 2023-10-04 17:39:06,545 - saq - worker - Worker shutting down\n```\n\nYou can also start the process for only specific queues. This is helpful if you want separated processes working on different queues instead of combining them.\n\n```shell\nlitestar --app-dir=examples/ --app basic:app workers run --queues sample\nUsing Litestar app from env: 'basic:app'\nStarting SAQ Workers \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\nINFO - 2023-10-04 17:39:03,255 - saq - worker - Worker starting: Queue<redis=Redis<ConnectionPool<Connection<host=localhost,port=6397,db=0>>>, name='samples'>\nINFO - 2023-10-04 17:39:06,545 - saq - worker - Worker shutting down\n```\n\nIf you are starting the process for only specific queues and still want to read from the other queues or enqueue a task into another queue that was not initialized in your worker or is found somewhere else, you can do so like here\n\n```python\nimport os\nfrom saq import Queue\n\n\ndef get_queue_directly(queue_name: str, redis_url: str) -> Queue:\n return Queue.from_url(redis_url, name=queue_name)\n\nredis_url = os.getenv(\"REDIS_URL\")\nqueue = get_queue_directly(\"queue-in-other-process\", redis_url)\n# Get queue info\ninfo = await queue.info(jobs=True)\n# Enqueue new task\nqueue.enqueue(\n ....\n)\n```\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Litestar integration for SAQ",
"version": "0.4.0",
"project_urls": {
"Changelog": "https://cofin.github.io/litesatr-saq/latest/changelog",
"Discord": "https://discord.gg/X3FJqy8d2j",
"Documentation": "https://cofin.github.io/litesatr-saq/latest/",
"Homepage": "https://cofin.github.io/litesatr-saq/latest/",
"Issue": "https://github.com/cofin/litestar-saq/issues/",
"Source": "https://github.com/cofin/litestar-saq"
},
"split_keywords": [
"litestar",
" saq"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "69b1fb86430a56c32a6bfcf96cf67280c1dcf318486fb39f972f77369d79ab9f",
"md5": "7069bc1dab7ca1c7c08f910ee84263f2",
"sha256": "38a6aede0c3e55e4fab5c175286e5c4f5237a3f2ef5dfd3bc0ab1f080db65c74"
},
"downloads": -1,
"filename": "litestar_saq-0.4.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7069bc1dab7ca1c7c08f910ee84263f2",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 14705,
"upload_time": "2025-01-22T05:07:33",
"upload_time_iso_8601": "2025-01-22T05:07:33.784847Z",
"url": "https://files.pythonhosted.org/packages/69/b1/fb86430a56c32a6bfcf96cf67280c1dcf318486fb39f972f77369d79ab9f/litestar_saq-0.4.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "ed31edb3935fba108af0ec0d759419c70479864f0e19d471c8db7dac7e751e90",
"md5": "00834d4bf310c8848515f29ac44e6b65",
"sha256": "2e49cb2cb6d413f27c1becff1fbe99a35ab986326eba91c1825a3e5cb5abf669"
},
"downloads": -1,
"filename": "litestar_saq-0.4.0.tar.gz",
"has_sig": false,
"md5_digest": "00834d4bf310c8848515f29ac44e6b65",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 125346,
"upload_time": "2025-01-22T05:07:35",
"upload_time_iso_8601": "2025-01-22T05:07:35.500684Z",
"url": "https://files.pythonhosted.org/packages/ed/31/edb3935fba108af0ec0d759419c70479864f0e19d471c8db7dac7e751e90/litestar_saq-0.4.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-22 05:07:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cofin",
"github_project": "litestar-saq",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "litestar-saq"
}
