aetos/docs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 6: replace app template with the vet sample domain + jobs slice, pipelines and worker

Browse Source 
- templates/app now carries the vet server (parents/pets/treatments/vets/appointments)
- added domain/jobs slice with /jobs OpenAPI paths and Job schema (x- extensions)
- added pipelines/shared scaffold (loader, ports, registry) and the
  appointment_reminder family; worker.py dispatches jobs by kind
- wired init.py, config.py worker knobs, .env.example, docker-compose worker
- unit tests for jobs routes, the pipeline family and worker dispatch (12 passing)
This commit is contained in:
Vishesh 'ironeagle' Bangotra
2026-10-20 14:49:10 +05:30
parent 6f6f0d6e6d
commit 56aa1322fb
83 changed files with 2946 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
standards/src/templates/app/.dockerignore
View File

@@ -5,4 +5,8 @@ __pycache__/
.git/
docs/
dist/
build/
build/
uploads/
_gen/
.pytest_cache/
.ruff_cache/

12
standards/src/templates/app/.env.example
View File

@@ -1,9 +1,9 @@
DEBUG=false
REPO_BACKEND=memory
MONGO_URI=mongodb://localhost:27017
MONGO_DB=app_template
PORT=8000
WORKER_POLL_SECONDS=30
WORKER_MAX_RETRIES=3
AUTH_SERVER=http://localhost:8000
JWT_SERVER=http://localhost:8000
MONGO_DB=vet
UPLOADS_DIR=uploads
PORT=8020
AUTH_SERVER=https://auth.aetoskia.com
WORKER_POLL_SECONDS=5
WORKER_MAX_RETRIES=3

3
standards/src/templates/app/.gitignore vendored
View File

@@ -15,6 +15,9 @@ venv/
# Generated wire contract (regenerate from openapi.yaml)
_gen/
# Uploads
uploads/
# Packaging / coverage
dist/
build/

8
standards/src/templates/app/Dockerfile
View File

@@ -12,9 +12,7 @@ COPY . .
RUN pip install --no-cache-dir --upgrade pip \
&& pip install --no-cache-dir --extra-index-url "${PIP_EXTRA_INDEX_URL}" .
EXPOSE 8000
EXPOSE 8020
# Two processes, per the two axes:
# - main.py : the spec-first FastAPI application
# - worker.py : the pipeline consumer polling pending jobs
CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port 8000 & exec python worker.py"]
# Spec-first FastAPI application (no worker by design for this service)
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8020"]

1
standards/src/templates/app/VERSION Normal file
View File

@@ -0,0 +1 @@
0.1.0

42
standards/src/templates/app/config.py
View File

