Skip to content

ADR 001: Hexagonal Architecture for Python SaaS Template

ADR 001: Hexagonal Architecture for Python SaaS Template

Status: βœ… Accepted Date: 2026-02-28 Authors: Seed & Source Team Decision Makers: Technical Architects


Context

The Problem

When building the python-saas template, we needed to choose an architecture that satisfies multiple critical requirements:

  1. Feature Injection: The template must support programmatic feature injection via AST manipulation without breaking existing code
  2. Testability: Business logic must be testable in isolation without spinning up databases, external APIs, or web servers
  3. Multiple Interfaces: Same business logic needs to be accessible from REST APIs, CLI commands, background jobs, and UI components
  4. Technology Independence: Users should be able to swap infrastructure components (PostgreSQL β†’ MongoDB, Stripe β†’ PactaPay) without rewriting business logic
  5. Clean Boundaries: Architecture must prevent β€œspaghetti code” and make dependencies explicit
  6. Maintainability: Code structure should be self-documenting and approachable for junior developers

Business Context

  • Alpha clients will extend templates with custom business logic specific to their domains
  • Stack CLI needs to inject features (Commerce, Auth, Payments, etc.) programmatically through code generation
  • Templates must scale from MVP prototypes to production systems without architectural rewrites
  • Multiple interface paradigms (REST, GraphQL, WebSockets, CLI) should share the same business logic
  • Junior developers should be able to understand the architecture from file structure alone

Technical Constraints

  • Must work with Python 3.11+
  • Should integrate with FastAPI, NiceGUI, Pydantic v2
  • Must support async/await patterns
  • Should enforce type safety with mypy strict mode
  • Must be Docker-ready and cloud-deployable

Decision

We adopt Hexagonal Architecture (also known as Ports & Adapters) for the python-saas template.

Architecture Overview

β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚ UI Layer (Adapters) β”‚
β”‚ "Driving Side" - Initiates Actions β”‚
β”‚ β”‚
β”‚ β€’ FastAPI Routes (REST API) β”‚
β”‚ β€’ NiceGUI Components (Web UI) β”‚
β”‚ β€’ CLI Commands (Click/Typer) β”‚
β”‚ β€’ Background Jobs (Celery/RQ) β”‚
β”‚ β€’ GraphQL Resolvers β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
β”‚
β”‚ calls
β–Ό
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚ Core Layer (Domain) β”‚
β”‚ "Business Logic & Rules" β”‚
β”‚ β”‚
β”‚ β€’ Entities (domain models) β”‚
β”‚ β€’ Use Cases (business operations) β”‚
β”‚ β€’ Interfaces/Ports (abstract contracts) β”‚
β”‚ β”‚
β”‚ βœ… Pure Python - No framework dependencies β”‚
β”‚ βœ… No imports from UI or Infrastructure β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
β”‚
β”‚ defines ports
β–Ό
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚ Infrastructure Layer (Adapters) β”‚
β”‚ "Driven Side" - Provides Capabilities β”‚
β”‚ β”‚
β”‚ β€’ SQLAlchemy Repositories (Database) β”‚
β”‚ β€’ Redis Cache (In-memory storage) β”‚
β”‚ β€’ PactaPay Gateway (Payment processing) β”‚
β”‚ β€’ SendGrid Client (Email service) β”‚
β”‚ β€’ S3 Storage (File uploads) β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜

The Dependency Rule

Critical: Dependencies MUST always point inward toward the Core.

βœ… ALLOWED:
UI β†’ Core
Infrastructure β†’ Core (implements interfaces)
❌ FORBIDDEN:
Core β†’ UI
Core β†’ Infrastructure

This is enforced through:

  • Import linting (mypy, ruff)
  • Code review checklists
  • CI/CD validation scripts

Directory Structure

