Architecture Patterns & Decoupled Domain Modeling
Apply the Dependency Inversion Principle from Architecture Patterns with Python (Cosmic Python). Pure domain aggregates contain zero framework imports, decoupled behind Repository abstractions, Unit of Work transaction managers, and CQRS query projections.
Senior Systems Studio: Architecture Patterns & Decoupled Domain Modeling
Select any senior topic below to inspect pure domain aggregate invariant enforcement, test in-memory Repository and Unit of Work (UoW) transaction context managers, and execute CQRS write vs raw read path projections.
Cosmic Python Architecture & Dependency Inversion
- Dependency Inversion: Domain aggregates use standard Python types (`dataclass`, `int`, `str`) with zero framework, ORM, or database imports.
- Repository & Unit of Work: Repositories emulate in-memory collections; UoW context managers (`with uow:`) orchestrate atomic transaction boundaries.
- CQRS Separation: Write paths route commands through aggregates to enforce invariants; Read paths query raw database views directly for sub-millisecond projections.
[ Outside World ] (HTTP Requests, CLI, Webhooks, Cron)
│
▼
┌────────────────────────────────────────────────────────┐
│ 1. Entrypoints / Ports (FastAPI, CLI, Flask) │
└───────────────────────┬────────────────────────────────┘
│ Invokes
▼
┌────────────────────────────────────────────────────────┐
│ 2. Service Layer / Message Bus (Orchestration) │
│ - Coordinates Unit of Work │
│ - Routes Commands & Dispatches Domain Events │
└───────────────┬───────────────────────┬────────────────┘
│ Uses │ Persists via
▼ ▼
┌──────────────────────────────┐ ┌──────────────────────┐
│ 3. Domain Model (Aggregates) │ │ 4. Adapters │
│ - Pure Business Rules │ │ - Repositories │
│ - Invariants & State │ │ - DB / Filesystem │
│ (Zero external imports) │ │ (Implements UoW) │
└──────────────────────────────┘ └──────────────────────┘
1. Domain Model (Aggregates) & Repository Pattern
Aggregates are pure Python classes enforcing domain invariants. The Repository pattern presents persistence as an in-memory collection, hiding database engine logic:
from dataclasses import dataclass
from typing import Protocol
@dataclass
class AllocationLine:
order_id: str
sku: str
qty: int
class Batch:
"""Pure Domain Aggregate with Zero External Imports."""
def __init__(self, ref: str, sku: str, qty: int):
self.ref = ref
self.sku = sku
self._available_qty = qty
def can_allocate(self, line: AllocationLine) -> bool:
return self.sku == line.sku and self._available_qty >= line.qty
def allocate(self, line: AllocationLine) -> None:
if not self.can_allocate(line):
raise ValueError(f"Cannot allocate {line.qty} of {line.sku}")
self._available_qty -= line.qty
class AbstractBatchRepository(Protocol):
def add(self, batch: Batch) -> None: ...
def get(self, ref: str) -> Batch | None: ...2. Unit of Work (UoW) & Service Layer
The Unit of Work manages transaction lifecycles using context managers. The Service Layer orchestrates business operations without coupling entrypoints to database engines:
class AbstractUnitOfWork(Protocol):
batches: AbstractBatchRepository
def __enter__(self) -> Self: ...
def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...
def commit(self) -> None: ...
def rollback(self) -> None: ...
def allocate_order_service(line: AllocationLine, uow: AbstractUnitOfWork) -> str:
"""Service layer coordinating domain workflow via Unit of Work."""
with uow:
batch = uow.batches.get_by_sku(line.sku)
if not batch:
raise ValueError(f"No batch found for SKU {line.sku}")
batch.allocate(line)
uow.commit()
return batch.ref3. CQRS: Write Path vs Direct Read Queries
Command Query Responsibility Segregation separates complex state mutation (Write Path) from fast, raw database reads (Read Path) projecting plain `TypedDict` data directly:
from typing import TypedDict
class BatchReadView(TypedDict):
batch_ref: str
sku: str
available_qty: int
def query_allocations_read_path(db_connection, sku: str) -> list[BatchReadView]:
"""CQRS Read Path: Bypasses domain aggregates; executes raw SQL."""
cursor = db_connection.execute(
"SELECT ref, sku, available_qty FROM batches WHERE sku = ?", (sku,)
)
return [
{"batch_ref": row[0], "sku": row[1], "available_qty": row[2]}
for row in cursor.fetchall()
]