Metadata-Version: 2.4
Name: litestar-saq
Version: 0.8.0
Summary: Litestar integration for SAQ
Project-URL: Changelog, https://cofin.github.io/litesatr-saq/latest/changelog
Project-URL: Discord, https://discord.gg/X3FJqy8d2j
Project-URL: Documentation, https://cofin.github.io/litesatr-saq/latest/
Project-URL: Homepage, https://cofin.github.io/litesatr-saq/latest/
Project-URL: Issue, https://github.com/cofin/litestar-saq/issues/
Project-URL: Source, https://github.com/cofin/litestar-saq
Author-email: Cody Fincher <cody.fincher@gmail.com>
License: MIT
License-File: LICENSE
Keywords: litestar,saq
Classifier: Development Status :: 3 - Alpha
Classifier: Environment :: Web Environment
Classifier: License :: OSI Approved :: MIT License
Classifier: Natural Language :: English
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: litestar>=2.0.1
Requires-Dist: saq>=0.26.0
Provides-Extra: hiredis
Requires-Dist: hiredis; extra == 'hiredis'
Provides-Extra: otel
Requires-Dist: opentelemetry-api; extra == 'otel'
Requires-Dist: opentelemetry-sdk; extra == 'otel'
Provides-Extra: psycopg
Requires-Dist: psycopg[binary,pool]; extra == 'psycopg'
Description-Content-Type: text/markdown

# Litestar SAQ

## Installation

```shell
pip install litestar-saq
```

For OpenTelemetry support:

```shell
pip install litestar-saq[otel]
```

## Usage

Here is a basic application that demonstrates how to use the plugin.

```python
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:

```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=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.

```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=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

```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
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:

```python
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

```python
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](https://opentelemetry.io/docs/specs/semconv/messaging/) with `messaging.system = "saq"`.