python-saas/
└── src/
β”œβ”€β”€ core/ # Domain Layer (Business Logic)
β”‚ β”œβ”€β”€ entities/ # Domain models (Pydantic)
β”‚ β”‚ └── user.py # User entity
β”‚ β”œβ”€β”€ interfaces/ # Ports (Abstract contracts)
β”‚ β”‚ └── user_repository.py # IUserRepository (ABC)
β”‚ └── use_cases/ # Business operations
β”‚ └── register_user.py # RegisterUser use case
β”‚
β”œβ”€β”€ infrastructure/ # Driven Adapters
β”‚ └── persistence/
β”‚ └── memory_repository.py # InMemoryUserRepository
β”‚
└── ui/ # Driving Adapters
└── main.py # NiceGUI/FastAPI entry point

Implementation Details

Layer 1: Core Domain

Entities (Domain Models)

Pure Python domain objects with no infrastructure dependencies:

src/core/entities/user.py
from datetime import datetime
from typing import Optional
from uuid import UUID, uuid4
from pydantic import BaseModel, ConfigDict, Field
class BaseEntity(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: UUID = Field(default_factory=uuid4)
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)
class User(BaseEntity):
email: str
is_active: bool = True
full_name: Optional[str] = None

Key characteristics:

  • Uses Pydantic for validation (not SQLAlchemy models)
  • No database columns, foreign keys, or ORM relationships
  • Can be serialized, validated, and tested without any infrastructure

Interfaces (Ports)

Abstract contracts defining what the core needs from external systems:

src/core/interfaces/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
from uuid import UUID
from ..entities.user import User
class IUserRepository(ABC):
@abstractmethod
async def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
@abstractmethod
async def save(self, user: User) -> User:
pass
@abstractmethod
async def get_by_email(self, email: str) -> Optional[User]:
pass

Key characteristics:

  • Uses Python ABC (Abstract Base Class)
  • Defines method signatures only (no implementation)
  • Returns domain entities, not infrastructure objects

Use Cases (Business Logic)

Orchestrates business operations using injected ports:

src/core/use_cases/register_user.py
from uuid import uuid4
from ..entities.user import User
from ..interfaces.user_repository import IUserRepository
class RegisterUser:
def __init__(self, repository: IUserRepository):
self.repository = repository # Dependency injection
async def execute(self, email: str, full_name: str) -> User:
# Business rule: Check for duplicate email
existing = await self.repository.get_by_email(email)
if existing:
raise ValueError(f"User with email {email} already exists")
# Business rule: Create new user
user = User(
id=uuid4(),
email=email,
full_name=full_name,
is_active=True
)
return await self.repository.save(user)

Key characteristics:

  • No direct infrastructure imports (no SQLAlchemy, Redis, etc.)
  • Dependencies injected via constructor
  • Returns domain entities
  • Contains business rules and validation

Layer 2: Infrastructure

Adapters (Concrete Implementations)

Implements the ports defined in the core:

src/infrastructure/persistence/memory_repository.py
from typing import Optional
from uuid import UUID
from ...core.entities.user import User
from ...core.interfaces.user_repository import IUserRepository
class InMemoryUserRepository(IUserRepository):
def __init__(self):
self._store = {}
async def get_by_id(self, user_id: UUID) -> Optional[User]:
return self._store.get(user_id)
async def save(self, user: User) -> User:
self._store[user.id] = user
return user
async def get_by_email(self, email: str) -> Optional[User]:
for user in self._store.values():
if user.email == email:
return user
return None

Key characteristics:

  • Implements the interface from core
  • Contains all infrastructure-specific code (database, caching, API clients)
  • Can be swapped without changing core logic

Layer 3: UI (Driving Adapters)

Routes that wire up use cases with concrete adapters:

