warren.workers

Async worker that declares its capabilities (accepts / produces).

The declarative counterpart to FilteringWorkerBase: instead of hand-writing should_process, the worker declares the input data_types it accepts, and should_process defaults to "is the message's data_type one I accept?". This doubles as a runtime type guard (defense in depth) even when a topic/direct exchange has already routed only matching messages.

accepts / produces are typically supplied from the WorkerSpec via WorkerFactoryContext (single source of truth, also used by routing-plan validation).

Async worker that self-selects messages via should_process().

Subclasses implement should_process() to decide whether a message is relevant, and process() for the actual work. When should_process() returns False, the message is ACKed with no downstream publish.

Abstract base class for async processing workers.

Satisfies the AsyncMessageConsumerInterface protocol. Workers receive plain dictionaries and return optional dictionaries — they never interact with RabbitMQ or any other message transport directly.

Heartbeat safety: this worker runs on the main event loop alongside the RabbitMQ connection. Every await yields control so heartbeats can be processed.

• Pure async I/O (HTTP clients, DB queries) is safe.
• CPU-bound work MUST be offloaded via asyncio.get_running_loop().run_in_executor().
• If your worker is primarily CPU-bound or uses sync libraries, use SyncProcessingWorkerBase instead — the framework handles threading automatically.

Use this base class when:

• Worker needs async I/O with shared connection pools (async HTTP clients, async DB clients).
• Worker combines async I/O with some CPU work (offload CPU parts via run_in_executor).

Abstract base class for sync processing workers.

Satisfies the SyncMessageConsumerInterface protocol. The framework automatically runs __call__ in a ThreadPoolExecutor via run_in_executor(), keeping the main event loop free for RabbitMQ heartbeats.

Threading behavior: each invocation runs in a thread from the default ThreadPoolExecutor (sized min(32, os.cpu_count() + 4)). The worker may spawn its own ProcessPoolExecutor or multiprocessing.Pool internally for CPU parallelism across cores.

• CPU-bound work and sync library calls (requests, file I/O, sync DB drivers) work naturally.
• Do NOT use asyncio.run() or asyncio.get_event_loop() — async objects from the main loop cannot be used from a different thread due to event loop affinity.
• If you need async I/O (async HTTP clients, async DB clients), use AsyncProcessingWorkerBase instead.

Use this base class when:

• Worker does CPU-bound work (parsing, extraction, computation).
• Worker uses sync libraries (requests, file I/O, sync DB drivers).
• Worker manages its own multiprocessing internally.

Canonical message format for the distributed processing framework.

Workers use data_type (and optionally job_metadata) to decide whether to process a message, and job_parameters to decide how.

Raises:

• ValueError: If priority is a bool, or an int outside 0 through 255.

Identifies the source that produced a message.

Retry state attached to messages on soft failure.

Base class for running a worker process.

Provides the generic lifecycle: signal handling, consume loop, and graceful shutdown. Subclasses implement setup() to wire their specific components (connections, stores, worker, consumer manager).

A subclass uses these protected names:

• self._consumer_manager: set it in setup(). run() and teardown() use it.
• self._mark_setup_succeeded(): call it at the end of setup(). Without it, run() raises RuntimeError.
• _on_teardown(): override it for subclass-specific cleanup (e.g., closing connections, cancelling timers). It runs after the consumer is stopped but before state is reset.
• self._exception_wrapping(description): a context manager for a setup phase. It re-raises a failure as WarrenError with the phase and the worker name.

Creates a consumer manager for the given consumer.

Signature: (consumer) -> ConsumerManagerInterface

The factory encapsulates transport-specific wiring (exchange, queue, connection) so runners stay transport-agnostic. The consumer carries its own identity (name and type) for use on failure envelopes.

Watchdog and endpoint settings.

Minimal HTTP readiness endpoint.

GET /ready → 200 while the last sample says ready, else 503; GET /live → 503 if a handler exceeds twice its timeout, else 200. Anything else → 404. The body is the sample as JSON.

Build a composite message key from identity fields.

Consistent key format used across ResultsStore, RetryWorker, and cache layers. Format: job:{job_id}:doc:{doc_id}:part:{part_idx}.

None values are normalized to "-" so keys remain valid and consistent regardless of whether identifiers are available.

Build a key prefix for matching all parts of a document.

Warning: doc_id is the first parameter, but build_message_key takes job_id first. Pass the arguments by keyword, or the prefix matches no key.

Use with CacheInterface.get_by_key_prefix() to retrieve all parts of a document within a job. Includes trailing separator to avoid matching unrelated keys.

None values are normalized to "-".

Extract identity fields from a message dict for logging.

Returns a human-readable string with job_id, doc_id, and part_idx. Missing fields are shown as "?". Safe to call on any message dict.

Callable that extracts a human-readable identity string from a message dict.