32 lines
966 B
Python
32 lines
966 B
Python
"""# Summary
|
|
|
|
Item-sync pipeline — the demo pipeline axis of the blueprint.
|
|
|
|
Runs on the worker, consumes a pending job, and materializes its payload as an
|
|
item. Steps are plain async functions here; once the chain branches or needs
|
|
deterministic replay, wrap the steps in a dagpipe DAG (see standards
|
|
"06_pipeline_scaffold").
|
|
"""
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def run_item_sync(job, resources: dict) -> None:
|
|
"""Materialize a pending job payload as an item.
|
|
|
|
Step 1: extract the payload from the job.
|
|
Step 2: create the item through the item service.
|
|
|
|
Args:
|
|
job: The pending job document.
|
|
resources (dict): Wired services and adaptors.
|
|
"""
|
|
payload = job.payload or {}
|
|
item_svc = resources["item_svc"]
|
|
|
|
await item_svc.create({
|
|
"name": payload.get("name", job.name),
|
|
"quantity": payload.get("quantity", 1),
|
|
})
|
|
logger.info("Created item from job %s", str(job.id)) |