# src/ui/main.py (simplified)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from ..infrastructure.persistence.memory_repository import InMemoryUserRepository
from ..core.use_cases.register_user import RegisterUser
app = FastAPI()
# Dependency injection at application startup
user_repository = InMemoryUserRepository()
class RegisterRequest(BaseModel):
email: str
full_name: str
@app.post("/users/register")
async def register_user_endpoint(request: RegisterRequest):
use_case = RegisterUser(repository=user_repository)
try:
user = await use_case.execute(
email=request.email,
full_name=request.full_name
)
return {"id": str(user.id), "email": user.email}
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))

Key characteristics:

  • Wire dependencies together (adapter β†’ use case)
  • Handle HTTP-specific concerns (request/response, status codes)
  • Convert between DTOs and domain entities

Rationale

Why Hexagonal Architecture?

1. Testability Without Infrastructure

Problem: Traditional architectures require databases and external services for testing business logic.

Solution: Hexagonal architecture allows testing with mock implementations:

tests/test_register_user.py
import pytest
from src.core.entities.user import User
from src.core.use_cases.register_user import RegisterUser
from src.core.interfaces.user_repository import IUserRepository
class FakeUserRepository(IUserRepository):
def __init__(self):
self.users = []
async def get_by_email(self, email: str):
return next((u for u in self.users if u.email == email), None)
async def save(self, user: User):
self.users.append(user)
return user
@pytest.mark.asyncio
async def test_register_user_success():
# Arrange
repo = FakeUserRepository()
use_case = RegisterUser(repository=repo)
# Act
user = await use_case.execute(email="test@example.com", full_name="Test User")
# Assert
assert user.email == "test@example.com"
assert user.is_active is True
assert len(repo.users) == 1
@pytest.mark.asyncio
async def test_register_user_duplicate_email():
# Arrange
repo = FakeUserRepository()
await repo.save(User(email="test@example.com", full_name="Existing"))
use_case = RegisterUser(repository=repo)
# Act & Assert
with pytest.raises(ValueError, match="already exists"):
await use_case.execute(email="test@example.com", full_name="Duplicate")

Benefits:

  • βœ… No database required for unit tests
  • βœ… Tests run in milliseconds
  • βœ… 100% reproducible (no flaky tests)
  • βœ… Easy to test edge cases

2. Feature Injection via AST

Problem: Stack CLI needs to inject features (Commerce, Auth, Payments) into existing templates programmatically.

Solution: Hexagonal architecture provides predictable injection points:

templates/python-saas/src/core/entities/payment.py
# Stack CLI can inject by:
# 2. Add interface: templates/python-saas/src/core/interfaces/payment_gateway.py
# 3. Add use case: templates/python-saas/src/core/use_cases/process_payment.py
# 4. Add adapter: templates/python-saas/src/infrastructure/payment/pactapay_gateway.py
# 5. Add route: templates/python-saas/src/ui/routes/payments.py

Benefits:

  • βœ… Features are self-contained (one feature per folder tree)
  • βœ… No cross-feature dependencies (Commerce doesn’t import Auth)
  • βœ… AST knows exactly where to inject code
  • βœ… Rollback is simple (delete added files)

3. Technology Swapping

Problem: Users want to change infrastructure (PostgreSQL β†’ MongoDB, Stripe β†’ PactaPay) without rewriting business logic.

Solution: Swap adapter implementation without touching core:

# Before: Using in-memory storage
user_repository = InMemoryUserRepository()
# After: Using PostgreSQL
from sqlalchemy.ext.asyncio import create_async_engine
from ...infrastructure.persistence.sqlalchemy_repository import SQLAlchemyUserRepository
engine = create_async_engine("postgresql+asyncpg://...")
user_repository = SQLAlchemyUserRepository(engine)
# Use case remains UNCHANGED:
use_case = RegisterUser(repository=user_repository)

Benefits:

  • βœ… Core logic never changes
  • βœ… No regression in business rules
  • βœ… Can run both implementations side-by-side (useful for migrations)

4. Multiple Interface Support

Problem: Same business logic needed for REST API, CLI, background jobs, and UI components.

Solution: All adapters call the same use case:

