home / skills / yonatangross / orchestkit / clean-architecture

clean-architecture skill

safe

/plugins/ork/skills/clean-architecture

This skill helps you implement clean architecture with SOLID and hexagonal patterns to build maintainable, testable backends.

npx playbooks add skill yonatangross/orchestkit --skill clean-architecture

Review the files below or copy the command above to add this skill to your agents.

Files (5)

SKILL.md

11.1 KB

---
name: clean-architecture
description: SOLID principles, hexagonal architecture, ports and adapters, and DDD tactical patterns for maintainable backends. Use when implementing clean architecture, decoupling services, separating domain logic, or creating testable architecture.
context: fork
agent: code-quality-reviewer
version: 1.0.0
tags: [architecture, solid, hexagonal, ddd, python, fastapi, 2026]
author: OrchestKit
user-invocable: false
---

# Clean Architecture Patterns

Build maintainable, testable backends with SOLID principles and hexagonal architecture.

## SOLID Principles (2026 Python)

### S - Single Responsibility

```python
# BAD: One class doing everything
class UserManager:
    def create_user(self, data): ...
    def send_welcome_email(self, user): ...
    def generate_report(self, users): ...

# GOOD: Separate responsibilities
class UserService:
    def create_user(self, data: UserCreate) -> User: ...

class EmailService:
    def send_welcome(self, user: User) -> None: ...

class ReportService:
    def generate_user_report(self, users: list[User]) -> Report: ...
```

### O - Open/Closed (Protocol-based)

```python
from typing import Protocol

class PaymentProcessor(Protocol):
    async def process(self, amount: Decimal) -> PaymentResult: ...

class StripeProcessor:
    async def process(self, amount: Decimal) -> PaymentResult:
        # Stripe implementation
        ...

class PayPalProcessor:
    async def process(self, amount: Decimal) -> PaymentResult:
        # PayPal implementation - extends without modifying
        ...
```

### L - Liskov Substitution

```python
# Any implementation of Repository can substitute another
class IUserRepository(Protocol):
    async def get_by_id(self, id: str) -> User | None: ...
    async def save(self, user: User) -> User: ...

class PostgresUserRepository:
    async def get_by_id(self, id: str) -> User | None: ...
    async def save(self, user: User) -> User: ...

class InMemoryUserRepository:  # For testing - fully substitutable
    async def get_by_id(self, id: str) -> User | None: ...
    async def save(self, user: User) -> User: ...
```

### I - Interface Segregation

```python
# BAD: Fat interface
class IRepository(Protocol):
    async def get(self, id: str): ...
    async def save(self, entity): ...
    async def delete(self, id: str): ...
    async def search(self, query: str): ...
    async def bulk_insert(self, entities): ...

# GOOD: Segregated interfaces
class IReader(Protocol):
    async def get(self, id: str) -> T | None: ...

class IWriter(Protocol):
    async def save(self, entity: T) -> T: ...

class ISearchable(Protocol):
    async def search(self, query: str) -> list[T]: ...
```

### D - Dependency Inversion

```python
from typing import Protocol
from fastapi import Depends

class IAnalysisRepository(Protocol):
    async def get_by_id(self, id: str) -> Analysis | None: ...

class AnalysisService:
    def __init__(self, repo: IAnalysisRepository):
        self._repo = repo  # Depends on abstraction, not concrete

# FastAPI DI
def get_analysis_service(
    db: AsyncSession = Depends(get_db)
) -> AnalysisService:
    repo = PostgresAnalysisRepository(db)
    return AnalysisService(repo)
```

## Hexagonal Architecture (Ports & Adapters)

```
┌─────────────────────────────────────────────────────────────┐
│                      DRIVING ADAPTERS                        │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐    │
│  │ FastAPI  │  │   CLI    │  │  Celery  │  │  Tests   │    │
│  │ Routes   │  │ Commands │  │  Tasks   │  │  Mocks   │    │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘    │
│       │             │             │             │           │
│       ▼             ▼             ▼             ▼           │
│  ╔═══════════════════════════════════════════════════════╗  │
│  ║                    INPUT PORTS                        ║  │
│  ║  ┌─────────────────┐  ┌─────────────────────────────┐ ║  │
│  ║  │ AnalysisService │  │ UserService                 │ ║  │
│  ║  │ (Use Cases)     │  │ (Use Cases)                 │ ║  │
│  ║  └────────┬────────┘  └──────────────┬──────────────┘ ║  │
│  ╠═══════════╪══════════════════════════╪════════════════╣  │
│  ║           ▼          DOMAIN          ▼                ║  │
│  ║  ┌─────────────────────────────────────────────────┐  ║  │
│  ║  │  Entities  │  Value Objects  │  Domain Events   │  ║  │
│  ║  │  Analysis  │  AnalysisType   │  AnalysisCreated │  ║  │
│  ║  └─────────────────────────────────────────────────┘  ║  │
│  ╠═══════════════════════════════════════════════════════╣  │
│  ║                   OUTPUT PORTS                        ║  │
│  ║  ┌──────────────────┐  ┌────────────────────────────┐ ║  │
│  ║  │ IAnalysisRepo    │  │ INotificationService       │ ║  │
│  ║  │ (Protocol)       │  │ (Protocol)                 │ ║  │
│  ║  └────────┬─────────┘  └──────────────┬─────────────┘ ║  │
│  ╚═══════════╪══════════════════════════╪════════════════╝  │
│              ▼                          ▼                   │
│  ┌───────────────────┐  ┌────────────────────────────────┐ │
│  │ PostgresRepo      │  │ EmailNotificationService       │ │
│  │ (SQLAlchemy)      │  │ (SMTP/SendGrid)                │ │
│  └───────────────────┘  └────────────────────────────────┘ │
│                      DRIVEN ADAPTERS                        │
└─────────────────────────────────────────────────────────────┘
```

