fix: repair cp1252-mangled arrows in the reference-app pipelines page
All checks were successful
continuous-integration/drone/push Build is passing
All checks were successful
continuous-integration/drone/push Build is passing
PowerShell editing round-tripped the U+2192 arrows through cp1252 in 06_reference_app/03_pipelines_and_worker.md; restored them and regenerated the wiki HTML + MCP module.
This commit is contained in:
@@ -2,7 +2,7 @@
|
||||
"module": "wiki_reference_app_pipelines_and_worker",
|
||||
"content": {
|
||||
"path": "wiki_reference_app_pipelines_and_worker",
|
||||
"docstring": "# Pipelines & Worker — the pipeline axis\n\nThe sample proves the full loop: **API creates a job document → worker polls →\npipeline family runs → job reaches a terminal status.**\n\n## The jobs slice\n\n`domain/jobs/` is an ordinary CRUD slice; its model is the durable state\nmachine the worker drives:\n\n```python\nclass Job(BaseModel):\n name: str\n kind: str # dispatch key\n status: Literal[\"pending\", \"processing\", \"done\", \"failed\"] = \"pending\"\n payload: dict | None = None\n retry_count: int = 0\n error_message: str | None = None\n```\n\n`POST /jobs` with `{\"name\": \"...\", \"kind\": \"appointment_reminder\",\n\"payload\": {\"appointment_id\": 1}}` enqueues work; `GET /jobs?status=...`\ninspects the queue.\n\n## The appointment_reminder family\n\n```\npipelines/appointment_reminder/\n abc.py # LoadPort / ComposePort / DeliverPort + root interface\n impl.py # concrete stages, wired root, register(\"appointment_reminder\", ...)\n pipeline.yaml # graph spec (nodes, edges, roots)\n resources/\n pipeline.yaml # runtime spec consumed by the loader\n patterns/reminder.yaml # message template fragments\npipelines/shared/\n ports.py loader.py registry.py # reused by every family\n```\n\nEach stage is a small, independently testable unit. The loader\n(`pipelines/shared/loader.py`) wires the root through `hexa.build` and merges\n`resources/patterns/*.yaml` into the stage-injected registry:\n\n```python\npipeline = get_appointment_reminder_pipeline(appointment_svc)\nreminder = await pipeline.run(appointment_id)\n```\n\nTo add a family: copy `appointment_reminder/`, replace the stages, call\n`register(kind, variant, root_cls)` — see\n[Pipeline Scaffold](../lib/50_pipeline_scaffold/).\n\n## The worker\n\n`worker.py` is the process that turns\ndocuments into execution:\n\n1. poll `job_svc.search({\"status\": \"pending\"})` on `WORKER_POLL_SECONDS`\n2. claim a job: `status → processing`\n3. dispatch through the `HANDLERS` table — `HANDLERS[job.kind](job, resources)`\n4. on success `status → done`; on exception `status → failed`,\n `retry_count += 1`, `error_message` captured; jobs at\n `WORKER_MAX_RETRIES` are skipped\n\n```python\nHANDLERS: dict[str, Callable] = {\n \"appointment_reminder\": run_appointment_reminder,\n # add yours here\n}\n```\n\nThe worker shares the composition root: `from init import init` gives it the\nsame wired services the API uses — one place to add a dependency.\n\n!!! note \"Same process model in production\"\n API and worker are separate processes (see `docker-compose.yaml`: `api`,\n `worker`, `mongo`). Any in-memory progress bus is per-process; to surface\n live progress through the API, swap the bus adaptor for a broker-backed\n one — see [Worker & Progress](../lib/51_worker_and_progress/).\n\n## Tests\n\n- `tests/unit/test_appointment_reminder_pipeline.py` — stage units + the wired\n root against a fake service.\n- `tests/unit/test_worker.py` — dispatch, done/failed transitions, retry\n exhaustion, using an in-memory job service double and a cancelled poll task.\n",
|
||||
"docstring": "# Pipelines & Worker — the pipeline axis\n\nThe sample proves the full loop: **API creates a job document → worker polls →\npipeline family runs → job reaches a terminal status.**\n\n## The jobs slice\n\n`domain/jobs/` is an ordinary CRUD slice; its model is the durable state\nmachine the worker drives:\n\n```python\nclass Job(BaseModel):\n name: str\n kind: str # dispatch key\n status: Literal[\"pending\", \"processing\", \"done\", \"failed\"] = \"pending\"\n payload: dict | None = None\n retry_count: int = 0\n error_message: str | None = None\n```\n\n`POST /jobs` with `{\"name\": \"...\", \"kind\": \"appointment_reminder\",\n\"payload\": {\"appointment_id\": 1}}` enqueues work; `GET /jobs?status=...`\ninspects the queue.\n\n## The appointment_reminder family\n\n```\npipelines/appointment_reminder/\n abc.py # LoadPort / ComposePort / DeliverPort + root interface\n impl.py # concrete stages, wired root, register(\"appointment_reminder\", ...)\n pipeline.yaml # graph spec (nodes, edges, roots)\n resources/\n pipeline.yaml # runtime spec consumed by the loader\n patterns/reminder.yaml # message template fragments\npipelines/shared/\n ports.py loader.py registry.py # reused by every family\n```\n\nEach stage is a small, independently testable unit. The loader\n(`pipelines/shared/loader.py`) wires the root through `hexa.build` and merges\n`resources/patterns/*.yaml` into the stage-injected registry:\n\n```python\npipeline = get_appointment_reminder_pipeline(appointment_svc)\nreminder = await pipeline.run(appointment_id)\n```\n\nTo add a family: copy `appointment_reminder/`, replace the stages, call\n`register(kind, variant, root_cls)` — see\n[Pipeline Scaffold](../lib/50_pipeline_scaffold/).\n\n## The worker\n\n`worker.py` is the process that turns\ndocuments into execution:\n\n1. poll `job_svc.search({\"status\": \"pending\"})` on `WORKER_POLL_SECONDS`\n2. claim a job: `status → processing`\n3. dispatch through the `HANDLERS` table — `HANDLERS[job.kind](job, resources)`\n4. on success `status → done`; on exception `status → failed`,\n `retry_count += 1`, `error_message` captured; jobs at\n `WORKER_MAX_RETRIES` are skipped\n\n```python\nHANDLERS: dict[str, Callable] = {\n \"appointment_reminder\": run_appointment_reminder,\n # add yours here\n}\n```\n\nThe worker shares the composition root: `from init import init` gives it the\nsame wired services the API uses — one place to add a dependency.\n\n!!! note \"Same process model in production\"\n API and worker are separate processes (see `docker-compose.yaml`: `api`,\n `worker`, `mongo`). Any in-memory progress bus is per-process; to surface\n live progress through the API, swap the bus adaptor for a broker-backed\n one — see [Worker & Progress](../lib/51_worker_and_progress/).\n\n## Tests\n\n- `tests/unit/test_appointment_reminder_pipeline.py` — stage units + the wired\n root against a fake service.\n- `tests/unit/test_worker.py` — dispatch, done/failed transitions, retry\n exhaustion, using an in-memory job service double and a cancelled poll task.\n",
|
||||
"objects": {}
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
# Pipelines & Worker — the pipeline axis
|
||||
|
||||
The sample proves the full loop: **API creates a job document → worker polls →
|
||||
pipeline family runs → job reaches a terminal status.**
|
||||
The sample proves the full loop: **API creates a job document → worker polls →
|
||||
pipeline family runs → job reaches a terminal status.**
|
||||
|
||||
## The jobs slice
|
||||
|
||||
@@ -55,9 +55,9 @@ To add a family: copy `appointment_reminder/`, replace the stages, call
|
||||
documents into execution:
|
||||
|
||||
1. poll `job_svc.search({"status": "pending"})` on `WORKER_POLL_SECONDS`
|
||||
2. claim a job: `status → processing`
|
||||
2. claim a job: `status → processing`
|
||||
3. dispatch through the `HANDLERS` table — `HANDLERS[job.kind](job, resources)`
|
||||
4. on success `status → done`; on exception `status → failed`,
|
||||
4. on success `status → done`; on exception `status → failed`,
|
||||
`retry_count += 1`, `error_message` captured; jobs at
|
||||
`WORKER_MAX_RETRIES` are skipped
|
||||
|
||||
|
||||
@@ -852,8 +852,8 @@
|
||||
|
||||
|
||||
<h1 id="pipelines-worker-a-the-pipeline-axis">Pipelines & Worker — the pipeline axis</h1>
|
||||
<p>The sample proves the full loop: <strong>API creates a job document → worker polls →
|
||||
pipeline family runs → job reaches a terminal status.</strong></p>
|
||||
<p>The sample proves the full loop: <strong>API creates a job document → worker polls →
|
||||
pipeline family runs → job reaches a terminal status.</strong></p>
|
||||
<h2 id="the-jobs-slice">The jobs slice</h2>
|
||||
<p><code>domain/jobs/</code> is an ordinary CRUD slice; its model is the durable state
|
||||
machine the worker drives:</p>
|
||||
@@ -908,9 +908,9 @@ inspects the queue.</p>
|
||||
documents into execution:</p>
|
||||
<ol>
|
||||
<li>poll <code>job_svc.search({"status": "pending"})</code> on <code>WORKER_POLL_SECONDS</code></li>
|
||||
<li>claim a job: <code>status → processing</code></li>
|
||||
<li>claim a job: <code>status → processing</code></li>
|
||||
<li>dispatch through the <code>HANDLERS</code> table — <code>HANDLERS[job.kind](job, resources)</code></li>
|
||||
<li>on success <code>status → done</code>; on exception <code>status → failed</code>,
|
||||
<li>on success <code>status → done</code>; on exception <code>status → failed</code>,
|
||||
<code>retry_count += 1</code>, <code>error_message</code> captured; jobs at
|
||||
<code>WORKER_MAX_RETRIES</code> are skipped</li>
|
||||
</ol>
|
||||
|
||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user