# REST API
@app.post("/users/register")
async def register_api(request: RegisterRequest):
use_case = RegisterUser(repository=user_repository)
return await use_case.execute(**request.dict())
# CLI Command
@click.command()
@click.option('--email', required=True)
@click.option('--name', required=True)
def register_cli(email: str, name: str):
use_case = RegisterUser(repository=user_repository)
user = asyncio.run(use_case.execute(email=email, full_name=name))
print(f"βœ… Registered: {user.email}")
# Background Job
@celery.task
def register_background(email: str, name: str):
use_case = RegisterUser(repository=user_repository)
asyncio.run(use_case.execute(email=email, full_name=name))
# NiceGUI Component
def register_form():
async def on_submit():
use_case = RegisterUser(repository=user_repository)
user = await use_case.execute(email=email_input.value, full_name=name_input.value)
ui.notify(f"βœ… Registered {user.email}")
email_input = ui.input("Email")
name_input = ui.input("Full Name")
ui.button("Register", on_click=on_submit)

Benefits:

  • βœ… Business logic written once
  • βœ… Consistent behavior across all interfaces
  • βœ… Easy to add new interfaces (WebSockets, GraphQL, gRPC)

Consequences

Positive Outcomes

βœ… Separation of Concerns: Core logic has zero framework dependencies (no FastAPI, SQLAlchemy, Redis imports)

βœ… Testability: 100% unit test coverage of business logic without infrastructure

βœ… Feature Independence: Commerce feature doesn’t know Auth exists (loose coupling prevents conflicts)

βœ… Maintainability: File structure is self-documenting (entities/, use_cases/, interfaces/)

βœ… Onboarding: Junior developers understand architecture from folder names alone

βœ… Framework Agnostic: Can migrate from FastAPI β†’ Flask β†’ Django without touching core

βœ… Injection-Friendly: AST manipulation has predictable targets for feature injection

βœ… Technology Swapping: Replace PostgreSQL with MongoDB by changing one adapter

βœ… Multiple Interfaces: Same use case called from REST, CLI, jobs, UI without duplication

Negative Outcomes

⚠️ Increased Boilerplate: Each feature requires 4+ files (entity, interface, use case, adapter, route)

⚠️ Learning Curve: Junior developers must learn dependency inversion principle

⚠️ Indirection: More layers means more navigation (entity β†’ interface β†’ adapter β†’ route)

⚠️ Over-Engineering Risk: Simple CRUD operations feel heavyweight with full hexagonal pattern

⚠️ Initial Development Slower: More upfront design required compared to β€œjust ship it” approach

Mitigation Strategies

For Boilerplate:

  • Provide code generators: sscli generate use-case ProcessPayment
  • Document common patterns in templates
  • Include working examples in every injected feature

For Learning Curve:

  • Comprehensive onboarding documentation (HEXAGONAL_PYTHON.md)
  • Video walkthroughs of architecture
  • Code comments explaining patterns
  • Pair programming sessions for alpha clients

For Over-Engineering:

  • Allow pragmatic shortcuts for simple CRUD (routes can skip use case layer)
  • Document when to use full hexagonal vs simplified approach
  • Provide β€œlite” mode for prototyping

For Initial Slowdown:

  • Pre-built use case templates
  • AST injection handles boilerplate automatically
  • Emphasize long-term maintainability over short-term velocity

Alternatives Considered

Alternative 1: Traditional MVC (Model-View-Controller)

Structure:

mvc-example/
models/ # Database models (SQLAlchemy)
views/ # Templates or API serializers
controllers/ # Request handlers with business logic

Pros:

  • βœ… Familiar pattern (Django, Rails, Laravel)
  • βœ… Less boilerplate (fewer files per feature)
  • βœ… Faster initial development
  • βœ… Well-documented (thousands of tutorials)

