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
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""# Summary
|
|
|
|
Appointment-reminder pipeline family — concrete stages, the wired root, and its
|
|
registration. The concrete subclass re-annotates each slot with a stage
|
|
implementation; the worker wires the tree once per run with the CRUD services
|
|
injected by slot name.
|
|
"""
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from pipelines.appointment_reminder.abc import (
|
|
AppointmentReminderPipelinePort,
|
|
ComposePort,
|
|
DeliverPort,
|
|
LoadPort,
|
|
)
|
|
from pipelines.shared.loader import build_pipeline
|
|
from pipelines.shared.registry import register
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AppointmentLoader(LoadPort):
|
|
"""Load an appointment through the appointment CRUD service.
|
|
|
|
Notes:
|
|
The service instance is injected by the worker (``appointment_svc``).
|
|
"""
|
|
|
|
appointment_svc: Any
|
|
|
|
async def load(self, appointment_id: int) -> dict[str, Any]:
|
|
doc = await self.appointment_svc.read(appointment_id)
|
|
if doc is None:
|
|
raise ValueError(f"appointment {appointment_id} not found")
|
|
return doc.model_dump(mode="json", by_alias=True)
|
|
|
|
|
|
class ReminderComposer(ComposePort):
|
|
"""Compose the reminder text from the family's pattern fragments.
|
|
|
|
Notes:
|
|
``registry`` is the merged pattern map injected by the shared loader.
|
|
"""
|
|
|
|
registry: Any
|
|
|
|
def compose(self, appointment: dict[str, Any], pet: dict[str, Any] | None,
|
|
parent: dict[str, Any] | None) -> dict[str, Any]:
|
|
pattern = (self.registry or {}).get("reminder", {})
|
|
pet_name = (pet or {}).get("name", "your pet")
|
|
parent_name = (parent or {}).get("name", "there")
|
|
to = (parent or {}).get("email", "")
|
|
subject = pattern.get("subject", "Appointment reminder")
|
|
body = pattern.get(
|
|
"body",
|
|
"Hi {parent_name}, {pet_name} has an appointment on {date}.",
|
|
).format(
|
|
parent_name=parent_name,
|
|
pet_name=pet_name,
|
|
date=appointment.get("date", ""),
|
|
)
|
|
return {"to": to, "subject": subject, "body": body}
|
|
|
|
|
|
class LogDeliver(DeliverPort):
|
|
"""Deliver the reminder by logging it (swap for an email/SMS adaptor)."""
|
|
|
|
async def deliver(self, reminder: dict[str, Any]) -> dict[str, Any]:
|
|
logger.info("REMINDER to=%s subject=%s body=%s",
|
|
reminder.get("to"), reminder.get("subject"), reminder.get("body"))
|
|
reminder["delivered"] = "log"
|
|
return reminder
|
|
|
|
|
|
class AppointmentReminderPipeline(AppointmentReminderPipelinePort):
|
|
"""Concrete appointment-reminder family — the dependency decisions live here."""
|
|
|
|
loader: AppointmentLoader
|
|
compose: ReminderComposer
|
|
deliver: LogDeliver
|
|
|
|
async def run(self, appointment_id: int) -> dict[str, Any]:
|
|
appointment = await self.loader.load(appointment_id)
|
|
pet = appointment.get("pet") if isinstance(appointment.get("pet"), dict) else None
|
|
parent = None
|
|
if pet and pet.get("parents"):
|
|
first = pet["parents"][0]
|
|
parent = first if isinstance(first, dict) else None
|
|
reminder = self.compose.compose(appointment, pet, parent)
|
|
return await self.deliver.deliver(reminder)
|
|
|
|
|
|
def get_appointment_reminder_pipeline(appointment_svc) -> AppointmentReminderPipeline:
|
|
"""Wire and return the appointment-reminder pipeline through the shared loader.
|
|
|
|
Args:
|
|
appointment_svc: Appointment CRUD service injected into the load stage.
|
|
|
|
Returns:
|
|
AppointmentReminderPipeline: The wired pipeline root.
|
|
"""
|
|
resources = Path(__file__).parent / "resources"
|
|
return build_pipeline(
|
|
AppointmentReminderPipeline,
|
|
resources_dir=resources,
|
|
instances={"appointment_svc": appointment_svc},
|
|
)
|
|
|
|
|
|
register("appointment_reminder", "default", AppointmentReminderPipeline)
|