## DDD Tactical Patterns

### Entity (Identity-based)

```python
from dataclasses import dataclass, field
from uuid import UUID, uuid4

@dataclass
class Analysis:
    id: UUID = field(default_factory=uuid4)
    source_url: str
    status: AnalysisStatus
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Analysis):
            return False
        return self.id == other.id  # Identity equality
```

### Value Object (Structural equality)

```python
from dataclasses import dataclass

@dataclass(frozen=True)  # Immutable
class AnalysisType:
    category: str
    depth: int

    def __post_init__(self):
        if self.depth < 1 or self.depth > 3:
            raise ValueError("Depth must be 1-3")
```

### Aggregate Root

```python
class AnalysisAggregate:
    def __init__(self, analysis: Analysis, artifacts: list[Artifact]):
        self._analysis = analysis
        self._artifacts = artifacts
        self._events: list[DomainEvent] = []

    def complete(self, summary: str) -> None:
        self._analysis.status = AnalysisStatus.COMPLETED
        self._analysis.summary = summary
        self._events.append(AnalysisCompleted(self._analysis.id))

    def collect_events(self) -> list[DomainEvent]:
        events = self._events.copy()
        self._events.clear()
        return events
```

## Directory Structure

```
backend/app/
├── api/v1/              # Driving adapters (FastAPI routes)
├── domains/
│   └── analysis/
│       ├── entities.py      # Domain entities
│       ├── value_objects.py # Value objects
│       ├── services.py      # Domain services (use cases)
│       ├── repositories.py  # Output port protocols
│       └── events.py        # Domain events
├── infrastructure/
│   ├── repositories/    # Driven adapters (PostgreSQL)
│   ├── services/        # External service adapters
│   └── messaging/       # Event publishers
└── core/
    ├── dependencies.py  # FastAPI DI configuration
    └── protocols.py     # Shared protocols
```

## Anti-Patterns (FORBIDDEN)

```python
# NEVER import infrastructure in domain
from app.infrastructure.database import engine  # In domain layer

# NEVER leak ORM models to API
@router.get("/users/{id}")
async def get_user(id: str, db: Session) -> UserModel:  # Returns ORM model
    return db.query(UserModel).get(id)

# NEVER have domain depend on framework
from fastapi import HTTPException
class UserService:
    def get(self, id: str):
        if not user:
            raise HTTPException(404)  # Framework in domain!
```

## Key Decisions

| Decision | Recommendation |
|----------|----------------|
| Protocol vs ABC | Use Protocol (structural typing) |
| Dataclass vs Pydantic | Dataclass for domain, Pydantic for API |
| Repository granularity | One per aggregate root |
| Transaction boundary | Service layer, not repository |
| Event publishing | Collect in aggregate, publish after commit |

## Related Skills

- `repository-patterns` - Detailed repository implementations
- `api-design-framework` - REST API patterns
- `database-schema-designer` - Schema design

## Capability Details

### solid-principles
**Keywords:** SOLID, single responsibility, open closed, liskov, interface segregation, dependency inversion
**Solves:**
- How do I apply SOLID principles in Python?
- My classes are doing too much

### hexagonal-architecture
**Keywords:** hexagonal, ports and adapters, clean architecture, onion
**Solves:**
- How do I structure my FastAPI app?
- How to separate infrastructure from domain?

### ddd-tactical
**Keywords:** entity, value object, aggregate, domain event, DDD
**Solves:**
- What's the difference between entity and value object?
- How to design aggregates?

Overview

This skill teaches how to build maintainable, testable backends using SOLID principles, hexagonal (ports-and-adapters) architecture, and DDD tactical patterns. It focuses on separating domain logic from infrastructure, creating clear ports and adapters, and designing aggregates, entities, and value objects. Use it to make services decoupled, testable, and easy to evolve.

How this skill works

The skill describes concrete patterns and code-level guidance: single-responsibility services, protocol-based abstractions for Open/Closed, and dependency inversion via interfaces and DI. It maps input ports (use cases) to domain entities and output ports (repositories, notification services) implemented by driven adapters. It also prescribes DDD tactical patterns—entities, immutable value objects, and aggregate roots that collect domain events.

When to use it

Implementing a new backend or refactoring an existing monolith
Decoupling business logic from frameworks and external services
Designing testable services and swap-in test doubles (in-memory repos, mocks)
Defining transaction boundaries and publishing domain events safely
Creating a maintainable folder layout for teams (domains, infrastructure, api)

Best practices

Keep domain pure: no framework imports or ORM leakage into domain types
Depend on protocols/abstractions, not concrete adapters; wire implementations in the DI layer
Split responsibilities: one service/class per responsibility, small interfaces (IReader/IWriter)
Use dataclasses (or plain domain models) for domain and Pydantic only at API boundaries
Collect events in aggregates and publish them after transaction commit to avoid side-effects in domain methods

Example use cases

FastAPI app where routes are driving adapters and services are input ports
Replacing Postgres repo with an in-memory repo for fast tests without changing domain code
Modeling an Analysis aggregate that updates status and emits AnalysisCompleted events
Adding a new payment processor by implementing a PaymentProcessor protocol without changing core logic
Designing a search feature by separating reader, writer, and searchable interfaces to avoid fat repositories

FAQ

Should domain models use Pydantic or dataclasses?

Prefer dataclasses or plain domain models inside the domain. Use Pydantic at the API boundary for validation and serialization.

Where should I publish domain events?

Have aggregates collect events and publish them after the transaction commits, usually in the service layer or an outbox mechanism.

How do I test without touching the database?

Provide protocol-based repositories and supply in-memory or mocked implementations in tests so domain and service tests remain fast and deterministic.