@@ -6,15 +6,17 @@ import time.
---
The template reads these knobs:
The server reads six knobs:
- ``repo_backend`` — ``memory`` (default, zero-dependency process-local store)
or ``mongo`` (persistent, via the shared mongo-ops connection manager).
- ``repo_backend`` — ``memory`` (default, zero-dependency demo store) or
``mongo`` (persistent, via the shared mongo-ops connection manager).
- ``mongo_uri`` / ``mongo_db`` — connection targets when the Mongo backend is
selected.
- ``port`` — the port uvicorn binds (keep in sync with the OpenAPI
``servers:`` entry and the client's ``VITE_API_BASE_URL``).
- ``worker_poll_seconds`` / ``worker_max_retries`` — background worker cadence.
- ``uploads_dir`` — directory that receives uploaded pet photos.
- ``port`` — the port uvicorn binds (kept in sync with the OpenAPI
``servers:`` entry and the frontend's ``VITE_API_BASE_URL``).
- ``worker_poll_seconds`` — background worker poll cadence.
- ``worker_max_retries`` — failed-attempt budget before a job is skipped.
``AUTH_SERVER`` is intentionally *not* re-exported here: ``openapi_first``
resolves ``{AUTH_SERVER}`` straight from the process environment when the
@@ -34,18 +36,24 @@ class Settings:
repo_backend (str): Persistence flavour, ``memory`` or ``mongo``.
Defaults to ``memory``.
mongo_uri (str): MongoDB connection URI. Defaults to mongodb://localhost:27017.
mongo_db (str): MongoDB database name. Defaults to app_template.
port (int): Local HTTP port. Defaults to 8000.
worker_poll_seconds (int): Worker poll interval in seconds. Defaults to 30.
worker_max_retries (int): Maximum job retries before skipping. Defaults to 3.
mongo_db (str): MongoDB database name. Defaults to vet.
auth_server (str): Auth-server base URL surfaced to operators and
scripts. Defaults to https://auth.aetoskia.com.
port (int): Local HTTP port. Defaults to 8020.
uploads_dir (str): Directory for uploaded pet photos. Defaults to uploads.
worker_poll_seconds (float): Worker poll cadence in seconds. Defaults to 5.
worker_max_retries (int): Failed attempts before a job is skipped.
Defaults to 3.
"""
debug: bool = False
repo_backend: str = "memory"
mongo_uri: str = "mongodb://localhost:27017"
mongo_db: str = "app_template"
port: int = 8000
worker_poll_seconds: int = 30
mongo_db: str = "vet"
auth_server: str = "https://auth.aetoskia.com"
port: int = 8020
uploads_dir: str = "uploads"
worker_poll_seconds: float = 5.0
worker_max_retries: int = 3
@@ -53,8 +61,10 @@ settings = Settings(
debug=os.environ.get("DEBUG", "").lower() in ("1", "true", "yes"),
repo_backend=os.environ.get("REPO_BACKEND", "memory").lower(),
mongo_uri=os.environ.get("MONGO_URI", "mongodb://localhost:27017"),
mongo_db=os.environ.get("MONGO_DB", "app_template"),
port=int(os.environ.get("PORT", "8000")),
worker_poll_seconds=int(os.environ.get("WORKER_POLL_SECONDS", "30")),
mongo_db=os.environ.get("MONGO_DB", "vet"),
auth_server=os.environ.get("AUTH_SERVER", "https://auth.aetoskia.com"),
port=int(os.environ.get("PORT", "8020")),
uploads_dir=os.environ.get("UPLOADS_DIR", "uploads"),
worker_poll_seconds=float(os.environ.get("WORKER_POLL_SECONDS", "5")),
worker_max_retries=int(os.environ.get("WORKER_MAX_RETRIES", "3")),
)

4
standards/src/templates/app/docker-compose.yaml
View File

@@ -1,10 +1,10 @@
services:
api:
build: .
command: uvicorn main:app --host 0.0.0.0 --port 8000
command: uvicorn main:app --host 0.0.0.0 --port 8020
env_file: .env
ports:
- "8000:8000"
- "8020:8020"
depends_on:
- mongo

3
standards/src/templates/app/domain/appointments/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Appointments domain slice."""

68
standards/src/templates/app/domain/appointments/models.py Normal file
View File

@@ -0,0 +1,68 @@
"""# Summary
Appointments domain — scheduled visits linking a pet, a vet and a treatment.
"""
from pydantic import BaseModel
from domain.parents.models import Metadata
class ProcedureNote(BaseModel):
"""A single procedure note; ``noteType`` selects the shape.
Attributes:
noteType (str): Discriminator (basic, heart_rate, dental, vaccine,
preop, surgery).
summary (str | None): Short summary. Defaults to None.
details (str | None): Longer details. Defaults to None.
"""
noteType: str
summary: str | None = None
details: str | None = None
model_config = {"extra": "ignore"}
class Procedure(BaseModel):
"""A procedure performed during an appointment.
Attributes:
name (str): Procedure name.
description (str | None): Procedure description. Defaults to None.
cost (float | None): Cost of the procedure. Defaults to None.
notes (ProcedureNote | None): Typed procedure note. Defaults to None.
"""
name: str
description: str | None = None
cost: float | None = None
notes: ProcedureNote | None = None
model_config = {"extra": "ignore"}
class Appointment(BaseModel):
"""A scheduled appointment.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
date (str): ISO date-time of the appointment.
notes (str | None): Additional notes. Defaults to None.
procedures (list[Procedure]): Procedures performed. Defaults to empty.
pet (dict | int | None): Embedded pet object or pet id.
vet (dict | int | None): Embedded vet object or vet id.
treatment (dict | int | None): Embedded treatment or treatment id.
metadata (Metadata | None): Presentation metadata. Defaults to None.
"""
id: int | None = None
date: str
notes: str | None = None
procedures: list[Procedure] = []
pet: dict | int | None = None
vet: dict | int | None = None
treatment: dict | int | None = None
metadata: Metadata | None = None
model_config = {"extra": "ignore"}

3
standards/src/templates/app/domain/appointments/ports/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Appointments domain ports."""

25
standards/src/templates/app/domain/appointments/ports/db.py Normal file
View File

@@ -0,0 +1,25 @@
"""# Summary
Appointments persistence ports — integer-id document and repository factory.
"""
from mongo_ops import BaseDocument
from pydantic import Field
from domain.appointments.models import Appointment
from infra.db.repos import create_repo
class AppointmentDocument(Appointment, BaseDocument):
"""Persisted appointment document. The integer ``id`` shadows the ObjectId."""
id: int | None = Field(default=None, validation_alias="_id")
def create_appointment_repo():
"""Create the appointment repository selected by ``settings.repo_backend``.
Returns:
IntIdMemoryRepository | IntIdMongoRepository: Repository wired to the
``appointments`` collection/store.
"""
return create_repo("appointments", AppointmentDocument)

34
standards/src/templates/app/domain/appointments/ports/service.py Normal file
View File

@@ -0,0 +1,34 @@
"""# Summary
CRUD service wiring for the appointment aggregate via the hexa container.
"""
from hexa import build
from domain.appointments.models import Appointment
from domain.appointments.ports.db import AppointmentDocument
from infra.adapters.service import CrudsServiceAdaptor
class AppointmentCrudsService(CrudsServiceAdaptor[Appointment, AppointmentDocument]):
"""Appointment CRUD service resolving :class:`AppointmentDocument` from a repo.
Notes:
Instances are created by :func:`get_appointment_service`. The ``model``
and ``schema`` class attributes bind the domain model and its document.
"""
model = Appointment
schema = AppointmentDocument
def get_appointment_service(repo) -> AppointmentCrudsService:
"""Build an appointment CRUD service wired to ``repo``.
Args:
repo (Any): Integer-id appointment repository from
:func:`create_appointment_repo`.
Returns:
AppointmentCrudsService: Ready-to-use appointment CRUD service.
"""
return build(AppointmentCrudsService, instances={"repo": repo})

21
standards/src/templates/app/domain/item/models.py
View File

@@ -1,21 +0,0 @@
"""# Summary
Item domain model — the neutral resource used throughout the template.
"""
from pydantic import BaseModel, Field
class Item(BaseModel):
"""A single countable resource.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
name (str): Item name.
quantity (int): Item quantity. Defaults to 1.
"""
id: int | None = None
name: str
quantity: int = Field(default=1)
model_config = {"extra": "ignore"}

18
standards/src/templates/app/domain/item/ports/converters.py
View File

@@ -1,18 +0,0 @@
"""# Summary
Optional converters for the item slice — pure functions mapping between domain
types. Add a converter here when a service must reshape data; keep it pure.
"""
from domain.item.models import Item
def to_item(data: dict) -> Item:
"""Build an :class:`Item` from a raw mapping.
Args:
data (dict): Field mapping matching the item model.
Returns:
Item: The constructed item model.
"""
return Item(**data)

25
standards/src/templates/app/domain/item/ports/db.py
View File

@@ -1,25 +0,0 @@
"""# Summary
Items persistence ports — integer-id document and repository factory.
"""
from mongo_ops import BaseDocument
from pydantic import Field
from domain.item.models import Item
from infra.db.repos import create_repo
class ItemDocument(Item, BaseDocument):
"""Persisted item document. The integer ``id`` shadows the ObjectId."""
id: int | None = Field(default=None, validation_alias="_id")
def create_item_repo():
"""Create the item repository selected by ``settings.repo_backend``.
Returns:
IntIdMemoryRepository | IntIdMongoRepository: Repository wired to the
``items`` collection/store.
"""
return create_repo("items", ItemDocument)

33
standards/src/templates/app/domain/item/ports/service.py
View File

@@ -1,33 +0,0 @@
"""# Summary
CRUD service wiring for the item aggregate via the hexa container.
"""
from hexa import build
from domain.item.models import Item
from domain.item.ports.db import ItemDocument
from infra.adapters.service import CrudsServiceAdaptor
class ItemCrudsService(CrudsServiceAdaptor[Item, ItemDocument]):
"""Item CRUD service resolving :class:`ItemDocument` from a repo.
Notes:
Instances are created by :func:`get_item_service`. The ``model`` and
``schema`` class attributes bind the domain model and its document type.
"""
model = Item
schema = ItemDocument
def get_item_service(repo) -> ItemCrudsService:
"""Build an item CRUD service wired to ``repo``.
Args:
repo (Any): Mongo-backed item repository from :func:`create_item_repo`.
Returns:
ItemCrudsService: Ready-to-use item CRUD service.
"""
return build(ItemCrudsService, instances={"repo": repo})

29
standards/src/templates/app/domain/job/models.py
View File

@@ -1,29 +0,0 @@
"""# Summary
Job domain model — a pipeline work item consumed by the background worker.
"""
from typing import Literal
from pydantic import BaseModel, Field
class Job(BaseModel):
"""A unit of background work.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
name (str): Job name.
status (str): Lifecycle status. Defaults to "pending".
retry_count (int): Times the job failed. Defaults to 0.
error_message (str | None): Last failure message. Defaults to None.
payload (dict | None): Arbitrary job payload. Defaults to None.
"""
id: int | None = None
name: str
status: Literal["pending", "processing", "done", "failed"] = "pending"
retry_count: int = Field(default=0)
error_message: str | None = None
payload: dict | None = None
model_config = {"extra": "ignore"}

0
standards/src/templates/app/domain/item/__init__.py → standards/src/templates/app/domain/jobs/__init__.py
View File

37
standards/src/templates/app/domain/jobs/models.py Normal file
View File

@@ -0,0 +1,37 @@
"""# Summary
Jobs domain — the worker's queue. Each job is a unit of pipeline work the
background worker claims and executes.
"""
from typing import Literal
from pydantic import BaseModel
from domain.parents.models import Metadata
class Job(BaseModel):
"""A unit of background work.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
name (str): Human label for the job.
kind (str): Dispatch key — selects the pipeline (e.g.
``appointment_reminder``).
status (str): Lifecycle status. Defaults to ``pending``.
payload (dict | None): Job-specific input. Defaults to None.
retry_count (int): Failed attempts so far. Defaults to 0.
error_message (str | None): Last failure detail. Defaults to None.
metadata (Metadata | None): Presentation metadata. Defaults to None.
"""
id: int | None = None
name: str
kind: str
status: Literal["pending", "processing", "done", "failed"] = "pending"
payload: dict | None = None
retry_count: int = 0
error_message: str | None = None
metadata: Metadata | None = None
model_config = {"extra": "ignore"}

0
standards/src/templates/app/domain/item/ports/__init__.py → standards/src/templates/app/domain/jobs/ports/__init__.py
View File

4
standards/src/templates/app/domain/job/ports/db.py → standards/src/templates/app/domain/jobs/ports/db.py
View File

@@ -5,7 +5,8 @@ Jobs persistence ports — integer-id document and repository factory.
from mongo_ops import BaseDocument
from pydantic import Field
from domain.job.models import Job
from config import settings
from domain.jobs.models import Job
from infra.db.repos import create_repo
@@ -22,4 +23,5 @@ def create_job_repo():
IntIdMemoryRepository | IntIdMongoRepository: Repository wired to the
``jobs`` collection/store.
"""
_ = settings # backend chosen centrally by create_repo
return create_repo("jobs", JobDocument)

8
standards/src/templates/app/domain/job/ports/service.py → standards/src/templates/app/domain/jobs/ports/service.py
View File

@@ -4,8 +4,8 @@ CRUD service wiring for the job aggregate via the hexa container.
"""
from hexa import build
from domain.job.models import Job
from domain.job.ports.db import JobDocument
from domain.jobs.models import Job
from domain.jobs.ports.db import JobDocument
from infra.adapters.service import CrudsServiceAdaptor
@@ -25,9 +25,9 @@ def get_job_service(repo) -> JobCrudsService:
"""Build a job CRUD service wired to ``repo``.
Args:
repo (Any): Mongo-backed job repository from :func:`create_job_repo`.
repo (Any): Integer-id job repository from :func:`create_job_repo`.
Returns:
JobCrudsService: Ready-to-use job CRUD service.
"""
return build(JobCrudsService, instances={"repo": repo})
return build(JobCrudsService, instances={"repo": repo})

3
standards/src/templates/app/domain/parents/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Parents domain slice."""

39
standards/src/templates/app/domain/parents/models.py Normal file
View File

@@ -0,0 +1,39 @@
"""# Summary
Parents domain — pet owners referenced by :class:`domain.pets.models.Pet`.
"""
from pydantic import BaseModel
class Metadata(BaseModel):
"""Optional presentation metadata carried alongside a resource.
Attributes:
createdOn (str | None): ISO creation timestamp. Defaults to None.
updatedOn (str | None): ISO update timestamp. Defaults to None.
"""
createdOn: str | None = None
updatedOn: str | None = None
model_config = {"extra": "ignore"}
class Parent(BaseModel):
"""A pet owner.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
name (str): Parent's full name.
email (str): Email address.
phone (str | None): Contact phone number. Defaults to None.
metadata (Metadata | None): Presentation metadata. Defaults to None.
"""
id: int | None = None
name: str
email: str
phone: str | None = None
metadata: Metadata | None = None
model_config = {"extra": "ignore"}

3
standards/src/templates/app/domain/parents/ports/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Parents domain ports."""

27
standards/src/templates/app/domain/parents/ports/db.py Normal file
View File

@@ -0,0 +1,27 @@
"""# Summary
Parents persistence ports — integer-id document and repository factory.
"""
from mongo_ops import BaseDocument
from pydantic import Field
from config import settings
from domain.parents.models import Parent
from infra.db.repos import create_repo
class ParentDocument(Parent, BaseDocument):
"""Persisted parent document. The integer ``id`` shadows the ObjectId."""
id: int | None = Field(default=None, validation_alias="_id")
def create_parent_repo():
"""Create the parent repository selected by ``settings.repo_backend``.
Returns:
IntIdMemoryRepository | IntIdMongoRepository: Repository wired to the
``parents`` collection/store.
"""
_ = settings # backend chosen centrally by create_repo
return create_repo("parents", ParentDocument)

33
standards/src/templates/app/domain/parents/ports/service.py Normal file
View File

@@ -0,0 +1,33 @@
"""# Summary
CRUD service wiring for the parent aggregate via the hexa container.
"""
from hexa import build
from domain.parents.models import Parent
from domain.parents.ports.db import ParentDocument
from infra.adapters.service import CrudsServiceAdaptor
class ParentCrudsService(CrudsServiceAdaptor[Parent, ParentDocument]):
"""Parent CRUD service resolving :class:`ParentDocument` from a repo.
Notes:
Instances are created by :func:`get_parent_service`. The ``model`` and
``schema`` class attributes bind the domain model and its document type.
"""
model = Parent
schema = ParentDocument
def get_parent_service(repo) -> ParentCrudsService:
"""Build a parent CRUD service wired to ``repo``.
Args:
repo (Any): Integer-id parent repository from :func:`create_parent_repo`.
Returns:
ParentCrudsService: Ready-to-use parent CRUD service.
"""
return build(ParentCrudsService, instances={"repo": repo})

3
standards/src/templates/app/domain/pets/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Pets domain slice."""

37
standards/src/templates/app/domain/pets/models.py Normal file
View File

@@ -0,0 +1,37 @@
"""# Summary
Pets domain — animals owned by parents and treated in appointments.
"""
from typing import Literal
from pydantic import BaseModel
from domain.parents.models import Metadata
class Pet(BaseModel):
"""A pet patient.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
name (str): Name of the pet.
species (str): Type of animal (dog, cat, bird).
age (int | None): Age in years. Defaults to None.
weight (float | None): Weight in kilograms. Defaults to None.
birthDate (str | None): ISO date of birth. Defaults to None.
photo (str | None): Stored photo path after upload. Defaults to None.
parents (list[dict]): Embedded parent objects (or bare ids on input).
metadata (Metadata | None): Presentation metadata. Defaults to None.
"""
id: int | None = None
name: str
species: Literal["dog", "cat", "bird"]
age: int | None = None
weight: float | None = None
birthDate: str | None = None
photo: str | None = None
parents: list[dict] = []
metadata: Metadata | None = None
model_config = {"extra": "ignore"}

3
standards/src/templates/app/domain/pets/ports/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Pets domain ports."""

25
standards/src/templates/app/domain/pets/ports/db.py Normal file
View File

@@ -0,0 +1,25 @@
"""# Summary
Pets persistence ports — integer-id document and repository factory.
"""
from mongo_ops import BaseDocument
from pydantic import Field
from domain.pets.models import Pet
from infra.db.repos import create_repo
class PetDocument(Pet, BaseDocument):
"""Persisted pet document. The integer ``id`` shadows the ObjectId."""
id: int | None = Field(default=None, validation_alias="_id")
def create_pet_repo():
"""Create the pet repository selected by ``settings.repo_backend``.
Returns:
IntIdMemoryRepository | IntIdMongoRepository: Repository wired to the
``pets`` collection/store.
"""
return create_repo("pets", PetDocument)

33
standards/src/templates/app/domain/pets/ports/service.py Normal file
View File

@@ -0,0 +1,33 @@
"""# Summary
CRUD service wiring for the pet aggregate via the hexa container.
"""
from hexa import build
from domain.pets.models import Pet
from domain.pets.ports.db import PetDocument
from infra.adapters.service import CrudsServiceAdaptor
class PetCrudsService(CrudsServiceAdaptor[Pet, PetDocument]):
"""Pet CRUD service resolving :class:`PetDocument` from a repo.
Notes:
Instances are created by :func:`get_pet_service`. The ``model`` and
``schema`` class attributes bind the domain model and its document type.
"""
model = Pet
schema = PetDocument
def get_pet_service(repo) -> PetCrudsService:
"""Build a pet CRUD service wired to ``repo``.
Args:
repo (Any): Integer-id pet repository from :func:`create_pet_repo`.
Returns:
PetCrudsService: Ready-to-use pet CRUD service.
"""
return build(PetCrudsService, instances={"repo": repo})

3
standards/src/templates/app/domain/treatments/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Treatments domain slice."""

25
standards/src/templates/app/domain/treatments/models.py Normal file
View File

@@ -0,0 +1,25 @@
"""# Summary
Treatments domain — the catalogue of treatments offered by the clinic.
"""
from pydantic import BaseModel
from domain.parents.models import Metadata
class Treatment(BaseModel):
"""A catalogue treatment.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
label (str): Name of the treatment.
description (str | None): Detailed description. Defaults to None.
metadata (Metadata | None): Presentation metadata. Defaults to None.
"""
id: int | None = None
label: str
description: str | None = None
metadata: Metadata | None = None
model_config = {"extra": "ignore"}

3
standards/src/templates/app/domain/treatments/ports/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Treatments domain ports."""

25
standards/src/templates/app/domain/treatments/ports/db.py Normal file
View File

@@ -0,0 +1,25 @@
"""# Summary
Treatments persistence ports — integer-id document and repository factory.
"""
from mongo_ops import BaseDocument
from pydantic import Field
from domain.treatments.models import Treatment
from infra.db.repos import create_repo
class TreatmentDocument(Treatment, BaseDocument):
"""Persisted treatment document. The integer ``id`` shadows the ObjectId."""
id: int | None = Field(default=None, validation_alias="_id")
def create_treatment_repo():
"""Create the treatment repository selected by ``settings.repo_backend``.
Returns:
IntIdMemoryRepository | IntIdMongoRepository: Repository wired to the
``treatments`` collection/store.
"""
return create_repo("treatments", TreatmentDocument)

34
standards/src/templates/app/domain/treatments/ports/service.py Normal file
View File

@@ -0,0 +1,34 @@
"""# Summary
CRUD service wiring for the treatment aggregate via the hexa container.
"""
from hexa import build
from domain.treatments.models import Treatment
from domain.treatments.ports.db import TreatmentDocument
from infra.adapters.service import CrudsServiceAdaptor
class TreatmentCrudsService(CrudsServiceAdaptor[Treatment, TreatmentDocument]):
"""Treatment CRUD service resolving :class:`TreatmentDocument` from a repo.
Notes:
Instances are created by :func:`get_treatment_service`. The ``model``
and ``schema`` class attributes bind the domain model and its document.
"""
model = Treatment
schema = TreatmentDocument
def get_treatment_service(repo) -> TreatmentCrudsService:
"""Build a treatment CRUD service wired to ``repo``.
Args:
repo (Any): Integer-id treatment repository from
:func:`create_treatment_repo`.
Returns:
TreatmentCrudsService: Ready-to-use treatment CRUD service.
"""
return build(TreatmentCrudsService, instances={"repo": repo})

3
standards/src/templates/app/domain/vets/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Vets domain slice."""

29
standards/src/templates/app/domain/vets/models.py Normal file
View File

@@ -0,0 +1,29 @@
"""# Summary
Vets domain — veterinarians referenced by appointments.
"""
from pydantic import BaseModel
from domain.parents.models import Metadata
class Vet(BaseModel):
"""A veterinarian.
Attributes:
id (int | None): Server-assigned identifier. Defaults to None.
name (str): Veterinarian's full name.
specialty (str | None): Area of specialization. Defaults to None.
email (str | None): Email address. Defaults to None.
phone (str | None): Contact phone number. Defaults to None.
metadata (Metadata | None): Presentation metadata. Defaults to None.
"""
id: int | None = None
name: str
specialty: str | None = None
email: str | None = None
phone: str | None = None
metadata: Metadata | None = None
model_config = {"extra": "ignore"}

3
standards/src/templates/app/domain/vets/ports/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Vets domain ports."""

25
standards/src/templates/app/domain/vets/ports/db.py Normal file
View File

@@ -0,0 +1,25 @@
"""# Summary
Vets persistence ports — integer-id document and repository factory.
"""
from mongo_ops import BaseDocument
from pydantic import Field
from domain.vets.models import Vet
from infra.db.repos import create_repo
class VetDocument(Vet, BaseDocument):
"""Persisted vet document. The integer ``id`` shadows the ObjectId."""
id: int | None = Field(default=None, validation_alias="_id")
def create_vet_repo():
"""Create the vet repository selected by ``settings.repo_backend``.
Returns:
IntIdMemoryRepository | IntIdMongoRepository: Repository wired to the
``vets`` collection/store.
"""
return create_repo("vets", VetDocument)

33
standards/src/templates/app/domain/vets/ports/service.py Normal file
View File

@@ -0,0 +1,33 @@
"""# Summary
CRUD service wiring for the vet aggregate via the hexa container.
"""
from hexa import build
from domain.vets.models import Vet
from domain.vets.ports.db import VetDocument
from infra.adapters.service import CrudsServiceAdaptor
class VetCrudsService(CrudsServiceAdaptor[Vet, VetDocument]):
"""Vet CRUD service resolving :class:`VetDocument` from a repo.
Notes:
Instances are created by :func:`get_vet_service`. The ``model`` and
``schema`` class attributes bind the domain model and its document type.
"""
model = Vet
schema = VetDocument
def get_vet_service(repo) -> VetCrudsService:
"""Build a vet CRUD service wired to ``repo``.
Args:
repo (Any): Integer-id vet repository from :func:`create_vet_repo`.
Returns:
VetCrudsService: Ready-to-use vet CRUD service.
"""
return build(VetCrudsService, instances={"repo": repo})

88
standards/src/templates/app/infra/adapters/actions_stream.py Normal file
View File

@@ -0,0 +1,88 @@
"""# Summary
Species-scoped action stream — the SSE event source behind ``stream_actions``.
---
Every pet gets a background asyncio task that emits random, species-appropriate
behaviors (a wagging dog, a kneading cat, a chirping bird) onto per-subscriber
queues. The HTTP handler subscribes one queue per connected client and
unsubscribes when the response closes.
Notes:
Process-local and best-effort — the stream is generated, not persisted,
and is intentionally independent of the persistence backend.
"""
import asyncio
import json
import random
_BEHAVIORS_BY_SPECIES: dict[str, list[str]] = {
"dog": ["wagging_tail", "stretching", "showing_belly", "barking", "panting"],
"cat": ["blep", "stretching", "showing_belly", "purring", "kneading"],
"bird": ["head_tilt", "stretching", "chirping", "wing_flap"],
}
_subscribers: dict[int, list[asyncio.Queue]] = {}
_worker_tasks: dict[int, asyncio.Task] = {}
async def _behavior_worker(pet_id: int, species: str) -> None:
"""Emit random behaviors for one pet until cancelled.
Args:
pet_id (int): Pet whose subscribers receive the events.
species (str): Animal species selecting the behavior pool.
"""
behaviors = _BEHAVIORS_BY_SPECIES.get(species, ["wagging_tail"])
while True:
action = random.choice(behaviors)
data = json.dumps({"action": action})
for q in _subscribers.get(pet_id, []):
await q.put(data)
await asyncio.sleep(random.uniform(1, 5))
def _ensure_worker(pet_id: int, species: str) -> None:
"""Start the behavior worker for ``pet_id`` if it is not running.
Args:
pet_id (int): Pet identifier.
species (str): Animal species selecting the behavior pool.
"""
if pet_id not in _worker_tasks or _worker_tasks[pet_id].done():
_subscribers.setdefault(pet_id, [])
_worker_tasks[pet_id] = asyncio.create_task(_behavior_worker(pet_id, species))
async def subscribe(pet_id: int, species: str) -> asyncio.Queue:
"""Subscribe a new listener queue for a pet's action stream.
Args:
pet_id (int): Pet identifier.
species (str): Animal species selecting the behavior pool.
Returns:
asyncio.Queue: Queue that receives JSON-encoded action payloads.
"""
q: asyncio.Queue = asyncio.Queue()
_subscribers.setdefault(pet_id, []).append(q)
_ensure_worker(pet_id, species)
return q
def unsubscribe(pet_id: int, q: asyncio.Queue) -> None:
"""Remove a listener queue and stop the worker when the last one leaves.
Args:
pet_id (int): Pet identifier.
q (asyncio.Queue): Queue returned by :func:`subscribe`.
"""
queues = _subscribers.get(pet_id, [])
if q in queues:
queues.remove(q)
if not queues:
task = _worker_tasks.pop(pet_id, None)
if task and not task.done():
task.cancel()
_subscribers.pop(pet_id, None)

38
standards/src/templates/app/init.py
View File

@@ -2,8 +2,8 @@
Composition root — connects MongoDB (when selected), wires all repos and
services, and returns the resources dict. Shared bootstrap for the FastAPI
app, the worker and the seed/drop scripts, guarded against repeated
initialization via a module-level flag.
app and the seed/drop scripts, guarded against repeated initialization via a
module-level flag.
---
@@ -14,10 +14,18 @@ external services.
from typing import Any
from config import settings
from domain.item.ports.db import create_item_repo
from domain.item.ports.service import get_item_service
from domain.job.ports.db import create_job_repo
from domain.job.ports.service import get_job_service
from domain.appointments.ports.db import create_appointment_repo
from domain.appointments.ports.service import get_appointment_service
from domain.jobs.ports.db import create_job_repo
from domain.jobs.ports.service import get_job_service
from domain.parents.ports.db import create_parent_repo
from domain.parents.ports.service import get_parent_service
from domain.pets.ports.db import create_pet_repo
from domain.pets.ports.service import get_pet_service
from domain.treatments.ports.db import create_treatment_repo
from domain.treatments.ports.service import get_treatment_service
from domain.vets.ports.db import create_vet_repo
from domain.vets.ports.service import get_vet_service
from infra.db.setup import setup_database, shutdown_database
_resources: dict[str, Any] = {}
@@ -41,12 +49,24 @@ async def init() -> dict[str, Any]:
if settings.repo_backend == "mongo":
await setup_database(uri=settings.mongo_uri, db_name=settings.mongo_db)
item_repo = create_item_repo()
parent_repo = create_parent_repo()
vet_repo = create_vet_repo()
treatment_repo = create_treatment_repo()
pet_repo = create_pet_repo()
appointment_repo = create_appointment_repo()
job_repo = create_job_repo()
_resources["item_repo"] = item_repo
_resources["parent_repo"] = parent_repo
_resources["vet_repo"] = vet_repo
_resources["treatment_repo"] = treatment_repo
_resources["pet_repo"] = pet_repo
_resources["appointment_repo"] = appointment_repo
_resources["job_repo"] = job_repo
_resources["item_svc"] = get_item_service(item_repo)
_resources["parent_svc"] = get_parent_service(parent_repo)
_resources["vet_svc"] = get_vet_service(vet_repo)
_resources["treatment_svc"] = get_treatment_service(treatment_repo)
_resources["pet_svc"] = get_pet_service(pet_repo)
_resources["appointment_svc"] = get_appointment_service(appointment_repo)
_resources["job_svc"] = get_job_service(job_repo)
_initialized = True

3
standards/src/templates/app/interfaces/appointments/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Appointments interfaces."""

135
standards/src/templates/app/interfaces/appointments/routes.py Normal file
View File

@@ -0,0 +1,135 @@
"""# Summary
Appointments route handlers — names must match openapi.yaml operationIds.
Notes:
``pet``, ``vet`` and ``treatment`` are foreign keys expanded to full
objects on read and normalized from ids on write. The list endpoint
supports date-prefix and pet/vet id filters.
"""
from fastapi import Request, Response
from interfaces._common import (
delete_or_404,
dump,
expand_ref,
paginate,
read_or_404,
svc,
)
async def list_appointments(
request: Request,
limit: int = 20,
offset: int = 0,
date: str | None = None,
vet: int | None = None,
pet: int | None = None,
) -> dict:
"""List appointments (paginated, filterable by date/vet/pet).
Args:
request (Request): Incoming request.
limit (int): Page size. Defaults to 20.
offset (int): Items to skip. Defaults to 0.
date (str | None): ISO date prefix to match (e.g. ``2025-06-01``).
vet (int | None): Filter by embedded vet id.
pet (int | None): Filter by embedded pet id.
Returns:
dict: ``{"total": int, "items": list[Appointment]}``.
"""
docs = await svc(request, "appointment").search({})
items = []
for d in docs:
row = dump(d)
if date and not str(row.get("date", "")).startswith(date):
continue
if vet is not None and (row.get("vet") or {}).get("id") != vet:
continue
if pet is not None and (row.get("pet") or {}).get("id") != pet:
continue
items.append(row)
return paginate(items, limit, offset)
async def create_appointment(request: Request, body: dict, response: Response) -> dict:
"""Create an appointment; FK fields may be bare ids or expanded objects.
Args:
request (Request): Incoming request.
body (dict): Appointment fields (``id`` ignored; FKs normalized).
response (Response): Response object; set to 201 on success.
Returns:
dict: The created appointment with FKs expanded.
Raises:
HTTPException: 422 when an FK reference cannot be resolved.
"""
response.status_code = 201
body.pop("id", None)
body["pet"] = await expand_ref(svc(request, "pet"), body.get("pet"), "pet")
body["vet"] = await expand_ref(svc(request, "vet"), body.get("vet"), "vet")
body["treatment"] = await expand_ref(
svc(request, "treatment"), body.get("treatment"), "treatment"
)
doc = await svc(request, "appointment").create(body)
return dump(doc)
async def get_appointment(request: Request, id: int) -> dict:
"""Fetch one appointment by id.
Args:
request (Request): Incoming request.
id (int): Appointment identifier.
Returns:
dict: The appointment document.
Raises:
HTTPException: 404 when the appointment does not exist.
"""
return dump(await read_or_404(svc(request, "appointment"), id, "appointment"))
async def update_appointment(request: Request, id: int, body: dict) -> dict:
"""Replace an appointment by id; FK fields may be ids or objects.
Args:
request (Request): Incoming request.
id (int): Appointment identifier.
body (dict): Full replacement payload.
Returns:
dict: The updated appointment with FKs expanded.
Raises:
HTTPException: 404 when the appointment does not exist; 422 on bad FKs.
"""
await read_or_404(svc(request, "appointment"), id, "appointment")
body.pop("id", None)
body["pet"] = await expand_ref(svc(request, "pet"), body.get("pet"), "pet")
body["vet"] = await expand_ref(svc(request, "vet"), body.get("vet"), "vet")
body["treatment"] = await expand_ref(
svc(request, "treatment"), body.get("treatment"), "treatment"
)
doc = await svc(request, "appointment").update(id, body)
return dump(doc)
async def delete_appointment(request: Request, id: int, response: Response) -> None:
"""Delete an appointment by id.
Args:
request (Request): Incoming request.
id (int): Appointment identifier.
response (Response): Response object; set to 204 on success.
Raises:
HTTPException: 404 when the appointment does not exist.
"""
await delete_or_404(svc(request, "appointment"), id, "appointment")
response.status_code = 204

0
standards/src/templates/app/interfaces/item/__init__.py
View File

90
standards/src/templates/app/interfaces/item/routes.py
View File

@@ -1,90 +0,0 @@
"""# Summary
Item route handlers — names must match openapi.yaml operationIds.
"""
from fastapi import Request, Response
from interfaces._common import delete_or_404, dump, paginate, read_or_404, svc
async def list_items(request: Request, limit: int = 20, offset: int = 0) -> dict:
"""List items (paginated). ``limit=0`` returns all remaining items.
Args:
request (Request): Incoming request.
limit (int): Page size. Defaults to 20.
offset (int): Items to skip. Defaults to 0.
Returns:
dict: ``{"total": int, "items": list[Item]}``.
"""
docs = await svc(request, "item").search({})
return paginate([dump(d) for d in docs], limit, offset)
async def create_item(request: Request, body: dict, response: Response) -> dict:
"""Create an item.
Args:
request (Request): Incoming request.
body (dict): Item fields (``id`` ignored).
response (Response): Response object; set to 201 on success.
Returns:
dict: The created item.
"""
response.status_code = 201
body.pop("id", None)
doc = await svc(request, "item").create(body)
return dump(doc)
async def get_item(request: Request, id: int) -> dict:
"""Fetch one item by id.
Args:
request (Request): Incoming request.
id (int): Item identifier.
Returns:
dict: The item document.
Raises:
HTTPException: 404 when the item does not exist.
"""
return dump(await read_or_404(svc(request, "item"), id, "item"))
async def update_item(request: Request, id: int, body: dict) -> dict:
"""Replace an item by id.
Args:
request (Request): Incoming request.
id (int): Item identifier.
body (dict): Full replacement payload.
Returns:
dict: The updated item.
Raises:
HTTPException: 404 when the item does not exist.
"""
await read_or_404(svc(request, "item"), id, "item")
body.pop("id", None)
doc = await svc(request, "item").update(id, body)
return dump(doc)
async def delete_item(request: Request, id: int, response: Response) -> None:
"""Delete an item by id.
Args:
request (Request): Incoming request.
id (int): Item identifier.
response (Response): Response object; set to 204 on success.
Raises:
HTTPException: 404 when the item does not exist.
"""
await delete_or_404(svc(request, "item"), id, "item")
response.status_code = 204

0
standards/src/templates/app/interfaces/job/__init__.py
View File

0
standards/src/templates/app/domain/job/__init__.py → standards/src/templates/app/interfaces/jobs/__init__.py
View File

12
standards/src/templates/app/interfaces/job/routes.py → standards/src/templates/app/interfaces/jobs/routes.py
View File

@@ -1,25 +1,29 @@
"""# Summary
Job route handlers — names must match openapi.yaml operationIds.
Jobs route handlers — names must match openapi.yaml operationIds.
"""
from fastapi import Request, Response
from interfaces._common import delete_or_404, dump, paginate, read_or_404, svc
async def list_jobs(request: Request, limit: int = 20, offset: int = 0) -> dict:
"""List jobs (paginated). ``limit=0`` returns all remaining items.
async def list_jobs(request: Request, limit: int = 20, offset: int = 0, status: str | None = None) -> dict:
"""List jobs (paginated, optionally filtered by status).
Args:
request (Request): Incoming request.
limit (int): Page size. Defaults to 20.
offset (int): Items to skip. Defaults to 0.
status (str | None): Filter by status (pending/processing/done/failed).
Returns:
dict: ``{"total": int, "items": list[Job]}``.
"""
docs = await svc(request, "job").search({})
return paginate([dump(d) for d in docs], limit, offset)
items = [dump(d) for d in docs]
if status:
items = [row for row in items if row.get("status") == status]
return paginate(items, limit, offset)
async def create_job(request: Request, body: dict, response: Response) -> dict:

8
standards/src/templates/app/interfaces/openapi.py
View File

@@ -5,6 +5,10 @@ OpenAPI-first router resolves each operation to a handler by name, so this modul
re-exports every route handler from the subpackages into one flat namespace.
"""
# ruff: noqa: F403
from interfaces.item.routes import *
from interfaces.job.routes import *
from interfaces.appointments.routes import *
from interfaces.jobs.routes import *
from interfaces.parents.routes import *
from interfaces.pets.routes import *
from interfaces.system.routes import *
from interfaces.treatments.routes import *
from interfaces.vets.routes import *

3
standards/src/templates/app/interfaces/parents/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Parents interfaces."""

90
standards/src/templates/app/interfaces/parents/routes.py Normal file
View File

@@ -0,0 +1,90 @@
"""# Summary
Parents route handlers — names must match openapi.yaml operationIds.
"""
from fastapi import Request, Response
from interfaces._common import delete_or_404, dump, paginate, read_or_404, svc
async def list_parents(request: Request, limit: int = 20, offset: int = 0) -> dict:
"""List parents (paginated). ``limit=0`` returns all remaining items.
Args:
request (Request): Incoming request.
limit (int): Page size. Defaults to 20.
offset (int): Items to skip. Defaults to 0.
Returns:
dict: ``{"total": int, "items": list[Parent]}``.
"""
docs = await svc(request, "parent").search({})
return paginate([dump(d) for d in docs], limit, offset)
async def create_parent(request: Request, body: dict, response: Response) -> dict:
"""Create a parent.
Args:
request (Request): Incoming request.
body (dict): Parent fields (``id`` ignored).
response (Response): Response object; set to 201 on success.
Returns:
dict: The created parent.
"""
response.status_code = 201
body.pop("id", None)
doc = await svc(request, "parent").create(body)
return dump(doc)
async def get_parent(request: Request, id: int) -> dict:
"""Fetch one parent by id.
Args:
request (Request): Incoming request.
id (int): Parent identifier.
Returns:
dict: The parent document.
Raises:
HTTPException: 404 when the parent does not exist.
"""
return dump(await read_or_404(svc(request, "parent"), id, "parent"))
async def update_parent(request: Request, id: int, body: dict) -> dict:
"""Replace a parent by id.
Args:
request (Request): Incoming request.
id (int): Parent identifier.
body (dict): Full replacement payload.
Returns:
dict: The updated parent.
Raises:
HTTPException: 404 when the parent does not exist.
"""
await read_or_404(svc(request, "parent"), id, "parent")
body.pop("id", None)
doc = await svc(request, "parent").update(id, body)
return dump(doc)
async def delete_parent(request: Request, id: int, response: Response) -> None:
"""Delete a parent by id.
Args:
request (Request): Incoming request.
id (int): Parent identifier.
response (Response): Response object; set to 204 on success.
Raises:
HTTPException: 404 when the parent does not exist.
"""
await delete_or_404(svc(request, "parent"), id, "parent")
response.status_code = 204

3
standards/src/templates/app/interfaces/pets/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Pets interfaces."""

172
standards/src/templates/app/interfaces/pets/routes.py Normal file
View File

@@ -0,0 +1,172 @@
"""# Summary
Pets route handlers — names must match openapi.yaml operationIds.
Notes:
``parents`` is a list foreign key expanded to full objects on read and
normalized from ids on write. ``upload_pet_photo`` stores the raw bytes
under ``settings.uploads_dir``; ``stream_actions`` returns an SSE stream
scoped to the pet's species.
"""
import os
from collections.abc import AsyncIterator
from fastapi import Request, Response, UploadFile
from fastapi.responses import StreamingResponse
from config import settings
from infra.adapters import actions_stream
from interfaces._common import (
delete_or_404,
dump,
expand_ref_list,
paginate,
read_or_404,
svc,
)
async def list_pets(request: Request, limit: int = 20, offset: int = 0) -> dict:
"""List pets (paginated). ``limit=0`` returns all remaining items.
Args:
request (Request): Incoming request.
limit (int): Page size. Defaults to 20.
offset (int): Items to skip. Defaults to 0.
Returns:
dict: ``{"total": int, "items": list[Pet]}``.
"""
docs = await svc(request, "pet").search({})
return paginate([dump(d) for d in docs], limit, offset)
async def create_pet(request: Request, body: dict, response: Response) -> dict:
"""Create a pet; parents may be bare ids or expanded objects.
Args:
request (Request): Incoming request.
body (dict): Pet fields (``id`` ignored; ``parents`` normalized).
response (Response): Response object; set to 201 on success.
Returns:
dict: The created pet with parents expanded.
Raises:
HTTPException: 422 when a parent reference cannot be resolved.
"""
response.status_code = 201
body.pop("id", None)
body["parents"] = await expand_ref_list(
svc(request, "parent"), body.get("parents"), "parent"
)
doc = await svc(request, "pet").create(body)
return dump(doc)
async def get_pet(request: Request, id: int) -> dict:
"""Fetch one pet by id.
Args:
request (Request): Incoming request.
id (int): Pet identifier.
Returns:
dict: The pet document.
Raises:
HTTPException: 404 when the pet does not exist.
"""
return dump(await read_or_404(svc(request, "pet"), id, "pet"))
async def update_pet(request: Request, id: int, body: dict) -> dict:
"""Replace a pet by id; parents may be bare ids or expanded objects.
Args:
request (Request): Incoming request.
id (int): Pet identifier.
body (dict): Full replacement payload.
Returns:
dict: The updated pet with parents expanded.
Raises:
HTTPException: 404 when the pet does not exist; 422 on a bad parent ref.
"""
await read_or_404(svc(request, "pet"), id, "pet")
body.pop("id", None)
body["parents"] = await expand_ref_list(
svc(request, "parent"), body.get("parents"), "parent"
)
doc = await svc(request, "pet").update(id, body)
return dump(doc)
async def delete_pet(request: Request, id: int, response: Response) -> None:
"""Delete a pet by id.
Args:
request (Request): Incoming request.
id (int): Pet identifier.
response (Response): Response object; set to 204 on success.
Raises:
HTTPException: 404 when the pet does not exist.
"""
await delete_or_404(svc(request, "pet"), id, "pet")
response.status_code = 204
async def upload_pet_photo(request: Request, id: int, file: UploadFile) -> dict:
"""Store an uploaded photo for a pet and remember its path.
Args:
request (Request): Incoming request.
id (int): Pet identifier.
file (UploadFile): Multipart upload carrying the photo bytes.
Returns:
dict: ``{"id": int, "status": "photo_uploaded", "path": str}``.
Raises:
HTTPException: 404 when the pet does not exist.
"""
pet = await read_or_404(svc(request, "pet"), id, "pet")
os.makedirs(settings.uploads_dir, exist_ok=True)
safe_name = os.path.basename(file.filename or "photo")
path = os.path.join(settings.uploads_dir, f"{id}_{safe_name}")
content = await file.read()
with open(path, "wb") as fh:
fh.write(content)
await svc(request, "pet").update(id, {"photo": path, "parents": pet.parents})
return {"id": id, "status": "photo_uploaded", "path": path}
async def stream_actions(request: Request, id: int) -> StreamingResponse:
"""Stream random, species-appropriate pet actions over SSE.
Args:
request (Request): Incoming request.
id (int): Pet identifier whose species scopes the stream.
Returns:
StreamingResponse: ``text/event-stream`` emitting ``data: {action}``.
Raises:
HTTPException: 404 when the pet does not exist.
"""
pet = await read_or_404(svc(request, "pet"), id, "pet")
async def event_gen() -> AsyncIterator[str]:
queue = await actions_stream.subscribe(id, pet.species)
try:
while True:
if await request.is_disconnected():
break
data = await queue.get()
yield f"data: {data}\n\n"
finally:
actions_stream.unsubscribe(id, queue)
return StreamingResponse(event_gen(), media_type="text/event-stream")

4
standards/src/templates/app/interfaces/system/routes.py
View File

@@ -8,6 +8,6 @@ async def health() -> dict:
"""Liveness probe — public, unauthenticated.
Returns:
dict: ``{"status": "ok", "service": "app-template"}``.
dict: ``{"status": "ok", "service": "vet-server"}``.
"""
return {"status": "ok", "service": "app-template"}
return {"status": "ok", "service": "vet-server"}

3
standards/src/templates/app/interfaces/treatments/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Treatments interfaces."""

92
standards/src/templates/app/interfaces/treatments/routes.py Normal file
View File

@@ -0,0 +1,92 @@
"""# Summary
Treatments route handlers — names must match openapi.yaml operationIds.
Notes:
``list_treatments`` returns a plain array (the catalogue is small and
unpaginated by contract).
"""
from fastapi import Request, Response
from interfaces._common import delete_or_404, dump, read_or_404, svc
async def list_treatments(request: Request) -> list[dict]:
"""List treatments (catalogue, unpaginated).
Args:
request (Request): Incoming request.
Returns:
list[dict]: All treatment documents.
"""
docs = await svc(request, "treatment").search({})
return [dump(d) for d in docs]
async def create_treatment(request: Request, body: dict, response: Response) -> dict:
"""Add a treatment (admin only).
Args:
request (Request): Incoming request.
body (dict): Treatment fields (``id`` ignored).
response (Response): Response object; set to 201 on success.
Returns:
dict: The created treatment.
"""
response.status_code = 201
body.pop("id", None)
doc = await svc(request, "treatment").create(body)
return dump(doc)
async def get_treatment(request: Request, id: int) -> dict:
"""Fetch one treatment by id.
Args:
request (Request): Incoming request.
id (int): Treatment identifier.
Returns:
dict: The treatment document.
Raises:
HTTPException: 404 when the treatment does not exist.
"""
return dump(await read_or_404(svc(request, "treatment"), id, "treatment"))
async def update_treatment(request: Request, id: int, body: dict) -> dict:
"""Replace a treatment by id.
Args:
request (Request): Incoming request.
id (int): Treatment identifier.
body (dict): Full replacement payload.
Returns:
dict: The updated treatment.
Raises:
HTTPException: 404 when the treatment does not exist.
"""
await read_or_404(svc(request, "treatment"), id, "treatment")
body.pop("id", None)
doc = await svc(request, "treatment").update(id, body)
return dump(doc)
async def delete_treatment(request: Request, id: int, response: Response) -> None:
"""Delete a treatment by id.
Args:
request (Request): Incoming request.
id (int): Treatment identifier.
response (Response): Response object; set to 204 on success.
Raises:
HTTPException: 404 when the treatment does not exist.
"""
await delete_or_404(svc(request, "treatment"), id, "treatment")
response.status_code = 204

3
standards/src/templates/app/interfaces/vets/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""# Summary
Vets interfaces."""

90
standards/src/templates/app/interfaces/vets/routes.py Normal file
View File

@@ -0,0 +1,90 @@
"""# Summary
Vets route handlers — names must match openapi.yaml operationIds.
"""
from fastapi import Request, Response
from interfaces._common import delete_or_404, dump, paginate, read_or_404, svc
async def list_vets(request: Request, limit: int = 20, offset: int = 0) -> dict:
"""List vets (paginated). ``limit=0`` returns all remaining items.
Args:
request (Request): Incoming request.
limit (int): Page size. Defaults to 20.
offset (int): Items to skip. Defaults to 0.
Returns:
dict: ``{"total": int, "items": list[Vet]}``.
"""
docs = await svc(request, "vet").search({})
return paginate([dump(d) for d in docs], limit, offset)
async def create_vet(request: Request, body: dict, response: Response) -> dict:
"""Create a vet.
Args:
request (Request): Incoming request.
body (dict): Vet fields (``id`` ignored).
response (Response): Response object; set to 201 on success.
Returns:
dict: The created vet.
"""
response.status_code = 201
body.pop("id", None)
doc = await svc(request, "vet").create(body)
return dump(doc)
async def get_vet(request: Request, id: int) -> dict:
"""Fetch one vet by id.
Args:
request (Request): Incoming request.
id (int): Vet identifier.
Returns:
dict: The vet document.
Raises:
HTTPException: 404 when the vet does not exist.
"""
return dump(await read_or_404(svc(request, "vet"), id, "vet"))
async def update_vet(request: Request, id: int, body: dict) -> dict:
"""Replace a vet by id.
Args:
request (Request): Incoming request.
id (int): Vet identifier.
body (dict): Full replacement payload.
Returns:
dict: The updated vet.
Raises:
HTTPException: 404 when the vet does not exist.
"""
await read_or_404(svc(request, "vet"), id, "vet")
body.pop("id", None)
doc = await svc(request, "vet").update(id, body)
return dump(doc)
async def delete_vet(request: Request, id: int, response: Response) -> None:
"""Delete a vet by id.
Args:
request (Request): Incoming request.
id (int): Vet identifier.
response (Response): Response object; set to 204 on success.
Raises:
HTTPException: 404 when the vet does not exist.
"""
await delete_or_404(svc(request, "vet"), id, "vet")
response.status_code = 204

2
standards/src/templates/app/main.py
View File

@@ -52,7 +52,7 @@ async def lifespan(app: OpenAPIFirstApp):
app = OpenAPIFirstApp(
openapi_path="openapi.yaml",
routes_module=openapi_routes,
title="App Template API",
title="Veterinary Clinic API",
version="0.1.0",
lifespan=lifespan,
)

1130
standards/src/templates/app/openapi.yaml
View File

File diff suppressed because it is too large Load Diff

0
standards/src/templates/app/domain/job/ports/__init__.py → standards/src/templates/app/pipelines/appointment_reminder/__init__.py
View File

50
standards/src/templates/app/pipelines/appointment_reminder/abc.py Normal file
View File

@@ -0,0 +1,50 @@
"""# 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."""

112
standards/src/templates/app/pipelines/appointment_reminder/impl.py Normal file
View File

@@ -0,0 +1,112 @@
"""# 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)

15
standards/src/templates/app/pipelines/appointment_reminder/pipeline.yaml Normal file
View File

@@ -0,0 +1,15 @@
# Graph spec for the appointment-reminder family: nodes, edges, and the root.
# Mirrors the port tree in abc.py.
schema: Reminder
nodes:
loader:
port: LoadPort
compose:
port: ComposePort
deliver:
port: DeliverPort
edges:
- loader -> compose
- compose -> deliver
roots:
default: AppointmentReminderPipeline

5
standards/src/templates/app/pipelines/appointment_reminder/resources/patterns/reminder.yaml Normal file
View File

@@ -0,0 +1,5 @@
# Pattern / rule fragments for the appointment-reminder family.
# The loader merges every patterns/*.yaml into one registry by key.
reminder:
subject: "Appointment reminder"
body: "Hi {parent_name}, {pet_name} has an appointment on {date}."

2
standards/src/templates/app/pipelines/example/resources/pipeline.yaml → standards/src/templates/app/pipelines/appointment_reminder/resources/pipeline.yaml
View File

@@ -1,5 +1,5 @@
# Runtime spec consumed by the shared loader.
kind: example
kind: appointment_reminder
variant: default
mode: default
patterns: []

7
standards/src/templates/app/pipelines/example/__init__.py
View File

@@ -1,7 +0,0 @@
"""# Summary
Example pipeline family — a reusable skeleton, not an app-specific pipeline.
Replace ``Example*`` names and the three stages with your family's stages. The
shape (abc → impl → pipeline.yaml → resources/) stays the same.
"""

53
standards/src/templates/app/pipelines/example/abc.py
View File

@@ -1,53 +0,0 @@
"""# Summary
Example pipeline family — port contracts (stage interfaces + root interface).
Ports are declared as a DI tree: the root's annotated slots *are* the dependency
decisions. The concrete subclass in ``impl.py`` pins the stage types.
"""
from abc import ABC, abstractmethod
from typing import Any
from pipelines.shared.ports import FinalizePort
class IngestPort(ABC):
"""Obtain the raw input as text/bytes."""
@abstractmethod
def materialize(self, source: Any) -> str:
"""Return the raw input materialized as text."""
class ReadPort(ABC):
"""Parse materialized input into records."""
@abstractmethod
def read(self, content: str) -> list[dict[str, Any]]:
"""Return parsed records."""
class NormalizePort(ABC):
"""Normalize raw records into the target shape."""
@abstractmethod
def build(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Return normalized records."""
class ExamplePipelinePort(ABC):
"""Root of the example family.
Notes:
The slot annotations below are contracts. A concrete subclass re-annotates
them with stage implementations; the DI container instantiates the tree.
"""
ingest: IngestPort
reader: ReadPort
normalize: NormalizePort
finalize: FinalizePort
@abstractmethod
async def run(self, source: Any) -> list[dict[str, Any]]:
"""Run the full stage chain and return the normalized records."""

71
standards/src/templates/app/pipelines/example/impl.py
View File

@@ -1,71 +0,0 @@
"""# Summary
Example pipeline family — concrete stages, the wired root, and its registration.
The concrete subclass re-annotates each slot with a stage implementation; the DI
container builds the whole tree. Register the family so the shared registry can
resolve it by ``(kind, variant, mode)``.
"""
from pathlib import Path
from typing import Any
from pipelines.example.abc import (
ExamplePipelinePort,
IngestPort,
NormalizePort,
ReadPort,
)
from pipelines.shared.loader import build_pipeline
from pipelines.shared.ports import FinalizePort
from pipelines.shared.registry import register
class TextIngest(IngestPort):
"""Materialize any source as text."""
def materialize(self, source: Any) -> str:
return source if isinstance(source, str) else str(source)
class LineReader(ReadPort):
"""Read one record per non-empty line."""
def read(self, content: str) -> list[dict[str, Any]]:
return [{"line": line} for line in content.splitlines() if line.strip()]
class PassthroughNormalize(NormalizePort):
"""Return records unchanged (replace with real normalization)."""
def build(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]:
return records
class NoopFinalize(FinalizePort):
"""Drop records (replace with a service-backed finalize stage)."""
async def finalize(self, records: list[dict[str, Any]]) -> list[Any]:
return records
class ExamplePipeline(ExamplePipelinePort):
"""Concrete example family — the dependency decisions live here."""
ingest: TextIngest
reader: LineReader
normalize: PassthroughNormalize
finalize: NoopFinalize
async def run(self, source: Any) -> list[dict[str, Any]]:
content = self.ingest.materialize(source)
records = self.reader.read(content)
return self.normalize.build(records)
def get_example_pipeline() -> ExamplePipeline:
"""Wire and return the example pipeline through the shared loader."""
resources = Path(__file__).parent / "resources"
return build_pipeline(ExamplePipeline, resources_dir=resources)
register("example", "default", ExamplePipeline)

18
standards/src/templates/app/pipelines/example/pipeline.yaml
View File

@@ -1,18 +0,0 @@
# Graph spec for the example family: nodes, edges, and the root.
# Mirrors the port tree in abc.py. Replace with your family's graph.
schema: ExampleRecord
nodes:
ingest:
port: IngestPort
reader:
port: ReadPort
normalize:
port: NormalizePort
finalize:
port: FinalizePort
edges:
- ingest -> reader
- reader -> normalize
- normalize -> finalize
roots:
default: ExamplePipeline

4
standards/src/templates/app/pipelines/example/resources/patterns/example.yaml
View File

@@ -1,4 +0,0 @@
# Pattern / rule fragments for the example family.
# The loader merges every patterns/*.yaml into one registry by key.
example:
enabled: true

32
standards/src/templates/app/pipelines/item_sync.py
View File

@@ -1,32 +0,0 @@
"""# 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
"50_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))

15
standards/src/templates/app/scripts/seed.py
View File

@@ -12,12 +12,19 @@ from init import init, shutdown
async def main() -> None:
"""Seed one demo item and shut down cleanly."""
"""Seed minimal demo records and shut down cleanly."""
resources = await init()
try:
item_svc = resources["item_svc"]
await item_svc.create({"name": "seed", "quantity": 1})
print("Seeded one item.")
await resources["parent_svc"].create(
{"name": "Alice Parker", "email": "alice@vet.example"}
)
await resources["vet_svc"].create(
{"name": "Dr. Chen", "specialty": "Canine", "email": "chen@vet.example"}
)
await resources["treatment_svc"].create(
{"label": "Checkup", "description": "Routine wellness exam"}
)
print("Seeded demo parents/vets/treatments.")
finally:
await shutdown()

32
standards/src/templates/app/tests/integration/test_item_routes.py
View File

@@ -1,32 +0,0 @@
import importlib
from openapi_first.app import OpenAPIFirstApp
import interfaces.openapi as routes_module
def test_openapi_app_builds():
"""The spec-first app must build and plant each route handler."""
module = importlib.import_module(routes_module.__name__)
for operation in (
"health",
"list_items",
"create_item",
"get_item",
"update_item",
"delete_item",
"list_jobs",
"create_job",
"get_job",
"update_job",
"delete_job",
):
assert hasattr(module, operation)
app = OpenAPIFirstApp(
openapi_path="openapi.yaml",
routes_module=module,
title="App Template API",
version="0.1.0",
)
assert app is not None

79
standards/src/templates/app/tests/unit/test_appointment_reminder_pipeline.py Normal file
View File

@@ -0,0 +1,79 @@
"""Unit tests for the appointment-reminder pipeline family."""
from pathlib import Path
from pipelines.appointment_reminder.impl import (
AppointmentReminderPipeline,
ReminderComposer,
get_appointment_reminder_pipeline,
)
from pipelines.shared.loader import build_pipeline, load_patterns
RESOURCES = Path(__file__).resolve().parents[2] / "pipelines" / "appointment_reminder" / "resources"
class FakeDoc:
def __init__(self, data):
self._data = data
def model_dump(self, **kwargs):
return self._data
class FakeAppointmentSvc:
def __init__(self, data):
self._data = data
async def read(self, id):
return FakeDoc(self._data) if self._data else None
def _appointment():
return {
"id": 1,
"date": "2026-11-02T10:00:00",
"pet": {"name": "Rex", "species": "dog", "parents": [{"name": "Kim", "email": "kim@example.com"}]},
"vet": {"name": "Dr. Patel"},
}
def test_load_patterns_merges_fragments():
registry = load_patterns(RESOURCES)
assert registry["reminder"]["subject"] == "Appointment reminder"
assert "{pet_name}" in registry["reminder"]["body"]
def test_composer_uses_pattern_and_context():
composer = ReminderComposer()
composer.registry = load_patterns(RESOURCES)
reminder = composer.compose(_appointment(), _appointment()["pet"], _appointment()["pet"]["parents"][0])
assert reminder["to"] == "kim@example.com"
assert reminder["subject"] == "Appointment reminder"
assert "Rex" in reminder["body"] and "Kim" in reminder["body"]
assert "2026-11-02" in reminder["body"]
async def test_full_pipeline_delivers_reminder():
pipeline = get_appointment_reminder_pipeline(FakeAppointmentSvc(_appointment()))
assert isinstance(pipeline, AppointmentReminderPipeline)
result = await pipeline.run(1)
assert result["delivered"] == "log"
assert result["to"] == "kim@example.com"
async def test_pipeline_errors_on_missing_appointment():
pipeline = get_appointment_reminder_pipeline(FakeAppointmentSvc(None))
try:
await pipeline.run(99)
except ValueError as exc:
assert "not found" in str(exc)
else:
raise AssertionError("expected ValueError for missing appointment")
def test_build_pipeline_injects_registry():
pipeline = build_pipeline(
AppointmentReminderPipeline,
resources_dir=RESOURCES,
instances={"appointment_svc": FakeAppointmentSvc(_appointment())},
)
assert pipeline.compose.registry["reminder"]["subject"] == "Appointment reminder"

22
standards/src/templates/app/tests/unit/test_item_models.py
View File

@@ -1,22 +0,0 @@
from domain.item.models import Item
from domain.job.models import Job
def test_item_defaults():
item = Item(name="widget")
assert item.quantity == 1
def test_job_defaults():
job = Job(name="sync")
assert job.status == "pending"
assert job.retry_count == 0
assert job.error_message is None
def test_job_rejects_unknown_status():
try:
Job(name="sync", status="exploded")
except Exception:
return
raise AssertionError("status should be rejected")

28
standards/src/templates/app/tests/unit/test_item_sync.py
View File

@@ -1,28 +0,0 @@
from types import SimpleNamespace
from pipelines.item_sync import run_item_sync
class FakeItemSvc:
def __init__(self):
self.created = []
async def create(self, data: dict):
self.created.append(dict(data))
async def test_run_item_sync_creates_item():
svc = FakeItemSvc()
job = SimpleNamespace(id="j1", name="widget", payload={"quantity": 4})
await run_item_sync(job, {"item_svc": svc})
assert len(svc.created) == 1
assert svc.created[0] == {"name": "widget", "quantity": 4}
async def test_run_item_sync_uses_payload_name():
svc = FakeItemSvc()
job = SimpleNamespace(id="j1", name="ignored", payload={"name": "gadget"})
await run_item_sync(job, {"item_svc": svc})
assert svc.created[0]["name"] == "gadget"

100
standards/src/templates/app/tests/unit/test_jobs_routes.py Normal file
View File

@@ -0,0 +1,100 @@
"""Unit tests for the jobs interface helpers and routes (memory backend)."""
from types import SimpleNamespace
from interfaces.jobs.routes import (
create_job,
delete_job,
get_job,
list_jobs,
update_job,
)
class FakeDoc:
"""Minimal document exposing the pydantic dump interface."""
def __init__(self, **fields):
self.__dict__.update(fields)
def model_dump(self, **kwargs):
return dict(self.__dict__)
class FakeJobSvc:
"""In-memory job service double."""
def __init__(self):
self.store: dict[int, FakeDoc] = {}
self._next = 1
async def search(self, filter=None):
return list(self.store.values())
async def create(self, body):
body.setdefault("status", "pending")
body.setdefault("retry_count", 0)
doc = FakeDoc(id=self._next, **body)
self.store[self._next] = doc
self._next += 1
return doc
async def read(self, id):
return self.store.get(id)
async def update(self, id, body):
doc = FakeDoc(id=id, **body)
self.store[id] = doc
return doc
async def delete(self, id):
return self.store.pop(id, None) is not None
def _request(svc_obj):
state = SimpleNamespace(db_resources={"job_svc": svc_obj})
return SimpleNamespace(app=SimpleNamespace(state=state))
async def test_create_list_get_update_delete_job():
svc = FakeJobSvc()
request = _request(svc)
response = SimpleNamespace(status_code=200)
created = await create_job(
request, {"name": "remind", "kind": "appointment_reminder"}, response
)
assert response.status_code == 201
assert created["kind"] == "appointment_reminder"
assert created["status"] == "pending"
page = await list_jobs(request)
assert page["total"] == 1
assert page["items"][0]["name"] == "remind"
one = await get_job(request, created["id"])
assert one["id"] == created["id"]
updated = await update_job(
request, created["id"], {"name": "remind-2", "kind": "appointment_reminder"}
)
assert updated["name"] == "remind-2"
response = SimpleNamespace(status_code=200)
await delete_job(request, created["id"], response)
assert response.status_code == 204
assert (await list_jobs(request))["total"] == 0
async def test_list_jobs_filters_by_status():
svc = FakeJobSvc()
request = _request(svc)
await create_job(request, {"name": "a", "kind": "k"}, SimpleNamespace())
await create_job(request, {"name": "b", "kind": "k", "status": "done"}, SimpleNamespace())
pending = await list_jobs(request, status="pending")
assert pending["total"] == 1
assert pending["items"][0]["name"] == "a"
done = await list_jobs(request, status="done")
assert done["total"] == 1
assert done["items"][0]["name"] == "b"

121
standards/src/templates/app/tests/unit/test_worker.py Normal file
View File

@@ -0,0 +1,121 @@
"""Unit tests for the worker dispatch table and job status transitions."""
from worker import HANDLERS, poll_jobs
class FakeDoc:
def __init__(self, **fields):
self.__dict__.update(fields)
def model_dump(self, **kwargs):
return dict(self.__dict__)
class FakeJobSvc:
def __init__(self, jobs):
self.jobs = {j.id: j for j in jobs}
self.patches: list[tuple[int, dict]] = []
async def search(self, filter=None):
filter = filter or {}
return [j for j in self.jobs.values() if all(j.__dict__.get(k) == v for k, v in filter.items())]
async def patch(self, id, body):
self.patches.append((int(id), body))
job = self.jobs[int(id)]
for k, v in body.items():
setattr(job, k, v)
class FakeAppointmentSvc:
def __init__(self, data):
self._data = data
async def read(self, id):
return FakeDoc(**self._data) if self._data else None
def _job(id=1, kind="appointment_reminder", payload=None):
return FakeDoc(
id=id,
name="j",
kind=kind,
status="pending",
payload=payload or {"appointment_id": 1},
retry_count=0,
)
def test_dispatch_table_has_appointment_reminder():
assert "appointment_reminder" in HANDLERS
async def test_poll_jobs_runs_pending_and_marks_done(monkeypatch):
monkeypatch.setattr("worker.settings.worker_poll_seconds", 0)
appointment = {
"id": 1,
"date": "2026-11-02T10:00:00",
"pet": {"name": "Rex", "parents": [{"name": "Kim", "email": "kim@example.com"}]},
}
job_svc = FakeJobSvc([_job()])
resources = {"job_svc": job_svc, "appointment_svc": FakeAppointmentSvc(appointment)}
import asyncio
task = asyncio.create_task(poll_jobs(resources))
for _ in range(50):
await asyncio.sleep(0.01)
if job_svc.patches and job_svc.patches[-1][1].get("status") == "done":
break
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
statuses = [body["status"] for _, body in job_svc.patches]
assert statuses == ["processing", "done"]
async def test_poll_jobs_marks_unknown_kind_failed(monkeypatch):
monkeypatch.setattr("worker.settings.worker_poll_seconds", 0)
job_svc = FakeJobSvc([_job(kind="nope")])
resources = {"job_svc": job_svc, "appointment_svc": FakeAppointmentSvc(None)}
import asyncio
task = asyncio.create_task(poll_jobs(resources))
for _ in range(50):
await asyncio.sleep(0.01)
if job_svc.patches and job_svc.patches[-1][1].get("status") == "failed":
break
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
last = job_svc.patches[-1][1]
assert last["status"] == "failed"
assert "no handler" in last["error_message"]
assert last["retry_count"] == 1
async def test_poll_jobs_skips_retries_exhausted(monkeypatch):
monkeypatch.setattr("worker.settings.worker_poll_seconds", 0)
monkeypatch.setattr("worker.settings.worker_max_retries", 1)
job = _job()
job.retry_count = 1
job_svc = FakeJobSvc([job])
resources = {"job_svc": job_svc, "appointment_svc": FakeAppointmentSvc(None)}
import asyncio
task = asyncio.create_task(poll_jobs(resources))
await asyncio.sleep(0.05)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
assert job_svc.patches == []

51
standards/src/templates/app/worker.py
View File

@@ -1,8 +1,8 @@
"""# Summary
Background worker — polls for pending jobs and runs the item-sync pipeline.
The poll loop is the pipeline axis of the blueprint: resources are created by
the API axis, workers consume them on a fixed cadence.
Background worker — polls for pending jobs and runs the matching pipeline
family. The poll loop is the pipeline axis of the sample: resources are created
by the API axis, workers consume them on a fixed cadence.
Notes:
- This process is separate from the API process. Any in-memory event bus is per
@@ -11,9 +11,10 @@ Notes:
"""
import asyncio
import logging
from typing import Any, Awaitable, Callable
from config import settings
from pipelines.item_sync import run_item_sync
from pipelines.appointment_reminder.impl import get_appointment_reminder_pipeline
logging.basicConfig(
level=logging.INFO,
@@ -22,9 +23,29 @@ logging.basicConfig(
logger = logging.getLogger(__name__)
# job.kind -> async (job, resources) -> None. Add families here as they land.
HANDLERS: dict[str, Callable[..., Awaitable[None]]] = {}
async def poll_jobs(resources: dict):
"""Poll for pending jobs and run the item-sync pipeline on each.
def _register_default_handlers() -> None:
"""Populate the dispatch table with the shipped families."""
async def run_appointment_reminder(job, resources: dict[str, Any]) -> None:
payload = job.payload or {}
appointment_id = payload.get("appointment_id")
if appointment_id is None:
raise ValueError("appointment_reminder job requires payload.appointment_id")
pipeline = get_appointment_reminder_pipeline(resources["appointment_svc"])
await pipeline.run(int(appointment_id))
HANDLERS["appointment_reminder"] = run_appointment_reminder
_register_default_handlers()
async def poll_jobs(resources: dict) -> None:
"""Poll for pending jobs and dispatch each to its pipeline family.
Args:
resources (dict): Wired services from :func:`init.init`.
@@ -34,8 +55,9 @@ async def poll_jobs(resources: dict):
settings.worker_poll_seconds, settings.worker_max_retries)
while True:
pending = await job_svc.search(filter={"status": "pending"}, limit=10)
logger.info("Found %d pending job(s)", len(pending) if pending else 0)
pending = await job_svc.search({"status": "pending"})
pending = pending[:10]
logger.info("Found %d pending job(s)", len(pending))
for job in pending:
job_id = str(job.id)
@@ -43,13 +65,20 @@ async def poll_jobs(resources: dict):
logger.warning("SKIPPING job %s — retries exhausted", job_id)
continue
handler = HANDLERS.get(job.kind)
await job_svc.patch(job_id, {"status": "processing"})
try:
await run_item_sync(job, resources)
if handler is None:
raise ValueError(f"no handler for job kind '{job.kind}'")
await handler(job, resources)
await job_svc.patch(job_id, {"status": "done", "error_message": ""})
except Exception as exc: # noqa: BLE001
logger.exception("job %s FAILED: %s", job_id, exc)
await job_svc.patch(job_id, {"status": "failed", "error_message": str(exc)})
await job_svc.patch(job_id, {
"status": "failed",
"error_message": str(exc),
"retry_count": (job.retry_count or 0) + 1,
})
await asyncio.sleep(settings.worker_poll_seconds)
@@ -64,4 +93,4 @@ if __name__ == "__main__":
finally:
await shutdown()
asyncio.run(_run())
asyncio.run(_run())