Cons:

  • ❌ Business logic scattered between controllers and models
  • ❌ Models tightly coupled to database (hard to test)
  • ❌ Controllers grow into β€œgod objects” (fat controller anti-pattern)
  • ❌ No clear boundary between domain and infrastructure
  • ❌ Feature injection breaks MVC boundaries

Why Rejected: Poor separation of concerns makes feature injection and testing difficult. Business logic mixed with framework code prevents technology swapping.


Alternative 2: Layered Architecture (3-Tier)

Structure:

layered-example/
presentation/ # UI, API routes
business/ # Business logic
data/ # Database access

Pros:

  • βœ… Clear separation into horizontal layers
  • βœ… Simple mental model (top to bottom data flow)
  • βœ… Well-understood pattern

Cons:

  • ❌ Business layer often depends on data layer (imports ORM models)
  • ❌ Testing requires database (can’t mock data layer easily)
  • ❌ Technology changes require rewriting business logic
  • ❌ No dependency inversion (business knows about database)

Why Rejected: Business layer typically imports from data layer, creating tight coupling. Hexagonal inverts this dependency through interfaces, enabling true testability.


Alternative 3: Microservices Architecture

Structure:

microservices-example/
user-service/ #독립 User API
payment-service/ # 독립 Payment API
auth-service/ # 독립 Auth API

Pros:

  • βœ… Independent deployment and scaling
  • βœ… Technology diversity (Python + Go + Node.js)
  • βœ… Team autonomy (each service owned by separate team)

Cons:

  • ❌ Massive operational overhead (Kubernetes, service mesh, monitoring)
  • ❌ Network latency between services
  • ❌ Distributed transactions and eventual consistency
  • ❌ Overkill for templates (most users don’t need microservices)
  • ❌ Feature injection across services is extremely complex
  • ❌ Debugging and testing requires running multiple services

Why Rejected: Far too complex for MVP templates. Operational burden (Docker Compose, Kubernetes, service discovery) is inappropriate for 80% of use cases. Microservices should be an evolutionary architecture, not a starting point.


Alternative 4: Clean Architecture (Uncle Bob)

Structure:

clean-example/
entities/ # Enterprise business rules
use_cases/ # Application business rules
interface_adapters/ # Controllers, presenters, gateways
frameworks_drivers/ # UI, DB, Web

Pros:

  • βœ… Same principles as Hexagonal (dependency inversion)
  • βœ… Extremely well-documented (books, courses, talks)
  • βœ… Clear dependency rule

Cons:

  • ❌ More layers than Hexagonal (4+ layers vs 3 layers)
  • ❌ Additional complexity (entities vs use case-specific models)
  • ❌ More boilerplate (even more files per feature)
  • ❌ May confuse developers (what’s the difference between entities and use case models?)

Why Rejected: Clean Architecture achieves the same goals as Hexagonal but with more layers. For our use case (injectable templates), the additional layers add complexity without proportional benefits.


Implementation Guidelines

For Template Users

When to Use Full Hexagonal Pattern

Use full pattern (entity β†’ interface β†’ use case β†’ adapter β†’ route) for:

  • βœ… Complex business logic with multiple rules
  • βœ… Operations requiring thorough unit testing
  • βœ… Features that may swap infrastructure (payment gateways, email services)
  • βœ… Multi-step workflows (checkout, onboarding, approval processes)
  • βœ… Features with high business value (revenue-generating)

When to Simplify

Skip use case layer for:

  • βœ… Simple CRUD operations (GET /users, POST /users)
  • βœ… MVP prototyping (iterate fast, refactor later)
  • βœ… Internal tools with low change frequency
  • βœ… Read-only queries without business rules

Example of simplified approach:

# Direct route to repository (no use case)
@app.get("/users/{user_id}")
async def get_user(user_id: UUID):
user = await user_repository.get_by_id(user_id)
if not user:
raise HTTPException(404, "User not found")
return user

For Feature Developers

