Vishesh 'ironeagle' Bangotra
0cb35d10a6
- build.py: _TEMPLATE_COMPONENTS rewritten around the vet sample (53 kinds —
per-resource slices, jobs, worked appointment_reminder family, tests);
seed tokens dropped from the manifest ('sample: vet' instead)
- registry.py: get_app_component(kind) no longer renames item/example seeds;
prompts and tool docs updated to 'copy a slice and adapt'
- templates/README.md, 01_build_recipe, 40_ui_admin, wiki/03 updated off the
old seed scheme; repo inventory/config now exclude vet (docs move under
standards; react/vet itself is deleted in a later phase)
- regenerated MCP bundle, vendored templates, and site output
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""# Summary
|
|
|
|
Appointment-reminder pipeline family — port contracts (stage interfaces + root
|
|
interface). The demo pipeline axis of the vet sample: each stage is a small,
|
|
testable unit; the DI container wires the tree.
|
|
"""
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
|
|
class LoadPort(ABC):
|
|
"""Load the appointment the job refers to."""
|
|
|
|
@abstractmethod
|
|
async def load(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Return the appointment as a JSON-ready mapping."""
|
|
|
|
|
|
class ComposePort(ABC):
|
|
"""Compose the reminder message from the appointment context."""
|
|
|
|
@abstractmethod
|
|
def compose(self, appointment: dict[str, Any], pet: dict[str, Any] | None,
|
|
parent: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""Return a reminder record ``{to, subject, body}``."""
|
|
|
|
|
|
class DeliverPort(ABC):
|
|
"""Deliver the composed reminder (log stub today, email/SMS adaptor later)."""
|
|
|
|
@abstractmethod
|
|
async def deliver(self, reminder: dict[str, Any]) -> dict[str, Any]:
|
|
"""Deliver the reminder and return it with a delivery marker."""
|
|
|
|
|
|
class AppointmentReminderPipelinePort(ABC):
|
|
"""Root of the appointment-reminder family.
|
|
|
|
Notes:
|
|
The slot annotations below are contracts. A concrete subclass re-annotates
|
|
them with stage implementations; the DI container instantiates the tree.
|
|
"""
|
|
|
|
loader: LoadPort
|
|
compose: ComposePort
|
|
deliver: DeliverPort
|
|
|
|
@abstractmethod
|
|
async def run(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Run the full stage chain and return the delivered reminder."""
|