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

pip install mindtrace-jobs

Quick Start

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:

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.

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().

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

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

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.

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:

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:

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

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.

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.

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.

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.

# 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

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

Stack / LIFO queue

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

Priority queue

Higher numeric values are consumed first.

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.

RabbitMQ priority queues

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

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

Then publish with a priority value:

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.

$ redis-server

Or with Docker:

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

RabbitMQ Setup

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

$ 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

Testing

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

$ 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.