Must-Follow Rules

  1. Core Never Imports Infrastructure or UI

    # ❌ WRONG: Core importing from infrastructure
    from src.infrastructure.persistence.sqlalchemy_models import UserModel
    # βœ… CORRECT: Core imports only from core
    from src.core.entities.user import User
  2. Use Cases Return Domain Entities, Not DTOs

    # ❌ WRONG: Returning API response model
    async def execute() -> UserResponse:
    return UserResponse(id=1, email="...")
    # βœ… CORRECT: Returning domain entity
    async def execute() -> User:
    return User(id=uuid4(), email="...")
  3. Ports Defined in Core, Implemented in Infrastructure

    src/core/interfaces/email_service.py
    # βœ… CORRECT: ABC in core
    class IEmailService(ABC):
    @abstractmethod
    async def send(self, to: str, subject: str, body: str):
    pass
    # βœ… CORRECT: Implementation in infrastructure
    # src/infrastructure/email/sendgrid_service.py
    class SendGridEmailService(IEmailService):
    async def send(self, to: str, subject: str, body: str):
    # SendGrid-specific code here
    pass
  4. Dependency Injection via Constructor

    # ❌ WRONG: Hardcoded dependency
    class ProcessPayment:
    def __init__(self):
    self.gateway = PactaPayGateway() # Tight coupling!
    # βœ… CORRECT: Injected dependency
    class ProcessPayment:
    def __init__(self, gateway: IPaymentGateway):
    self.gateway = gateway # Loose coupling via interface

Validation and Enforcement

Automated Checks

Terminal window
# Check for forbidden imports in core
grep -r "from.*infrastructure" src/core/
grep -r "from.*ui" src/core/
# Should return 0 results

CI/CD Pipeline

.github/workflows/architecture-lint.yml
- name: Validate Hexagonal Architecture
run: |
python bin/validate_architecture.py
# Fails build if core imports from infrastructure/ui

Code Review Checklist

  • Core files have no imports from infrastructure/ or ui/
  • New interfaces defined as ABCs in core/interfaces/
  • Use cases receive dependencies via constructor
  • Domain entities use Pydantic, not SQLAlchemy models
  • Routes map between DTOs and domain entities

References

Academic & Industry Sources

Internal Documentation


Review & Approval

PhaseDateReviewersStatus
Proposed2026-02-15Architecture Teamβœ… Draft Complete
Technical Review2026-02-20Lead Engineersβœ… Approved with Minor Revisions
Security Review2026-02-22Security Teamβœ… No Security Concerns
Final Approval2026-02-28CTOβœ… Accepted

Next Review: 2026-08-28 (6 months)


Changelog

DateAuthorChange
2026-02-15Architecture TeamInitial draft with context and alternatives
2026-02-20Lead EngineersAdded real code examples from python-saas template
2026-02-22Security TeamAdded validation and enforcement sections
2026-02-28CTOFinal approval and acceptance

Appendix: Complete Working Example

Scenario: Payment Processing Feature

