Handler and registry pattern

Every function separates three concerns absolutely:

Layer Responsibility

handler.py

Detect the event shape, resolve the routing key, validate the payload, call the handler. No business logic, ever.

registry.py

Map routing keys to (handler class, Pydantic model) pairs. Nothing else.

events/<name>.py

One event type’s business logic, in a class extending the function’s base handler.

Adding behaviour therefore never means editing the dispatcher.

The dispatcher

def lambda_handler(event: dict, context) -> dict | None:
    records = event.get("Records", [])
    if records and records[0].get("eventSource") == "aws:sqs":
        return _handle_sqs_batch(records, context)      # → {"batchItemFailures": [...]}

    detail_type = event.get("detail-type", "")
    if detail_type:
        return _handle_eventbridge(event, detail_type, context)   # → None

    logger.warning("Unrecognised event shape — neither SQS batch nor EventBridge")
    return None

Shape detection is by inspection, not configuration, so one function can serve both an SQS queue and any number of schedules.

SQS path

Per record:

  1. Read the routing key from the event_type message attribute.

  2. Look it up in the registry. Unknown key → log a warning, continue; the message is deleted. Retrying an unroutable message cannot help.

  3. json.loads the body. Failure → add to batchItemFailures (the body is malformed, so it belongs in the DLQ).

  4. model_validate() the body. ValidationError → add to batchItemFailures.

  5. Instantiate the handler and call handle(parsed, context). An exception → log with traceback and add to batchItemFailures.

Returns {"batchItemFailures": [{"itemIdentifier": messageId}, …]}. Every SQS trigger declares FunctionResponseTypes: [ReportBatchItemFailures], so a failure isolates to the offending message.

EventBridge path

  1. Read detail-type from the event root.

  2. Look it up in the scheduled registry. Unknown → log a warning and return.

  3. model_validate() the whole event (not a body) — scheduled models are extra="ignore" so the envelope’s extra fields are tolerated. A ValidationError logs the errors plus a 500-character raw preview, then re-raises.

  4. Call the handler. An exception is logged with traceback and re-raised.

Scheduled failures propagate because there is no queue to absorb them; the failure should be visible in the function’s error metric.

Per-function variations

Function Variation

morpheus-hubspot

Two separate registries: event_registry (SQS) and scheduled_registry (EventBridge). The routing key comes from the event_type field inside the message body, not from a message attribute. A body may also be a JSON array of event objects, each dispatched individually; if any element fails, the whole message is marked failed. An unknown event_type returns True (success) so the message is discarded.

enrichment-scorer

A single registry serves both SQS and EventBridge. Falls back to a default routing key SellerRecord.Received when the message attribute is absent, for backward compatibility with messages published before the attribute existed.

email-agent, phone-agent

A missing event_type message attribute logs a warning and skips the record.

enrichment-scraper

EventBridge only — no SQS path at all.

Registry shapes

Two styles, functionally equivalent.

Dataclass registry — scraper, scorer, email-agent, phone-agent
@dataclass
class EventRegistration:
    handler_class: type[BaseEventHandler]
    event_model: type[BaseModel]

@dataclass
class EventRegistry:
    _registry: dict[str, EventRegistration] = field(default_factory=dict)
    def register(self, key, handler_class, event_model) -> None: ...
    def lookup(self, key) -> EventRegistration | None: ...
    def registered_types(self) -> list[str]: ...

def build_registry() -> EventRegistry:
    registry = EventRegistry()
    registry.register("SellerRecord.Received", SellerReceivedHandler, SellerRecord)
    return registry

event_registry: EventRegistry = build_registry()     # module-level singleton

+ Handlers are stored as classes and instantiated per message.

Plain dict registry — morpheus-hubspot
event_registry: dict = {
    "Seller.CrmSync": (SellerCrmSyncHandler(), SellerSyncEvent),
    "Contact.Update": (ContactUpdateHandler(), ContactUpdateEvent),
    # …
}

+ Handlers are stored as instances, constructed once at import. The module carries a # TODO: Class should be defined and not initialized here noting the divergence.

Either way registered_types() — or the dict’s keys — is what appears in the "no handler registered" log line, which makes a typo in a detail-type easy to diagnose.

Base handler classes

Function Base class

enrichment-scraper, morpheus-hubspot

BaseEventHandler

enrichment-scorer

BaseEventHandler, and BaseEnrichmentHandler for enrichment work

email-agent

BaseEmailEventHandler

phone-agent

BasePhoneEventHandler

All are the same two lines:

class BaseEventHandler(ABC):
    @abstractmethod
    def handle(self, event: BaseModel, context: Any) -> None: ...

BaseEnrichmentHandler is the exception — it carries real shared behaviour: class-level repositories plus _enrich(), _apply_enrichment(), _score(), _apply_score(), and _publish_crm_sync(). Its subclasses differ only in how they source sellers (an SQS message versus a database query) and call the inherited steps in whatever order their trigger requires. See enrichment-scorer.

Logging

Both dispatch paths call configure_logging(level, aws_request_id, event_type) before handing off, so every line in an invocation carries the request id and the routing key. Implementations live in each function’s utils/logging.py.

Failure semantics

Situation Message fate Rationale

Unknown routing key

deleted

Retrying cannot route it

Malformed JSON body

DLQ

The producer is broken; keep the evidence

ValidationError

DLQ

Same

Subscription mismatch

deleted (handler returns)

A poison message from a misconfigured producer must not loop

Missing database row

deleted (handler returns)

Nothing to act on

Handler raised

DLQ

Possibly transient; keep it for inspection or replay

Scheduled handler raised

n/a

Propagates; the next tick retries

The distinction that matters: bad data returns normally, transient failure raises. Getting it backwards either loses real work or fills a DLQ with unfixable messages.

Adding an event type

  1. Create events/<name>.py with a handler extending the function’s base class.

  2. Define or reuse a Pydantic model — in models/ for a function-local event, or in morpheus_common.events.models for anything crossing a function boundary.

  3. Register the routing key in registry.py.

  4. Add the trigger to lambdas/template.yaml — an SQS Events entry, or a Schedule whose Input detail-type matches the registry key exactly.

  5. Add a sample payload under events/samples/ and a test.

Full walkthrough with code: Adding a new event.