17759ef564
- apps/api: psycopg v3 repositories, schema.sql + migration runner, state machine implementing the data-model transition table, approval gate with sha256 hash match + 24h expiry enforced at send time, EchoTransport pluggable sender, LLM calls routed through packages.llm-gateway with mock mode, 47 pytest tests green - apps/api/Dockerfile.test: python 3.13-slim test image (DinD-safe: migrations copied as directory) - docker-compose.yml: add api-test service on compose network (host port publishing is broken in this sandbox; container-to-container networking used) - Fix: replace masked placeholder password in config.py/conftest.py defaults
51 lines
No EOL
1.2 KiB
Python
51 lines
No EOL
1.2 KiB
Python
"""Send transport interface.
|
|
|
|
Pluggable Transport interface for the outbox send operation.
|
|
Default EchoTransport records the payload and marks as sent.
|
|
No real email in POC.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Protocol
|
|
|
|
|
|
class Transport(Protocol):
|
|
def send(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Send a payload. Returns a result dict with at least 'success' bool."""
|
|
...
|
|
|
|
|
|
class EchoTransport:
|
|
"""Default transport: records payload, returns success.
|
|
|
|
No real email is sent. Used for POC.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.sent: list[dict[str, Any]] = []
|
|
|
|
def send(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
self.sent.append(payload)
|
|
return {
|
|
"success": True,
|
|
"sent_at": datetime.now(timezone.utc).isoformat(),
|
|
"echo": payload,
|
|
}
|
|
|
|
|
|
# Singleton instance
|
|
_default_transport: Transport | None = None
|
|
|
|
|
|
def get_transport() -> Transport:
|
|
global _default_transport
|
|
if _default_transport is None:
|
|
_default_transport = EchoTransport()
|
|
return _default_transport
|
|
|
|
|
|
def set_transport(t: Transport) -> None:
|
|
global _default_transport
|
|
_default_transport = t |