Litestar SAQ
Installation
pip install litestar-saq
For OpenTelemetry support:
pip install litestar-saq[otel]
Usage
Here is a basic application that demonstrates how to use the plugin.
from litestar import Litestar
from litestar_saq import QueueConfig, SAQConfig, SAQPlugin
saq = SAQPlugin(
config=SAQConfig(
use_server_lifespan=True,
queue_configs=[
QueueConfig(name="samples", dsn="redis://localhost:6379/0")
],
)
)
app = Litestar(plugins=[saq])
You can start a background worker with the following command now:
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=6379,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.
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=6379,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
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
await queue.enqueue("task_name", arg1="value1")
Monitored Jobs
For long-running tasks, use the monitored_job decorator to automatically send heartbeats and prevent SAQ from marking jobs as stuck:
from litestar_saq import monitored_job
@monitored_job() # Auto-calculates interval from job.heartbeat
async def long_running_task(ctx):
await process_large_dataset()
return {"status": "complete"}
@monitored_job(interval=30.0) # Explicit 30-second interval
async def train_model(ctx, model_id: str):
for epoch in range(100):
await train_epoch(model)
return {"model_id": model_id}
OpenTelemetry Integration
litestar-saq supports optional OpenTelemetry instrumentation for distributed tracing.
Configuration
from litestar_saq import SAQConfig, QueueConfig
config = SAQConfig(
queue_configs=[QueueConfig(dsn="redis://localhost:6379/0")],
enable_otel=None, # Auto-detect (default) - enabled if OTEL installed AND Litestar OpenTelemetryPlugin is active
# enable_otel=True, # Force enable (raises error if not installed)
# enable_otel=False, # Force disable
)
When enabled, the plugin creates:
- CONSUMER spans for job processing
- PRODUCER spans for job enqueue operations
- Automatic context propagation across process boundaries
Spans follow OpenTelemetry messaging semantic conventions with messaging.system = "saq".
2 maintainers
Metadata
2 maintainers
Release files for litestar-saq 0.8.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| litestar_saq-0.8.0.tar.gz | 286.6 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| litestar_saq-0.8.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 326.4 kB
Release files / litestar_saq-0.8.0.tar.gz
| Download URL | litestar_saq-0.8.0.tar.gz |
|---|---|
| Size | 286.6 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
4d6b6bd6e8c76a6dbf8537f51f71180a78ff7ba4a09294fbb62965653bb04a31
|
|
BLAKE2b-256 checksum How to use checksums |
be80054b7b73f384c0a5ea555130e0f18233e98d3341a7b8fadcf42f1968b75f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/6.1.0 CPython/3.13.12
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Jun 3, 2026.
Transparency logRelease files / litestar_saq-0.8.0-py3-none-any.whl
| Download URL | litestar_saq-0.8.0-py3-none-any.whl |
|---|---|
| Size | 39.8 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
0a47fd054aef997c9870f821638af4dfb2804bfb4652549c96689d81d8f99a92
|
|
BLAKE2b-256 checksum How to use checksums |
4ae7205c29fa6c60fb6d841f8cb3880147963a4f26efa6f3f4815edb189082f8
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/6.1.0 CPython/3.13.12
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Jun 3, 2026.
Transparency log
