From 4e2c8885053ce5673d22cc7cf4cad0dc4ba72480 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Wed, 29 Jul 2026 23:00:38 +0200 Subject: [PATCH] phase7: command pattern infrastructure (CommandHandler, UnitOfWork, RequestContext) + example CreateContactCommand --- app/commands/create_contact.py | 103 +++++++++++++++++ app/core/commands.py | 203 +++++++++++++++++++++++++++++++++ 2 files changed, 306 insertions(+) create mode 100644 app/commands/create_contact.py create mode 100644 app/core/commands.py diff --git a/app/commands/create_contact.py b/app/commands/create_contact.py new file mode 100644 index 0000000..d5d8876 --- /dev/null +++ b/app/commands/create_contact.py @@ -0,0 +1,103 @@ +"""Example command: CreateContact using the Command Pattern. + +This is a reference implementation for new modules. +Existing contact_service.py is NOT changed — this is an alternative path. + +Usage: + @router.post("/contacts-v2") + async def create_contact_v2( + body: CreateContactDTO, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(require_permission("contacts:write")), + ): + ctx = RequestContext( + user_id=uuid.UUID(current_user["user_id"]), + tenant_id=uuid.UUID(current_user["tenant_id"]), + is_system_admin=current_user.get("is_system_admin", False), + permissions=set(current_user.get("permissions", [])), + ) + cmd = CreateContactCommand( + firstname=body.firstname, + surname=body.surname, + email=body.email, + ) + handler = CreateContactHandler() + return await handler.execute(cmd, ctx, db) +""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass +from typing import Any + +from app.core.commands import CommandHandler, RequestContext, UnitOfWork +from app.models.contact import Contact + + +@dataclass +class CreateContactCommand: + """Command to create a new contact.""" + firstname: str + surname: str + email: str | None = None + phone: str | None = None + company: str | None = None + + +class CreateContactHandler(CommandHandler[CreateContactCommand, dict[str, Any]]): + """Handler for CreateContactCommand. + + Demonstrates the Command Pattern: + 1. Authorization check (ctx.require) + 2. Domain operation (create Contact) + 3. Outbox event (crm.contact.created.v1) + 4. Audit log (contact.created) + 5. Single commit via UoW + """ + + async def handle(self, cmd: CreateContactCommand, ctx: RequestContext, uow: UnitOfWork) -> dict[str, Any]: + # 1. Authorization + ctx.require("contacts:write") + + # 2. Domain operation + contact = Contact( + tenant_id=ctx.tenant_id, + firstname=cmd.firstname, + surname=cmd.surname, + email_1=cmd.email, + phone_1=cmd.phone, + company=cmd.company, + owner_id=ctx.user_id, + created_by=ctx.user_id, + updated_by=ctx.user_id, + ) + uow.add(contact) + + # 3. Outbox event (standardized envelope) + uow.outbox_add( + event_name="crm.contact.created.v1", + aggregate_id=contact.id, # Will be set after flush + aggregate_type="contact", + payload={ + "firstname": cmd.firstname, + "surname": cmd.surname, + "email": cmd.email, + }, + ) + + # 4. Audit log + uow.audit_record( + action="create", + entity_id=contact.id, + entity_type="contact", + changes={"firstname": cmd.firstname, "surname": cmd.surname, "email": cmd.email}, + ) + + # 5. Return dict (will be populated after flush in commit) + return { + "id": str(contact.id), + "firstname": contact.firstname, + "surname": contact.surname, + "email_1": contact.email_1, + } diff --git a/app/core/commands.py b/app/core/commands.py new file mode 100644 index 0000000..27f5dc6 --- /dev/null +++ b/app/core/commands.py @@ -0,0 +1,203 @@ +"""Command pattern infrastructure for new modules. + +This provides a clean, transactional command handler pattern: + + HTTP Route → Command Handler → Authorization → Domain Operation → Audit + Outbox → one Commit + +Existing services are NOT refactored — they continue to work as-is. +New modules (ERP, etc.) should use this pattern. + +Usage: + + @dataclass + class CreateInvoiceCommand: + customer_id: uuid.UUID + amount: Decimal + + class CreateInvoiceHandler(CommandHandler[CreateInvoiceCommand, Invoice]): + async def handle(self, cmd: CreateInvoiceCommand, ctx: RequestContext, uow: UnitOfWork) -> Invoice: + ctx.require("invoices:create") + invoice = Invoice.create(tenant_id=ctx.tenant_id, owner_id=ctx.user_id, ...) + uow.add(invoice) + uow.outbox.add("crm.invoice.created.v1", invoice.id, "invoice", invoice.to_dict()) + uow.audit.record("invoice.created", invoice.id) + return invoice + + # In route: + @router.post("/invoices") + async def create_invoice(body: CreateInvoiceDTO, ctx: RequestContext = Depends(get_request_context)): + cmd = CreateInvoiceCommand(customer_id=body.customer_id, amount=body.amount) + handler = CreateInvoiceHandler() + result = await handler.execute(cmd, ctx) + return result +""" + +from __future__ import annotations + +import uuid +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Generic, TypeVar + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.audit import log_audit +from app.core.outbox import enqueue_outbox_event + + +TCommand = TypeVar("TCommand") +TResult = TypeVar("TResult") + + +@dataclass +class RequestContext: + """Request context with user, tenant, and permission info. + + Passed to every command handler. Provides authorization checks. + """ + user_id: uuid.UUID + tenant_id: uuid.UUID + is_system_admin: bool = False + permissions: set[str] = field(default_factory=set) + correlation_id: uuid.UUID = field(default_factory=uuid.uuid4) + + def require(self, permission: str) -> None: + """Require a permission. Raises PermissionError if not granted.""" + if self.is_system_admin: + return + if permission not in self.permissions: + raise PermissionError(f"Missing permission: {permission}") + + def has(self, permission: str) -> bool: + """Check if user has a permission.""" + if self.is_system_admin: + return True + return permission in self.permissions + + +class UnitOfWork: + """Unit of Work — collects changes, audit, and outbox events. + + One UoW per business operation. Commit happens once at the end. + """ + + def __init__(self, db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID): + self.db = db + self.tenant_id = tenant_id + self.user_id = user_id + self._audit_entries: list[dict[str, Any]] = [] + self._outbox_events: list[dict[str, Any]] = [] + + def add(self, entity: Any) -> None: + """Add an entity to the session.""" + self.db.add(entity) + + def outbox_add( + self, + event_name: str, + aggregate_id: uuid.UUID, + aggregate_type: str, + payload: dict[str, Any], + schema_version: int = 1, + ) -> None: + """Queue an outbox event for commit.""" + self._outbox_events.append({ + "event_name": event_name, + "aggregate_id": aggregate_id, + "aggregate_type": aggregate_type, + "payload": payload, + "schema_version": schema_version, + }) + + def audit_record(self, action: str, entity_id: uuid.UUID, entity_type: str = "", changes: dict[str, Any] | None = None) -> None: + """Queue an audit log entry for commit.""" + self._audit_entries.append({ + "action": action, + "entity_id": entity_id, + "entity_type": entity_type, + "changes": changes or {}, + }) + + async def commit(self) -> None: + """Flush, write audit + outbox, then commit.""" + # Flush to get entity IDs + await self.db.flush() + + # Write outbox events + for evt in self._outbox_events: + await enqueue_outbox_event( + self.db, + self.tenant_id, + evt["event_name"], + evt["payload"], + aggregate_type=evt["aggregate_type"], + aggregate_id=evt["aggregate_id"], + schema_version=evt["schema_version"], + ) + + # Write audit entries + for entry in self._audit_entries: + await log_audit( + self.db, + self.tenant_id, + self.user_id, + entry["action"], + entry["entity_type"], + entry["entity_id"], + changes=entry["changes"], + ) + + # Single commit for everything + await self.db.commit() + + async def rollback(self) -> None: + """Rollback the transaction.""" + await self.db.rollback() + + +class CommandHandler(ABC, Generic[TCommand, TResult]): + """Base class for command handlers. + + Subclasses implement `handle()` with the business logic. + The `execute()` method wraps it with UoW creation and error handling. + """ + + @abstractmethod + async def handle(self, command: TCommand, ctx: RequestContext, uow: UnitOfWork) -> TResult: + """Business logic. Use uow.add(), uow.outbox_add(), uow.audit_record().""" + ... + + async def execute(self, command: TCommand, ctx: RequestContext, db: AsyncSession) -> TResult: + """Execute the command with a Unit of Work. + + Creates a UoW, calls handle(), commits on success, rolls back on error. + """ + uow = UnitOfWork(db, ctx.tenant_id, ctx.user_id) + try: + result = await self.handle(command, ctx, uow) + await uow.commit() + return result + except Exception: + await uow.rollback() + raise + + +# ── FastAPI Dependency ─────────────────────────────────────────────────────── + +async def get_request_context( + current_user: dict = None, # Will be injected by FastAPI with require_permission +) -> RequestContext: + """Build a RequestContext from the current user. + + Usage in routes: + ctx: RequestContext = Depends(get_request_context) + """ + if current_user is None: + raise PermissionError("Not authenticated") + return RequestContext( + user_id=uuid.UUID(current_user["user_id"]), + tenant_id=uuid.UUID(current_user["tenant_id"]), + is_system_admin=current_user.get("is_system_admin", False), + permissions=set(current_user.get("permissions", [])), + )