warren.jobs.publishing
Job publication: a worker that turns a job message into one message for
each document, and the base class for the publisher that loads and registers
each document.
Classes
DocumentsPublisherFactoryFunc
classDocumentsPublisherFactoryFuncFactory protocol for creating application-specific publishers.
Called by JobPublicationWorkerRunner.setup() after infrastructure
is created. The factory receives the runner's shared resources and
is responsible for creating all components the publisher needs
(stores, adapters, etc.).
method__call__async (publisher, infra, config, worker_name) -> JobDocumentsPublisherCreate the documents publisher.
parampublisherPublisherInterfacepublisher for downstream messages, shared with the consumer manager. Already created but not yet set up — setup happens when the consumer manager starts.
paraminfraRuntimeInfraruntime connections (MongoDB, Redis, pubsub connection). Use infra.mongo_client to create stores that share the runner's connection rather than opening a second one.
paramconfigRuntimeConfigruntime configuration. Use config.mongodb.database for store database names and any other infrastructure settings the publisher needs.
paramworker_namestrunique worker instance name, useful for naming the publisher and log context.
Returns
JobDocumentsPublisherThe documents publisher that the worker uses.
JobDocumentsPublisher
classJobDocumentsPublisher(*, publisher, tracker, job_store, name=None)Abstract publisher harness with automatic outcome tracking.
A subclass implements four abstract methods:
async _load_document(self, source: Any) -> Any— retrieve document bytes/dataasync _register_document(self, job_id: str, doc_data: Any) -> str— store in document DB, getdoc_id_create_message(self, job_id: str, doc_id: str, doc_data: Any, job_parameters: dict[str, Any]) -> dict— build the exchange message. The message must setorigin.typetoPUBLISHER_ORIGIN_TYPE("publisher")._get_source_id(self, source: Any) -> str— a readable identifier of the source, for logs and failure records
The publishing flow for each document is load, register, then create
and publish the message. If a step raises, the failure is recorded
with its stage and the next document continues.
HardFailureException is the exception: it stops the job.
parampublisherPublisherInterfacePublisher for sending messages to the exchange.
paramtrackerPublishingTrackerInterfaceTracks publishing outcomes per document.
paramjob_storeJobStoreInterfaceUsed to set num_documents after publishing and to mark zero-document jobs as completed.
paramnamestr | None= NoneOptional logger name.
methodpublish_jobasync (job_id, sources, job_parameters=None, *, scope=None) -> dictPublish documents for a job from an async source.
Iterates over sources, processing each through the load → register → publish pipeline.
paramjob_idstrThe job identifier (must already exist in job_store).
paramsourcesAsyncIterableAsync iterable of document sources (type is application-defined).
paramjob_parametersdict[str, Any] | None= NoneJob-level configuration passed through to _create_message so concrete publishers can fold job settings into the per-document message. None is normalised to an empty dict.
paramscopestr | None= NoneStorage scope for the job. Each published message gets it as scope. None keeps the current scope.
Returns
dictSummary dict with published, failed, and total counts.
JobPublicationWorker
classJobPublicationWorker(worker_name, *, documents_publisher, create_source_generator=None, worker_type=None)Consumes job messages and publishes their documents.
Filters for data_type == "job" on the observer exchange. For each
matching message, calls create_source_generator to extract an
async iterable of document sources from the message data, then
delegates to JobDocumentsPublisher.publish_job() for the actual
load -> register -> publish orchestration.
paramworker_namestrUnique worker instance name.
paramdocuments_publisherJobDocumentsPublisherPublisher harness for the load -> register -> publish flow.
paramcreate_source_generatorCallable[[dict], AsyncIterable] | None= NoneCallable that receives the message data dict and returns an AsyncIterable of document sources. When None, defaults to iterating data["items"] — the standard job message format where data contains a list of document source items.
paramworker_typestr | None= NoneOptional override for the worker type (default: "job_publication_worker").
methodshould_process(message) -> boolOnly process job messages whose preprocessing is complete.
Like the per-document workers, this consumer accepts a message
only when its preprocessing_required list is empty. Job-level
preprocessing steps gate document fan-out: a step's worker removes
its name from preprocessing_required and emits the job message
again.
parammessagedictmethodprocessasync (message) -> dict | NoneExtract sources from the job message and publish them.
parammessagedictJob message with job_id, data, and an optional job_parameters dict (forwarded to the publisher so per-document messages can carry job-level settings).
Returns
dict | NoneNone -- publishing produces individual document messages via the publisher's own PublisherInterface.
JobPublicationWorkerRunner
classJobPublicationWorkerRunner(config, worker_name, *, exchange, publish_exchange=None, route_func=None, documents_publisher_factory, consumer_manager_factory=None, create_source_generator=None)Runs a JobPublicationWorker with its lifecycle hooks.
Creates infrastructure from RuntimeConfig, then calls the
injected documents_publisher_factory to build the
application-specific publisher. The factory receives the runner's
shared publisher and infrastructure so it can create stores
from the same connections.
This runner has no infra parameter: it always creates its own
infrastructure. On backend: memory, it creates its own broker,
so it cannot reach the other runners of the process.
paramconfigRuntimeConfigruntime infrastructure configuration.
paramworker_namestrunique identifier for this worker instance.
paramexchangeRMQExchangeConfigThe observer exchange (observer_exchange(pipeline.exchange)).
parampublish_exchangeRMQExchangeConfig | None= NoneThe data exchange (pipeline.exchange). Defaults to exchange, which is correct for fanout and topic only.
paramroute_funcRouteFunc | None= Noneper-document publish routing function, as in PublishSpec.route_func. When None, preserves the default: no routing key on fanout, or routing by data_type otherwise.
paramdocuments_publisher_factoryDocumentsPublisherFactoryFuncasync callable matching DocumentsPublisherFactoryFunc. Called in setup() after infrastructure is created.
paramconsumer_manager_factoryConsumerManagerFactory | None= Noneoptional override for the consumer manager factory. Default: the consumer manager of config.backend (from create_consumer_manager) on the publication queue.
paramcreate_source_generatorCallable[[dict], AsyncIterable] | None= Noneoptional callable that receives the message data dict and returns an AsyncIterable of document sources. When None, defaults to iterating data["items"].
methodsetupasync () -> NoneCreate infrastructure, build publisher, wire the worker.
- Create infrastructure (MongoDB, Redis, pubsub connection)
- Create publisher for downstream messages
- Call documents publisher factory with shared infrastructure
- Build default consumer factory if not injected
- Create the JobPublicationWorker and consumer manager
Constants
PUBLICATION_WORKER_TYPE
constantPUBLICATION_WORKER_TYPE: str = 'publication'