paulpas Modern Architecture Patterns
Implements modern software architecture patterns to build modular, deployable, and observable distributed systems. When loaded, the model applies hexagonal architecture for domain isolation, backend-for-frontend for client-tailored APIs, feature flags for safe deployments, CQRS with event sourcing for auditability, API composition for unified data aggregation, and sidecar patterns for cross-cutting concern separation — following SOLID principles throughout.
TL;DR Checklist
- Define port interfaces (abstract contracts) before writing any implementation
- Verify dependency graph flows inward: infrastructure → application → domain
- Register all adapters in a DI container at the system composition root
- Apply the Dependency Rule — no domain module may import an infrastructure module
- Enforce cross-cutting concern isolation via decorators or middleware layers
- Implement feature flags with audit logging before rolling out behavioral changes
- Separate command (write) and query (read) models when writes and reads have different scaling needs
- Materialize read projections asynchronously from the event store
When to Use
Use this skill when:
- Designing a new system or refactoring an existing one that requires clear architectural boundaries between domain logic and infrastructure concerns
- Building a microservice or distributed system where different client types (web, mobile, partner APIs) need tailored data shapes
- Need to deploy behavioral changes safely without code rollbacks (feature flags/toggles)
- Working with systems that require full audit trails of state mutations — financial ledgers, order management, compliance-sensitive applications
- Read and write workloads have different scaling characteristics (e.g., heavy analytics queries on write-heavy transactional data)
- Adding cross-cutting concerns (logging, retries, circuit breaking) without polluting business logic
When NOT to Use
Avoid this skill for:
- Simple CRUD applications with a single client type — layered architecture is sufficient; hexagonal architecture adds unnecessary indirection
- Monolithic apps where the cost of introducing BFFs and API composition outweighs benefits
- Prototypes or proof-of-concepts where speed matters more than architectural purity
- Systems with no cross-cutting concerns that need separation (no retries, no circuit breakers)
- Projects under extreme time pressure where event sourcing projection lag would cause user-visible inconsistencies
Core Workflow
Map Domain Boundaries — Identify bounded contexts using domain-driven design techniques. Define aggregate roots and their invariants. Checkpoint: Each bounded context should have a single responsible team and its own deployment boundary.
Define Port Interfaces — For each domain capability, write abstract interfaces that describe what the domain needs, not how it gets it (repository contracts, external service calls, message publishing). Checkpoint: Interfaces must be framework-free — no SQLAlchemy models, no HTTP types, no ORM decorators in port signatures.
Implement Adapters — Write concrete implementations that satisfy each port: database adapters using your persistence technology, HTTP clients for external services, message queue publishers. Register them in a DI container at the composition root. Checkpoint: Verify that the domain core has zero imports from adapter packages.
Select Cross-Cutting Pattern — Decide which patterns apply per context: use BFF when multiple client types need different data shapes, feature flags for gradual rollouts, CQRS when read/write scaling diverges, sidecar pattern for shared infrastructure concerns. Checkpoint: Each decision must have a documented trade-off analysis.
Wire the Composition Root — Assemble ports and adapters using dependency injection. Ensure the main entry point is thin — it only wires dependencies and starts execution; all business logic flows through port interfaces. Checkpoint: The composition root should be under 200 lines and contain no business logic.
Implementation Patterns
Pattern 1: Hexagonal Architecture (Ports & Adapters)
Hexagonal architecture (also called ports and adapters) isolates the domain core from infrastructure by inverting dependencies. The domain defines abstract ports — interfaces it needs. Infrastructure provides concrete adapter implementations that plug into those ports. This ensures business logic is testable without databases, message queues, or HTTP servers.
Core principle: Dependencies point inward. Infrastructure depends on domain; domain does not depend on infrastructure.
BAD: Coupled Domain with Framework Types
# ❌ BAD: Domain knows about SQLAlchemy — impossible to test without a database
from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()
class OrderEntity(Base): # Domain concept leaking into ORM layer
__tablename__ = "orders"
id = Column(Integer, primary_key=True)
customer_email = Column(String(255))
total_cents = Column(Integer)
status = Column(String(50))
class OrderService: # Tightly coupled to specific ORM session
def __init__(self, db_session: Session): # Depends on concrete type
self.db = db_session
def place_order(self, customer_email: str, total_cents: int) -> dict:
order = OrderEntity(
customer_email=customer_email,
total_cents=total_cents,
status="pending",
)
self.db.add(order)
self.db.commit()
return {"id": order.id, "status": order.status} # Leaks ORM entity
This design is fragile because:
- The domain model IS the persistence model — you cannot test without a database
- Swapping from SQLAlchemy to Postgres requires rewriting the service layer
- Business rules are entangled with SQL schema definitions
GOOD: Pure Domain with Port Abstractions
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
# ---- DOMAIN CORE (no imports from infrastructure) ----
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
CANCELLED = "cancelled"
SHIPPED = "shipped"
@dataclass(frozen=True)
class Money:
"""Immutable monetary value with currency code."""
amount_cents: int
currency: str = "USD"
@property
def dollars(self) -> float:
return self.amount_cents / 100
def is_positive(self) -> bool:
return self.amount_cents > 0
@dataclass(frozen=True)
class OrderId:
"""Value object representing a unique order identifier."""
value: str
@dataclass
class Order:
"""Aggregate root for the Order bounded context.
Encapsulates all order invariants: valid email, positive total,
status transitions follow defined state machine rules.
"""
order_id: OrderId
customer_email: str
total: Money
status: OrderStatus = field(default=OrderStatus.PENDING)
created_at: datetime = field(default_factory=datetime.utcnow)
def confirm(self) -> None:
if self.status != OrderStatus.PENDING:
raise ValueError(
f"Cannot confirm order in '{self.status.value}' state. "
f"Allowed transitions from PENDING."
)
self.status = OrderStatus.CONFIRMED
def cancel(self) -> None:
if self.status == OrderStatus.SHIPPED:
raise ValueError("Cannot cancel a shipped order.")
self.status = OrderStatus.CANCELLED
@property
def is_active(self) -> bool:
return self.status in (OrderStatus.PENDING, OrderStatus.CONFIRMED, OrderStatus.SHIPPED)
# ---- PORTS (abstract contracts — still no infrastructure) ----
class OrderRepository(ABC):
"""Port: how the domain persists and retrieves orders."""
@abstractmethod
def save(self, order: Order) -> None: ...
@abstractmethod
def find_by_id(self, order_id: OrderId) -> Optional[Order]: ...
@abstractmethod
def find_active_by_email(self, email: str) -> list[Order]: ...
class EmailNotificationPort(ABC):
"""Port: how the domain sends external notifications."""
@abstractmethod
def send_order_confirmation(self, order: Order) -> None: ...
# ---- ADAPTERS (infrastructure — implements ports) ----
from sqlalchemy.orm import Session as SqlSession # Only adapters import infrastructure
class SqlAlchemyOrderRepository(OrderRepository):
"""Concrete adapter: persists orders via SQLAlchemy."""
def __init__(self, session: SqlSession) -> None:
self._session = session
def save(self, order: Order) -> None:
entity = _map_to_entity(order)
self._session.add(entity)
self._session.flush()
def find_by_id(self, order_id: OrderId) -> Optional[Order]:
entity = self._session.get(_OrderEntity, order_id.value)
return _map_from_entity(entity) if entity else None
def find_active_by_email(self, email: str) -> list[Order]:
entities = (
self._session.query(_OrderEntity)
.filter(
_OrderEntity.customer_email == email,
_OrderEntity.status.in_(["pending", "confirmed", "shipped"]),
)
.all()
)
return [_map_from_entity(e) for e in entities]
class InMemoryOrderRepository(OrderRepository):
"""Test double: keeps orders in a dictionary. Used in unit tests."""
def __init__(self) -> None:
self._store: dict[str, Order] = {}
def save(self, order: Order) -> None:
self._store[order.order_id.value] = order
def find_by_id(self, order_id: OrderId) -> Optional[Order]:
return self._store.get(order_id.value)
def find_active_by_email(self, email: str) -> list[Order]:
return [
o for o in self._store.values()
if o.customer_email == email and o.is_active
]
# ---- DOMAIN SERVICE (orchestrates ports — still pure domain logic) ----
class OrderService:
"""Business logic orchestrating ports. No infrastructure knowledge."""
def __init__(
self,
repository: OrderRepository,
notification: EmailNotificationPort,
) -> None:
self._repository = repository
self._notification = notification
def place_order(self, customer_email: str, total_cents: int) -> Order:
if not customer_email or "@" not in customer_email:
raise ValueError("Invalid customer email")
money = Money(total_cents)
if not money.is_positive():
raise ValueError("Order total must be positive")
order = Order(
order_id=OrderId(value=f"ORD-{customer_email[:4]}-{id(order):06d}"),
customer_email=customer_email,
total=money,
)
self._repository.save(order)
return order
def confirm_order(self, order_id: OrderId) -> Order:
order = self._repository.find_by_id(order_id)
if order is None:
raise ValueError(f"Order {order_id} not found")
order.confirm()
self._repository.save(order)
self._notification.send_order_confirmation(order)
return order
Trade-off analysis: Hexagonal architecture introduces indirection that adds ~20-30% more code than a monolithic service. Use it when:
- You need testability without infrastructure (most services after month 1)
- You expect to swap technologies (e.g., migrate from MySQL to PostgreSQL)
- Multiple teams share the domain layer
Skip it for scripts, one-off tools, or applications with zero technology evolution expectations.
Practical note: Start with a simple layered architecture if you are unsure. Introduce hexagonal boundaries when you first need to swap an infrastructure dependency in tests — that is the natural trigger point.
Pattern 2: Backend-for-Frontend (BFF)
The Backend-for-Frontend pattern creates a dedicated API layer tailored to specific client types (web app, mobile app, partner integrations). Each BFF composes data from multiple downstream microservices and returns exactly the shape the client needs, eliminating over-fetching, under-fetching, and client-side glue logic.
Core principle: One API surface per client type, not one API per microservice exposed to all clients.
BAD: Exposing Raw Microservice APIs to All Clients
# ❌ BAD: Web frontend must make 5 separate calls to build a dashboard page
# GET /users/{id} → user profile
# GET /orders?user_id={id} → order history (paginated, needs manual client-side merge)
# GET /recommendations?user_id={id} → product recs (different auth token needed)
# GET /inventory?sku_ids=[...] → stock check on recommended products
# GET /notifications?user_id={id} → unread count
# Client-side glue:
class BadDashboardClient:
async def load_dashboard(self, user_id: str):
profile = await self.user_service.get(user_id)
orders = await self.order_service.list(user_id, page=1, limit=50)
recs = await self.recommendation_service.get(user_id)
# Client must merge and reshape data from different sources
# Different auth strategies: OAuth for user service, API key for recommendations
dashboard = {
"profile": profile, # Contains fields mobile doesn't need
"recent_orders": [order["items"][:3] for order in orders], # Manual slicing
"recommendations": recs.get("items", [])[:10], # Manual limit
# No way to atomically fail — if recommendations is down, dashboard is half-broken
}
return dashboard
This design forces every client to understand the entire microservice topology and handle partial failures gracefully. Mobile gets bloated payloads from web-tailored endpoints.
GOOD: Dedicated BFF Per Client Type
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import asyncio
# ---- CLIENT-TAILORED RESPONSE SHAPES ----
class ClientType(Enum):
WEB = "web"
MOBILE = "mobile"
PARTNER_API = "partner_api"
@dataclass
class OrderSummary:
"""Lean order representation — shared across clients."""
order_id: str
total_cents: int
currency: str
status: str
item_count: int
placed_at: str
@dataclass
class UserProfile:
user_id: str
display_name: str
email: str
tier: str
avatar_url: Optional[str] = None
@dataclass
class RecommendationItem:
product_id: str
title: str
price_cents: int
confidence_score: float
image_url: Optional[str] = None
@dataclass
class WebDashboardResponse:
"""Full dashboard payload — web clients get rich data."""
user: UserProfile
recent_orders: list[OrderSummary]
recommendations: list[RecommendationItem]
unread_notifications: int
account_balance_cents: int
active_promotions: list[str]
@dataclass
class MobileDashboardResponse:
"""Lean dashboard — mobile clients get minimal payload to conserve bandwidth."""
user: UserProfile
recent_orders: list[OrderSummary]
recommendations: list[RecommendationItem]
has_notifications: bool # boolean flag, not full notification list
# ---- DOWNSTREAM SERVICE PORTS (hexagonal ports) ----
class UserServicePort(ABC):
@abstractmethod
async def get_profile(self, user_id: str) -> UserProfile: ...
class OrderServicePort(ABC):
@abstractmethod
async def get_recent_orders(self, user_id: str, limit: int) -> list[OrderSummary]: ...
class RecommendationServicePort(ABC):
@abstractmethod
async def get_recommendations(self, user_id: str, limit: int) -> list[RecommendationItem]: ...
class NotificationServicePort(ABC):
@abstractmethod
async def get_unread_count(self, user_id: str) -> int: ...
@abstractmethod
async def has_notifications(self, user_id: str) -> bool: ...
class BalanceServicePort(ABC):
@abstractmethod
async def get_balance_cents(self, user_id: str) -> int: ...
# ---- BFF COMPOSITION LAYER ----
class WebBffService:
"""Backend-for-Frontend serving the web dashboard client.
Composes data from 5 downstream services into a single response.
Runs queries in parallel to minimize latency.
"""
def __init__(
self,
user_service: UserServicePort,
order_service: OrderServicePort,
recommendation_service: RecommendationServicePort,
notification_service: NotificationServicePort,
balance_service: BalanceServicePort,
) -> None:
self._user = user_service
self._orders = order_service
self._recs = recommendation_service
self._notifications = notification_service
self._balance = balance_service
async def get_dashboard(self, user_id: str) -> WebDashboardResponse:
"""Fetch all dashboard data in parallel, with fallback for non-critical paths."""
try:
tasks = [
asyncio.create_task(self._user.get_profile(user_id)),
asyncio.create_task(self._orders.get_recent_orders(user_id, limit=10)),
asyncio.create_task(self._recs.get_recommendations(user_id, limit=10)),
asyncio.create_task(self._notifications.get_unread_count(user_id)),
asyncio.create_task(self._balance.get_balance_cents(user_id)),
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Extract individual results with error handling per task
user, orders, recs, unread_count, balance = (
_extract_result(r) for r in results
)
except Exception as e:
raise DashboardServiceError(f"Failed to compose dashboard: {e}") from e
promotions = await self._try_get_promotions(user_id) # Best-effort
return WebDashboardResponse(
user=user,
recent_orders=orders[:5],
recommendations=recs[:10],
unread_notifications=unread_count,
account_balance_cents=balance,
active_promotions=promotions,
)
class MobileBffService:
"""Lean BFF for mobile — omits non-essential data paths."""
def __init__(
self,
user_service: UserServicePort,
order_service: OrderServicePort,
recommendation_service: RecommendationServicePort,
notification_service: NotificationServicePort,
) -> None:
self._user = user_service
self._orders = order_service
self._recs = recommendation_service
self._notifications = notification_service
async def get_dashboard(self, user_id: str) -> MobileDashboardResponse:
"""Mobile dashboard — no balance, no promotions, simplified notifications."""
try:
tasks = [
asyncio.create_task(self._user.get_profile(user_id)),
asyncio.create_task(self._orders.get_recent_orders(user_id, limit=5)),
asyncio.create_task(self._recs.get_recommendations(user_id, limit=5)),
asyncio.create_task(self._notifications.has_notifications(user_id)),
]
user, orders, recs, has_notifs = await asyncio.gather(*tasks)
except Exception as e:
raise DashboardServiceError(f"Failed to compose mobile dashboard: {e}") from e
return MobileDashboardResponse(
user=user,
recent_orders=orders,
recommendations=recs,
has_notifications=has_notifs,
)
# ---- HELPERS ----
class DashboardServiceError(Exception):
"""Raised when dashboard composition fails due to critical source unavailability."""
def _extract_result(result: object) -> object:
if isinstance(result, Exception):
raise result
return result
async def _try_get_promotions(user_id: str) -> list[str]:
"""Best-effort promotion fetch — returns empty list on any failure.
Non-critical path; web dashboard degrades gracefully without promotions.
This is the BFF's responsibility to isolate client from downstream fragility.
"""
try:
# In production, call self._promotion_service.get_active(user_id)
return ["SUMMER2025"] # Placeholder for actual call
except Exception:
return []
Trade-off analysis: BFF adds a new service per client type, increasing operational overhead by 1-2 services. Use it when:
- Different clients need substantially different data shapes (web vs mobile)
- Client requires data from 3+ microservices to compose one screen
- You want to shield clients from downstream service evolution
Skip it for single-client apps or where one REST API suffices for all consumers. A well-designed API gateway can sometimes replace a BFF for simple cases.
Practical note: Start with one BFF for your primary client (usually web). Add mobile and partner BFFs only when the existing endpoints start causing measurable issues — large payloads on mobile, missing fields in partner integrations. The pattern pays for itself when composition complexity grows beyond 3 downstream calls per endpoint.
Pattern 3: Feature Flag / Toggle Pattern
Feature flags enable behavioral changes to be toggled without code deployment. They support gradual rollouts (percentage-based), audience targeting (user-based), time-based schedules, and kill switches for emergency disables. Combined with audit logging, they provide safe, reversible production changes.
Core principle: Every behavioral change controlled by a flag must have an associated expiry date or owner who reviews its usage. Stale flags are technical debt that obscures system behavior.
BAD: Hardcoded Boolean Checks With No Lifecycle Management
# ❌ BAD: Inline booleans scattered across codebase — impossible to audit
def process_payment(amount_cents: int, user_id: str) -> dict:
# Where is this defined? Who owns it? When does it expire?
USE_NEW_CHECKOUT = True # Hardcoded — never goes away
if USE_NEW_CHECKOUT:
return _process_via_new_gateway(amount_cents, user_id)
else:
return _process_via_legacy_gateway(amount_cents, user_id)
def calculate_discount(order_total: float, user_tier: str) -> float:
# Multiple inline flags with no central registry
ENABLE_VIP_DISCOUNT = True
USE_AI_PRICING = False
if not ENABLE_VIP_DISCOUNT:
return order_total * 0.10 # Old flat rate
if user_tier == "gold" or user_tier == "platinum":
base_discount = 0.25 if not USE_AI_PRICING else _ai_calculated_discount(order_total)
return order_total * (1 - base_discount)
return order_total
Problems with this approach:
- No way to toggle flags in production without redeploying code
- No audit trail — you cannot answer "who enabled this and when?"
- No expiry mechanism — flags accumulate as zombie code that nobody remembers
- Tests run both branches simultaneously, creating unpredictable behavior
GOOD: Centralized Flag Engine With Strategies And Audit Logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Optional
# ---- DOMAIN MODEL FOR FEATURE FLAGS ----
class RolloutStrategy(Enum):
"""How traffic is distributed across flag states."""
ALL_ON = "all_on"
ALL_OFF = "all_off"
PERCENTAGE = "percentage" # X% of users see the new behavior
USER_BASED = "user_based" # Specific user IDs or segments
TIME_BASED = "time_based" # On/off during specific windows
A/B_TEST = "ab_test" # Percentage split with experiment tracking
@dataclass(frozen=True)
class FlagCondition:
"""A single condition that must evaluate to True for the flag to be ON."""
strategy: RolloutStrategy
parameter: Any # Varies by strategy: int for percentage, list[str] for user IDs
@property
def is_always_on(self) -> bool:
return self.strategy == RolloutStrategy.ALL_ON
@property
def is_always_off(self) -> bool:
return self.strategy == RolloutStrategy.ALL_OFF
@dataclass
class FeatureFlag:
"""Central feature flag definition with lifecycle metadata."""
key: str
description: str
owner: str # Team or individual responsible for this flag
conditions: list[FlagCondition]
created_at: datetime = field(default_factory=datetime.utcnow)
expires_at: Optional[datetime] = None # Required — no permanent flags
is_enabled: bool = True # Kill switch
environment: str = "production" # Environment-scoped
def validate(self) -> None:
"""Ensure flag meets governance requirements."""
if not self.key or not self.description:
raise ValueError("Flag key and description are required")
if not self.owner:
raise ValueError("Every flag must have an owner")
if self.expires_at and self.expires_at < datetime.utcnow():
raise ValueError(f"Flag '{self.key}' has expired")
# ---- AUDIT LOGGING ----
@dataclass(frozen=True)
class FlagAuditEvent:
"""Immutable record of every flag evaluation for traceability."""
timestamp: datetime
flag_key: str
user_id: Optional[str]
evaluated_value: bool
reason: str # Why the flag was ON or OFF
# ---- FLAG EVALUATION ENGINE ----
class FlagStore(ABC):
"""Port: storage and retrieval of feature flags."""
@abstractmethod
def get_flag(self, key: str) -> Optional[FeatureFlag]: ...
@abstractmethod
def set_flag_state(self, key: str, is_enabled: bool, reason: str = "") -> None: ...
class InMemoryFlagStore(FlagStore):
"""Concrete in-memory implementation for tests and simple deployments."""
def __init__(self) -> None:
self._flags: dict[str, FeatureFlag] = {}
def get_flag(self, key: str) -> Optional[FeatureFlag]:
return self._flags.get(key)
def set_flag_state(self, key: str, is_enabled: bool, reason: str = "") -> None:
if key in self._flags:
flag = self._flags[key]
flag.is_enabled = is_enabled
# In production, this would persist to a database or config service
class FlagEvaluator:
"""Evaluates feature flags with rollout strategies and audit logging.
Thread-safe evaluation with deterministic user hashing for percentage-based rollouts.
"""
def __init__(self, flag_store: FlagStore) -> None:
self._store = flag_store
self._audit_log: list[FlagAuditEvent] = []
def is_enabled(
self,
flag_key: str,
user_id: Optional[str] = None,
metadata: Optional[dict[str, Any]] = None,
) -> bool:
"""Evaluate a feature flag for a specific user context.
Evaluation order:
1. Check if flag exists and is globally enabled (kill switch)
2. Evaluate conditions based on rollout strategy
3. Log the evaluation result to the audit trail
"""
flag = self._store.get_flag(flag_key)
# Guard: flag not found or globally disabled
if flag is None or not flag.is_enabled:
self._audit_log.append(FlagAuditEvent(
timestamp=datetime.utcnow(),
flag_key=flag_key,
user_id=user_id,
evaluated_value=False,
reason="flag_not_found_or_disabled",
))
return False
# Validate governance rules
try:
flag.validate()
except ValueError as e:
self._audit_log.append(FlagAuditEvent(
timestamp=datetime.utcnow(),
flag_key=flag_key,
user_id=user_id,
evaluated_value=False,
reason=f"governance_violation: {e}",
))
return False
# Evaluate against each condition — all must pass (AND logic)
for condition in flag.conditions:
if not self._evaluate_condition(condition, user_id, metadata or {}):
self._audit_log.append(FlagAuditEvent(
timestamp=datetime.utcnow(),
flag_key=flag_key,
user_id=user_id,
evaluated_value=False,
reason=f"condition_failed: {condition.strategy.value}",
))
return False
# All conditions passed — flag is ON
self._audit_log.append(FlagAuditEvent(
timestamp=datetime.utcnow(),
flag_key=flag_key,
user_id=user_id,
evaluated_value=True,
reason="all_conditions_met",
))
return True
def get_audit_log(self) -> list[FlagAuditEvent]:
"""Return the evaluation audit trail for this session."""
return list(self._audit_log)
# ---- STRATEGY EVALUATORS ----
def _evaluate_condition(
self,
condition: FlagCondition,
user_id: Optional[str],
metadata: dict[str, Any],
) -> bool:
"""Evaluate a single rollout strategy condition."""
match condition.strategy:
case RolloutStrategy.ALL_ON:
return True
case RolloutStrategy.ALL_OFF:
return False
case RolloutStrategy.PERCENTAGE:
percentage = condition.parameter # type: ignore[assignment]
if user_id is None:
return False
return self._hash_to_percentage(user_id) <= percentage
case RolloutStrategy.USER_BASED:
allowed_users = set(condition.parameter) # type: ignore[arg-type]
if user_id is None:
return False
return user_id in allowed_users
case RolloutStrategy.TIME_BASED:
start = metadata.get("start_time")
end = metadata.get("end_time")
now = datetime.utcnow()
if start and now < start:
return False
if end and now > end:
return False
return True
case RolloutStrategy.A_B_TEST:
group = condition.parameter # type: ignore[assignment]
if user_id is None:
return False
bucket = self._hash_to_bucket(user_id)
return bucket == group
case _:
return False
@staticmethod
def _hash_to_percentage(user_id: str) -> int:
"""Deterministic hash to [0, 99] for percentage-based rollouts."""
return abs(hash(user_id)) % 100
@staticmethod
def _hash_to_bucket(user_id: str) -> int:
"""Deterministic hash to group A or B (0 or 1)."""
return abs(hash(user_id)) % 2
# ---- USAGE EXAMPLE IN BUSINESS CODE ----
class PaymentProcessor:
"""Business code that delegates behavioral decisions to the flag engine.
The service has zero knowledge of how flags are stored — it only calls
evaluator.is_enabled() with a flag key and user context.
"""
def __init__(self, flag_evaluator: FlagEvaluator) -> None:
self._flags = flag_evaluator
def process(self, amount_cents: int, user_id: str) -> dict:
if self._flags.is_enabled("use_new_checkout_gateway", user_id=user_id):
return self._process_via_new_gateway(amount_cents, user_id)
else:
return self._process_via_legacy_gateway(amount_cents, user_id)
def _process_via_new_gateway(self, amount_cents: int, user_id: str) -> dict:
# New checkout flow — can be toggled without redeployment
return {"status": "processed", "gateway": "new", "amount_cents": amount_cents}
def _process_via_legacy_gateway(self, amount_cents: int, user_id: str) -> dict:
# Legacy checkout flow — deprecated but still active for some users
return {"status": "processed", "gateway": "legacy", "amount_cents": amount_cents}
Trade-off analysis: Feature flags add complexity to code paths (every flag introduces a branch). Use them when:
- You need to toggle behavior in production without deploys
- Running A/B tests or gradual percentage rollouts
- Building kill switches for high-risk features
Skip them for simple on/off configuration that rarely changes — environment variables and config files are lighter weight. Never use flags as a substitute for proper configuration management.
Practical note: Every flag must have an expires_at date. Set it to 30 days after creation. Add automated cleanup that disables expired flags and alerts their owners. Stale flags make debugging impossible because the code paths diverge permanently.
Pattern 4: CQRS with Event Sourcing Projections
Command Query Responsibility Segregation (CQRS) separates write operations (commands) from read operations (queries). When combined with event sourcing, every state mutation is stored as an immutable sequence of events. Read projections are built asynchronously by replaying these events into optimized materialized views. This pattern enables full audit trails, temporal queries, and independently scaling read and write models.
Core principle: Commands mutate state through events; queries read from denormalized projections that are derived from those events. The event store is the single source of truth — projections are disposable and rebuildable.
BAD: Monolithic ORM Model With All Reads and Writes Mixed
# ❌ BAD: Single model handles both complex writes and optimized reads
class OrderModel(Base): # SQLAlchemy model serves dual role
__tablename__ = "orders"
id = Column(Integer, primary_key=True)
status = Column(String(20))
items = Column(JSON) # JSON blob — hard to query individually
customer_id = Column(Integer)
total_cents = Column(Integer)
updated_at = Column(DateTime)
class OrderRepository:
"""Tries to serve every read pattern from one model."""
def get_order(self, order_id: int): # Simple lookup — fine
return self.session.query(OrderModel).get(order_id)
def get_orders_with_items_by_customer(
self, customer_id: int # Requires JOIN with separate items table
):
# Needs eager loading to avoid N+1 — performance degrades fast
return (
self.session.query(OrderModel)
.options(joinedload(OrderModel.items))
.filter(OrderModel.customer_id == customer_id)
.all()
)
def get_daily_revenue_report(self): # Aggregation query on same table
# This query scans the entire table and groups — competes with OLTP queries
return (
self.session.query(
func.date(OrderModel.updated_at).label("day"),
func.sum(OrderModel.total_cents).label("revenue")
)
.group_by("day")
.all()
)
def place_order(self, ...): # Write logic mixed into same repository
pass # Would contain validation, state transitions, persistence — all intertwined
This design fails when:
- Read queries compete with write transactions for the same database connection pool
- You need to audit who changed what and when — the model only stores current state
- Read and write scaling needs diverge (read-heavy analytics on write-heavy transactional data)
- Complex invariants span multiple aggregates
GOOD: CQRS With Event Sourcing And Projection-Based Reads
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Optional
# ---- EVENT SOURCING CORE (immutable events) ----
class DomainEvent:
"""Base class for all domain events. Immutable and timestamped."""
def __init__(self, aggregate_id: str, occurred_at: Optional[datetime] = None):
self.aggregate_id = aggregate_id
self.occurred_at = occurred_at or datetime.utcnow()
@dataclass(frozen=True)
class OrderCreated(DomainEvent):
"""Event: an order was created with initial state."""
customer_email: str
total_cents: int
currency: str = "USD"
@dataclass(frozen=True)
class OrderItemsAdded(DomainEvent):
"""Event: items were added to an existing order (before confirmation)."""
item_count: int
additional_cents: int
@dataclass(frozen=True)
class OrderConfirmed(DomainEvent):
"""Event: order moved to confirmed state."""
confirmed_at: datetime = field(default_factory=datetime.utcnow)
@dataclass(frozen=True)
class OrderCancelled(DomainEvent):
"""Event: order was cancelled."""
reason: str
cancelled_at: datetime = field(default_factory=datetime.utcnow)
# ---- AGGREGATE ROOT (rebuilds state from events) ----
class AggregateRoot:
"""Base class providing event sourcing infrastructure for aggregates."""
def __init__(self, aggregate_id: str):
self.aggregate_id = aggregate_id
self._domain_events: list[DomainEvent] = [] # Uncommitted events
@abstractmethod
def apply_event(self, event: DomainEvent) -> None: ...
def record_event(self, event: DomainEvent) -> None:
"""Record an event for later persistence and emit."""
self.apply_event(event)
self._domain_events.append(event)
@property
def domain_events(self) -> list[DomainEvent]:
"""Return uncommitted events to be persisted."""
return list(self._domain_events)
def clear_pending_events(self) -> None:
"""Clear committed events after persistence."""
self._domain_events.clear()
class Order(AggregateRoot):
"""Aggregate root that reconstructs state from events.
State is never set directly — it is always derived from applying events.
This ensures the event store is the single source of truth.
"""
def __init__(self, aggregate_id: str) -> None:
super().__init__(aggregate_id)
self.customer_email = ""
self.total_cents = 0
self.status = "pending"
self.confirmed_at: Optional[datetime] = None
self.cancelled_at: Optional[datetime] = None
# ---- COMMAND METHODS (these generate events, do not persist) ----
def create(
self,
customer_email: str,
total_cents: int,
currency: str = "USD",
) -> None:
if not customer_email or "@" not in customer_email:
raise ValueError("Invalid customer email")
if total_cents <= 0:
raise ValueError("Total must be positive")
self.customer_email = customer_email
self.total_cents = total_cents
self.currency = currency
self.record_event(OrderCreated(
aggregate_id=self.aggregate_id,
customer_email=customer_email,
total_cents=total_cents,
currency=currency,
))
def confirm(self) -> None:
if self.status != "pending":
raise ValueError(f"Cannot confirm order in '{self.status}' state")
self.status = "confirmed"
self.confirmed_at = datetime.utcnow()
self.record_event(OrderConfirmed(aggregate_id=self.aggregate_id))
def cancel(self, reason: str) -> None:
if self.status == "shipped":
raise ValueError("Cannot cancel a shipped order")
self.status = "cancelled"
self.cancelled_at = datetime.utcnow()
self.record_event(OrderCancelled(
aggregate_id=self.aggregate_id,
reason=reason,
))
# ---- EVENT APPLICATION (reconstructs state from each event type) ----
def apply_event(self, event: DomainEvent) -> None:
match event:
case OrderC
…(truncated)