Courierflow Api

Error Handling

Services raise domain exceptions; routes convert to HTTP responses:

Rules:

• Never expose internal error details to clients
• Never silently swallow exceptions (except SomeError: pass)
• At minimum, log a warning — silent catches violate "Fail Visible"

Transaction Boundaries

• Routes own transactions — They call db.commit()
• Services use db.flush() — When they need data visible within same transaction
• Never use db.commit() in services

Webhook Handlers Own Their Session

The "no db.commit() in services" rule applies to request-scoped services where the FastAPI route handler owns the transaction. Webhook handlers (Twilio inbound, DocuSeal callbacks, Stripe webhooks) ARE the route handler — they own their own session via async with AsyncSessionLocal() and MUST await db.commit() (and db.rollback() on error) before returning. The service called from inside still uses flush() only.

@router.post("/webhooks/twilio/inbound")
async def twilio_inbound(request: Request, ...):
    if not _validate_twilio_signature(request, params):
        return Response(status_code=403)
    async with AsyncSessionLocal() as db:
        try:
            reply = await get_concierge_service(db).process_inbound_sms(...)
            await db.commit()
        except Exception:
            await db.rollback()
            raise
    return Response(content=twiml(reply), media_type="application/xml")

See MEMORY pattern_webhook_handler_owns_session.md.

Route-to-Template Gotcha

Route URLs, handler names, and template filenames do not always match:

Builder context variables: Both builder routes must pass template_state (defaults to "draft") for the page header status badge. When adding new template context data to builder routes, update BOTH the new-workflow and edit-workflow handlers.

Always check the route handler before editing templates:

grep -r '"/home"' app/routes/

RORO Pattern (Receive an Object, Return an Object)

Services receive structured input and return structured output — never raw primitives for complex operations:

# GOOD - RORO with Pydantic models
class CreateWorkflowRequest(BaseModel):
    name: str
    event_type: LandlordEventType
    steps: list[StepCreate] = []

class WorkflowResponse(BaseModel):
    model_config = {"from_attributes": True}
    id: UUID
    name: str
    status: WorkflowState

async def create_workflow(
    db: AsyncSession,
    user_id: UUID,
    request: CreateWorkflowRequest,
) -> WorkflowResponse:
    template = WorkflowTemplate(user_id=user_id, **request.model_dump())
    db.add(template)
    await db.flush()
    return WorkflowResponse.model_validate(template)

# BAD - scattered primitives
async def create_workflow(db, user_id, name, event_type, steps=None):
    ...  # Easy to mix up argument order, no validation

When to use RORO:

• Service functions with 3+ parameters → wrap in a request model
• Functions returning data to routes → use a response model
• Simple lookups (get by ID) → primitives are fine

Code Style

• Python: flake8/PEP 8, max line length 79
• Type hints on all function parameters and return values
• Docstrings on all public functions and classes
• No unused imports or variables

Rate Limiting Tiers

Request Timeouts

• Default: 30s
• AI/bulk operations: 60s

Webhook Notifications (CI/CD)

CI/CD scripts use async httpx for webhook notifications:

async def send_webhook_notification(
    webhook_url: str,
    payload: Dict[str, Any],
    timeout: int = 10,
) -> bool:
    async with httpx.AsyncClient(timeout=timeout) as client:
        try:
            response = await client.post(webhook_url, json=payload)
            return response.status_code < 400
        except httpx.HTTPError:
            return False  # Best-effort, don't fail deployment

Key points:

• Use httpx.AsyncClient (not requests)
• Wrap in try-catch - webhook failure should not abort primary operation
• Configurable timeout with sensible default