Metadata-Version: 2.4
Name: mindtrace-jobs
Version: 0.16.0
Summary: Job management for Mindtrace
Author: Mindtrace Team
License-Expression: Apache-2.0
Project-URL: Homepage, https://mindtrace.ai
Project-URL: Repository, https://github.com/mindtrace/mindtrace/blob/main/mindtrace/jobs
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: mindtrace-core>=0.16.0
Requires-Dist: mindtrace-registry>=0.16.0
Requires-Dist: pika>=1.3.2
Requires-Dist: pydantic>=2.11.1
Requires-Dist: redis>=5.3.0
Dynamic: license-file

[![PyPI version](https://img.shields.io/pypi/v/mindtrace-jobs)](https://pypi.org/project/mindtrace-jobs/)
[![License](https://img.shields.io/pypi/l/mindtrace-jobs)](https://github.com/mindtrace/mindtrace/blob/main/mindtrace/jobs/LICENSE)
[![Downloads](https://static.pepy.tech/badge/mindtrace-jobs)](https://pepy.tech/projects/mindtrace-jobs)

# Mindtrace Jobs

The `Jobs` module provides Mindtrace’s backend-agnostic job queue system for publishing typed jobs, consuming them with Python workers, and switching between local, Redis, and RabbitMQ backends with minimal application changes.

## Features

- **Typed job definitions** with `JobSchema` and Pydantic models
- **Backend-agnostic orchestration** through `Orchestrator`
- **Consumer workers** built by subclassing `Consumer`
- **Multiple backends** for local, Redis, and RabbitMQ execution
- **Queue variants** including FIFO, stack, and priority queues
- **Convenient job creation** with `job_from_schema()`

## Installation

```bash
pip install mindtrace-jobs
```

## Quick Start

```python
from pydantic import BaseModel

from mindtrace.jobs import Consumer, JobSchema, LocalClient, Orchestrator, job_from_schema


class MathsInput(BaseModel):
    operation: str = "add"
    a: float = 2.0
    b: float = 1.0


class MathsOutput(BaseModel):
    result: float = 0.0
    operation_performed: str = ""


schema = JobSchema(
    name="maths_operations",
    input_schema=MathsInput,
    output_schema=MathsOutput,
)

orchestrator = Orchestrator(LocalClient())
orchestrator.register(schema)


class MathsConsumer(Consumer):
    def run(self, job_dict: dict) -> dict:
        payload = job_dict.get("payload", {})
        operation = payload.get("operation", "add")
        a = payload.get("a")
        b = payload.get("b")

        if operation == "add":
            result = a + b
        elif operation == "multiply":
            result = a * b
        else:
            raise ValueError(f"Unknown operation: {operation}")

        return {
            "result": result,
            "operation_performed": f"{operation}({a}, {b}) = {result}",
        }


consumer = MathsConsumer()
consumer.connect_to_orchestrator(orchestrator, "maths_operations")

job = job_from_schema(schema, MathsInput(operation="multiply", a=7.0, b=3.0))
orchestrator.publish("maths_operations", job)
consumer.consume(num_messages=1)
```

## Batch publishing

Use `publish_batch()` when several jobs should be sent to the same queue with
the same backend options:

```python
jobs = [
    job_from_schema(schema, MathsInput(operation="add", a=1, b=2)),
    job_from_schema(schema, MathsInput(operation="multiply", a=3, b=4)),
]

result = orchestrator.publish_batch("maths_operations", jobs)

print(result.job_ids)
print(result.successful_indices)
print(result.failed_indices)
print(result.unattempted_indices)
```

The complete input batch is validated before it reaches the backend. Batch
publication itself is not atomic: if a backend fails while publishing an item,
earlier messages may already be queued. `BatchPublishResult` preserves their
job IDs, records the failed item, and identifies later items that were not
attempted.

RabbitMQ publishes a non-empty batch through one connection and channel. Local,
Redis, and custom backends use the backend-neutral fallback unless they provide
their own optimized `publish_batch()` implementation. RabbitMQ publisher confirms
are not enabled, so returned job IDs represent calls that completed without a
synchronous publish error rather than broker-confirmed acceptance. A later
asynchronous broker rejection may not map precisely to the reported input index.
The returned IDs are backend-generated publication IDs and are not guaranteed to
match the `Job.id` field in the message body.

In practice, the jobs package is built around four concepts:

- a **schema** describing the job payload
- an **orchestrator** that owns queues and publishing
- a **backend** that stores/transports messages
- a **consumer** that processes jobs from one or more queues

## JobSchema and Job

`JobSchema` is currently an alias of `TaskSchema` from `mindtrace-core`. It gives a queue/job type a name plus typed input/output models.

```python
from pydantic import BaseModel

from mindtrace.jobs import JobSchema


class ReportInput(BaseModel):
    report_id: str
    include_charts: bool = True


class ReportOutput(BaseModel):
    path: str


schema = JobSchema(
    name="build_report",
    input_schema=ReportInput,
    output_schema=ReportOutput,
)
```

A `Job` is the executable instance that gets queued. In most cases you do not construct it by hand; you use `job_from_schema()`.

```python
from mindtrace.jobs import job_from_schema


job = job_from_schema(schema, {"report_id": "rpt-123", "include_charts": True})
print(job.id)
print(job.schema_name)
```

## Orchestrator

`Orchestrator` is the publishing and queue-management layer. It owns a backend and handles things like:

- registering schemas
- declaring queues
- publishing jobs
- counting queue messages
- cleaning or deleting queues

```python
from mindtrace.jobs import LocalClient, Orchestrator


orchestrator = Orchestrator(LocalClient())
queue_name = orchestrator.register(schema)
print(queue_name)
```

`clean_queue()` discards the jobs a queue holds and keeps the declaration.
`delete_queue()` discards them and removes the declaration as well, so a queue
declared again under the same name starts empty rather than serving jobs
published before the deletion. Both discard jobs permanently; drain a queue with
`consume_until_empty()` first if the work still matters.

### Publishing typed input directly

If a schema has been registered for a queue, you can publish either:

- a full `Job`
- or a matching Pydantic input model

```python
orchestrator.publish("build_report", ReportInput(report_id="rpt-001"))
```

That convenience is often nicer than manually creating the `Job` every time.

## Consumer

Subclass `Consumer` and implement `run(job_dict: dict) -> dict`.

```python
from mindtrace.jobs import Consumer


class ReportConsumer(Consumer):
    def run(self, job_dict: dict) -> dict:
        payload = job_dict.get("payload", {})
        report_id = payload.get("report_id")
        return {"path": f"/tmp/{report_id}.pdf"}
```

Then connect the consumer to an orchestrator and start consuming:

```python
consumer = ReportConsumer()
consumer.connect_to_orchestrator(orchestrator, "build_report")
attempted = consumer.consume(num_messages=1)
print(attempted)
```

`consume()` returns the number of deliveries attempted, not the number that
completed successfully. Malformed JSON, invalid UTF-8, and non-object bodies
(including JSON `null`) follow the failure policy and count as attempted even
though `run()` never sees them. `None` from a receive call means the queue is empty.
If a RabbitMQ job is interrupted during `run()`, it still counts as attempted;
with manual acknowledgement, connection cleanup leaves it available for broker
redelivery. An interrupt preserves the count of deliveries already attempted.
Consumer backends use the success or failure of `run()` to apply their failure
policy, but they do not persist or return the dictionary returned by `run()`.
Store results explicitly if your application needs them.

With RabbitMQ, messages are acknowledged only after `run()` succeeds. Failed
messages are dead-lettered by default (`basic_nack(requeue=False)`); when the
queue has no dead-letter exchange, RabbitMQ discards them. The Jobs package does
not create a dead-letter exchange or queue automatically; configure those on
the RabbitMQ queue before relying on dead-letter routing.

`REQUEUE` requeues a failed delivery when RabbitMQ has not already marked it as
redelivered. A failure on a delivery whose broker `redelivered` flag is already
set is rejected with `requeue=False`. That flag may also be set after a prior
worker or connection loss, so it is broker delivery history rather than a
dedicated processing-attempt counter:

```python
from mindtrace.jobs import ConsumerFailurePolicy

consumer.connect_to_orchestrator(
    orchestrator,
    "build_report",
    failure_policy=ConsumerFailurePolicy.REQUEUE,
)
```

`connect_to_orchestrator()` passes its keyword arguments to the consumer
backend, and a `RabbitMQClient` applies the same settings to every consumer it
creates through `consumer_backend_kwargs`. Connection parameters are not among
them: naming one raises `ValueError`, so a consumer reads from the broker its
orchestrator publishes to.

Local and Redis consumers poll their queues, and `poll_timeout` sets how long a
blocking call waits after a sweep that found nothing. Shutdown interrupts that
wait, so `poll_timeout` bounds polling frequency rather than shutdown latency.

RabbitMQ `auto_ack=True` acknowledges deliveries before `run()` executes, so
it is only valid with `failure_policy=ConsumerFailurePolicy.DISCARD`.
Combining auto-acknowledgement with `REQUEUE` or `DEAD_LETTER` raises
`ValueError` during consumer backend configuration.
Auto-acknowledged deliveries have at-most-once semantics: process failure,
connection loss, or shutdown may discard deliveries that RabbitMQ acknowledged
or Pika buffered but `Consumer.run()` did not complete. `prefetch_count` does
not bound auto-acknowledged deliveries. Use `auto_ack=False` for production
workloads that require acknowledgement after processing and redelivery after a
failure.

Local and Redis consumers support only `DISCARD`. Connecting either backend
with `REQUEUE` or `DEAD_LETTER` raises `NotImplementedError`; those policies
require backend-specific retry or dead-letter storage that they do not yet
provide.

A RabbitMQ consume call reads every queue over one channel, so a broker error
raised while reading a queue ends the whole call rather than skipping that
queue: the channel that error arrives on carries the other queues too. The
error reaches the caller as the exception Pika raised, or as
`RabbitMQSettlementError` when the broker refuses an acknowledgement or
rejection, and the channel and connection are released either way. Callers
own retry and backoff around a later `consume()` call.

Calling `consumer.stop()` requests graceful shutdown. An in-flight job finishes
and is acknowledged or rejected before the blocking consume loop exits, and a
drain in progress ends through the same shutdown path. The stop request is
latched: `consume()` and `consume_until_empty()`
raise `RuntimeError` before any backend setup until the caller explicitly
invokes `consumer.reset()`.
RabbitMQ channels and connections close automatically whenever `consume()`
returns; a later call reconnects.

`consume(..., block=True)` waits until the requested number of messages has
been attempted, shutdown is requested, or the caller interrupts the operation.
`num_messages=0` means to continue indefinitely. For RabbitMQ, this bare
blocking form registers every requested queue with `basic_consume` on one
channel and lets the broker push deliveries. Stopping the operation stops all
queues registered by that call together.

Finite RabbitMQ calls (`num_messages > 0`), `consume_until_empty()`, and
`consume(..., block=False)` remain pull-based. With `block=False`, consumption
returns as soon as no message is immediately available, even if the requested
count has not been reached. `consume_until_empty()` takes no `block` setting on
any backend: it drains the work already queued and returns without waiting for
new work to arrive.

RabbitMQ invokes `Consumer.run()` synchronously on Pika's I/O thread during
broker-pushed consumption. `Consumer.run()` must not return until processing is
complete because its return or exception determines whether the delivery is
acknowledged or rejected.
An unexpected broker cancellation, such as a queue being deleted or becoming
unavailable, ends the entire push-consume operation and raises
`RabbitMQConsumerCancelledError` with the affected queue and consumer tag.
Channel and connection failures also end the current consume operation. Callers
remain responsible for retry and backoff. Push and pull calls both return the
same attempted-delivery count described above.

Calling `consumer.close()` is different from normal per-operation cleanup: it
permanently closes the consumer backend. It is safe to call more than once,
but later calls to `consume()`, `consume_until_empty()`, or `reset()` raise a
clear `RuntimeError`. This terminal close contract is the same for Local,
Redis, and RabbitMQ consumers.

### Consuming until empty

```python
consumer.consume_until_empty()
```

That is useful for local scripts, test runs, or backlog-draining workflows.

A drain ends after a complete nonblocking sweep finds no deliveries across the
requested queues. It does not take queue-depth snapshots or wait for new work.
Jobs published while draining may also be consumed; continuous arrivals can keep
a drain running until shutdown is requested. Queue and connection failures
propagate to the caller.

## Backends

The package supports three backend families.

### Local backend

`LocalClient` is the simplest backend and a good default for local development or single-process workflows.

```python
from mindtrace.jobs import LocalClient, Orchestrator


backend = LocalClient()
orchestrator = Orchestrator(backend)
```

Internally, the local backend stores queues through the registry-backed local implementation and also supports local queue variants such as:

- `LocalQueue`
- `LocalStack`
- `LocalPriorityQueue`

### Redis backend

Use `RedisClient` when you want Redis-backed queues.

```python
from mindtrace.jobs import Orchestrator, RedisClient


backend = RedisClient(host="localhost", port=6379, db=0)
orchestrator = Orchestrator(backend)
```

Redis is a good fit when you want a lightweight shared broker across multiple processes or machines.

### RabbitMQ backend

Use `RabbitMQClient` when you want RabbitMQ-backed routing and queueing.

```python
from mindtrace.jobs import Orchestrator, RabbitMQClient


backend = RabbitMQClient(
    host="localhost",
    port=5672,
    username="user",
    password="password",
)
orchestrator = Orchestrator(backend)
```

RabbitMQ is a better fit when you want broker-oriented messaging behavior, exchanges, and mature queue features such as max-priority support.

## Switching Backends

One of the main design goals of the jobs package is that your job schema and consumer logic should not need major changes when switching backends.

```python
# Development
backend = LocalClient()

# Shared test environment
backend = RedisClient(host="localhost", port=6379, db=0)

# Production / broker-oriented setup
backend = RabbitMQClient(host="localhost", port=5672, username="user", password="password")

orchestrator = Orchestrator(backend)
consumer = ReportConsumer()
consumer.connect_to_orchestrator(orchestrator, "build_report")
```

The core publishing and consuming flow stays largely the same.

## Queue Types and Priority

The local and Redis backends expose queue-type selection when declaring a queue.

### FIFO queue

```python
orchestrator.register(schema, queue_type="fifo")
```

### Stack / LIFO queue

```python
backend.declare_queue("stack_tasks", queue_type="stack")
```

### Priority queue

Higher numeric values are consumed first.

```python
backend.declare_queue("priority_tasks", queue_type="priority")

priority_job = job_from_schema(schema, ReportInput(report_id="rpt-urgent"))
background_job = job_from_schema(schema, ReportInput(report_id="rpt-background"))

orchestrator.publish("priority_tasks", priority_job, priority=100)
orchestrator.publish("priority_tasks", background_job, priority=10)
```

Local priority queues preserve publish order when priorities are equal. Redis
keeps duplicate payloads as distinct jobs but does not order equal-priority jobs
by publish time; that gap is tracked in
[#536](https://github.com/Mindtrace/mindtrace/issues/536).

### RabbitMQ priority queues

RabbitMQ does not use the same `queue_type` argument. Instead, you declare a queue with `max_priority`.

```python
backend = RabbitMQClient(host="localhost", port=5672, username="user", password="password")
backend.declare_queue("rabbitmq_priority", max_priority=255)
```

Then publish with a priority value:

```python
orchestrator = Orchestrator(backend)
job = job_from_schema(schema, ReportInput(report_id="rpt-priority"))
orchestrator.publish("rabbitmq_priority", job, priority=255)
```

## Redis Setup

For Redis-backed jobs, start a Redis server first.

```bash
$ redis-server
```

Or with Docker:

```bash
$ docker run -d --name redis -p 6379:6379 redis:latest
$ redis-cli ping
```

## RabbitMQ Setup

For RabbitMQ-backed jobs, start a RabbitMQ server first.

```bash
$ docker run -d --name rabbitmq \
    -p 5672:5672 \
    -p 15672:15672 \
    -e RABBITMQ_DEFAULT_USER=user \
    -e RABBITMQ_DEFAULT_PASS=password \
    rabbitmq:3-management
```

## Examples

Related examples in the repo:

- [Simple orchestrator example](../../samples/jobs/orchestrator_simple.py)

## Testing

If you are working in the full Mindtrace repo, run tests for this module specifically:

```bash
$ git clone https://github.com/Mindtrace/mindtrace.git && cd mindtrace
$ uv sync --dev
$ ds test: jobs
$ ds test: --unit jobs
```

## Practical Notes and Caveats

- `JobSchema` is currently an alias of `TaskSchema`, so older naming in the jobs package may reflect that transition.
- Consumers operate on `job_dict` payloads, so your `run()` implementation should be defensive about the shape it expects.
- Consumer `run()` return values are not validated against `output_schema` or stored by the consumer backends.
- Local, Redis, and RabbitMQ backends expose similar high-level workflows, but their queue semantics and operational requirements differ.
- Redis and RabbitMQ require external services; the local backend is the simplest place to start.
- Priority queue support exists across backends, but the declaration model differs for RabbitMQ vs. local/Redis backends.