src/core/entities/payment.py
# ============================================================
# 1. ENTITY (Core Domain Model)
# ============================================================
from uuid import UUID, uuid4
from decimal import Decimal
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field
class PaymentStatus(str, Enum):
PENDING = "pending"
COMPLETED = "completed"
FAILED = "failed"
class Payment(BaseModel):
id: UUID = Field(default_factory=uuid4)
amount: Decimal
currency: str = "USD"
status: PaymentStatus = PaymentStatus.PENDING
user_id: UUID
created_at: datetime = Field(default_factory=datetime.utcnow)
# ============================================================
# 2. INTERFACE (Port)
# ============================================================
# src/core/interfaces/payment_gateway.py
from abc import ABC, abstractmethod
from ..entities.payment import Payment
class IPaymentGateway(ABC):
@abstractmethod
async def charge(self, payment: Payment) -> Payment:
"""Process payment and return updated payment with status"""
pass
# ============================================================
# 3. USE CASE (Business Logic)
# ============================================================
# src/core/use_cases/process_payment.py
from decimal import Decimal
from uuid import UUID
from ..entities.payment import Payment, PaymentStatus
from ..interfaces.payment_gateway import IPaymentGateway
class ProcessPayment:
def __init__(self, gateway: IPaymentGateway):
self.gateway = gateway
async def execute(self, user_id: UUID, amount: Decimal) -> Payment:
# Business rule: Minimum payment amount
if amount < Decimal("1.00"):
raise ValueError("Payment amount must be at least $1.00")
# Business rule: Maximum payment amount
if amount > Decimal("10000.00"):
raise ValueError("Payment amount cannot exceed $10,000")
# Create payment
payment = Payment(user_id=user_id, amount=amount)
# Process through gateway
return await self.gateway.charge(payment)
# ============================================================
# 4. ADAPTER (Infrastructure Implementation)
# ============================================================
# src/infrastructure/payment/pactapay_gateway.py
import httpx
from ...core.entities.payment import Payment, PaymentStatus
from ...core.interfaces.payment_gateway import IPaymentGateway
class PactaPayGateway(IPaymentGateway):
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
async def charge(self, payment: Payment) -> Payment:
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/charges",
json={
"amount": str(payment.amount),
"currency": payment.currency,
"user_id": str(payment.user_id)
},
headers={"Authorization": f"Bearer {self.api_key}"}
)
if response.status_code == 200:
payment.status = PaymentStatus.COMPLETED
else:
payment.status = PaymentStatus.FAILED
return payment
# ============================================================
# 5. ROUTE (UI Layer)
# ============================================================
# src/ui/routes/payments.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from decimal import Decimal
from uuid import UUID
from ...infrastructure.payment.pactapay_gateway import PactaPayGateway
from ...core.use_cases.process_payment import ProcessPayment
router = APIRouter(prefix="/payments")
# Dependency injection at module level
payment_gateway = PactaPayGateway(
api_key="pk_test_xxx",
base_url="https://api.pactapay.com"
)
class PaymentRequest(BaseModel):
user_id: UUID
amount: Decimal
@router.post("/")
async def create_payment(request: PaymentRequest):
use_case = ProcessPayment(gateway=payment_gateway)
try:
payment = await use_case.execute(
user_id=request.user_id,
amount=request.amount
)
return {
"id": str(payment.id),
"status": payment.status,
"amount": str(payment.amount)
}
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
# ============================================================
# 6. TEST (Isolated Unit Test)
# ============================================================
# tests/test_process_payment.py
import pytest
from decimal import Decimal
from uuid import uuid4
from src.core.entities.payment import Payment, PaymentStatus
from src.core.use_cases.process_payment import ProcessPayment
from src.core.interfaces.payment_gateway import IPaymentGateway
class FakePaymentGateway(IPaymentGateway):
async def charge(self, payment: Payment) -> Payment:
payment.status = PaymentStatus.COMPLETED
return payment
@pytest.mark.asyncio
async def test_process_payment_success():
gateway = FakePaymentGateway()
use_case = ProcessPayment(gateway=gateway)
payment = await use_case.execute(
user_id=uuid4(),
amount=Decimal("100.00")
)
assert payment.status == PaymentStatus.COMPLETED
assert payment.amount == Decimal("100.00")
@pytest.mark.asyncio
async def test_process_payment_minimum_amount():
gateway = FakePaymentGateway()
use_case = ProcessPayment(gateway=gateway)
with pytest.raises(ValueError, match="at least"):
await use_case.execute(user_id=uuid4(), amount=Decimal("0.50"))

This complete example demonstrates:

  • βœ… Clear separation of concerns across all layers
  • βœ… Testability without external dependencies
  • βœ… Dependency injection through interfaces
  • βœ… Business rules isolated in use case
  • βœ… Infrastructure swappable (PactaPay β†’ Stripe with no core changes)

End of ADR 001