Imported from shoshoavi/agentic_worflows (cursor/skills/clean-code-architecture/SKILL.md). Install upstream with npx skills add shoshoavi/agentic_worflows --skill clean-code-architecture. Copyright stays with the author.

Clean Code and Architecture Patterns

Apply these patterns to write maintainable, testable, and scalable code.

When to Use This Skill

• Designing new features or services
• Refactoring legacy code
• Building applications that will grow over time
• Writing code that others will maintain

1. Project Structure (Clean Architecture)

project/
├── src/
│   ├── domain/           # Core business logic (no dependencies)
│   │   ├── entities/     # Business objects
│   │   ├── services/     # Domain services
│   │   └── exceptions.py # Domain exceptions
│   │
│   ├── application/      # Use cases (depends only on domain)
│   │   ├── use_cases/    # Application logic
│   │   ├── interfaces/   # Abstract repositories, services
│   │   └── dto/          # Data transfer objects
│   │
│   ├── infrastructure/   # External implementations
│   │   ├── database/     # ORM, repositories
│   │   ├── api/          # External API clients
│   │   └── messaging/    # Queue implementations
│   │
│   └── presentation/     # Entry points
│       ├── api/          # REST/GraphQL handlers
│       ├── cli/          # Command line interface
│       └── web/          # Web interface
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
│
├── config/
└── scripts/

2. Repository Pattern

from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Sequence

T = TypeVar('T')

# Abstract interface (in application layer)
class Repository(ABC, Generic[T]):
    @abstractmethod
    async def get_by_id(self, id: int) -> T | None: ...
    
    @abstractmethod
    async def list(self, limit: int = 100, offset: int = 0) -> Sequence[T]: ...
    
    @abstractmethod
    async def save(self, entity: T) -> T: ...
    
    @abstractmethod
    async def delete(self, id: int) -> bool: ...


# Concrete implementation (in infrastructure layer)
class SQLAlchemyUserRepository(Repository[User]):
    def __init__(self, session: AsyncSession):
        self._session = session
    
    async def get_by_id(self, id: int) -> User | None:
        result = await self._session.execute(
            select(UserModel).where(UserModel.id == id)
        )
        row = result.scalar_one_or_none()
        return self._to_entity(row) if row else None
    
    async def save(self, user: User) -> User:
        model = self._to_model(user)
        self._session.add(model)
        await self._session.flush()
        return self._to_entity(model)
    
    def _to_entity(self, model: UserModel) -> User:
        return User(id=model.id, email=model.email, name=model.name)
    
    def _to_model(self, entity: User) -> UserModel:
        return UserModel(id=entity.id, email=entity.email, name=entity.name)

3. Use Case / Service Pattern

from dataclasses import dataclass

# Input DTO
@dataclass(frozen=True)
class CreateOrderRequest:
    user_id: int
    product_ids: list[int]
    shipping_address: str

# Output DTO
@dataclass(frozen=True)
class CreateOrderResponse:
    order_id: int
    total_amount: float
    estimated_delivery: str

# Use Case (application layer)
class CreateOrderUseCase:
    def __init__(
        self,
        user_repo: UserRepository,
        product_repo: ProductRepository,
        order_repo: OrderRepository,
        payment_service: PaymentService,
        notification_service: NotificationService,
    ):
        self._user_repo = user_repo
        self._product_repo = product_repo
        self._order_repo = order_repo
        self._payment = payment_service
        self._notifications = notification_service
    
    async def execute(self, request: CreateOrderRequest) -> CreateOrderResponse:
        # 1. Validate user exists
        user = await self._user_repo.get_by_id(request.user_id)
        if not user:
            raise UserNotFoundError(request.user_id)
        
        # 2. Get products and calculate total
        products = await self._product_repo.get_by_ids(request.product_ids)
        if len(products) != len(request.product_ids):
            raise ProductNotFoundError()
        
        total = sum(p.price for p in products)
        
        # 3. Create order
        order = Order(
            user_id=user.id,
            products=products,
            total=total,
            shipping_address=request.shipping_address,
        )
        
        # 4. Process payment
        await self._payment.charge(user, total)
        
        # 5. Save order
        saved_order = await self._order_repo.save(order)
        
        # 6. Send notification
        await self._notifications.send_order_confirmation(user, saved_order)
        
        return CreateOrderResponse(
            order_id=saved_order.id,
            total_amount=total,
            estimated_delivery=calculate_delivery_date(),
        )

4. Factory Pattern

from abc import ABC, abstractmethod
from enum import Enum

class NotificationType(Enum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"

class Notification(ABC):
    @abstractmethod
    async def send(self, user_id: int, message: str) -> None: ...

class EmailNotification(Notification):
    async def send(self, user_id: int, message: str) -> None:
        # Send email
        pass

class SMSNotification(Notification):
    async def send(self, user_id: int, message: str) -> None:
        # Send SMS
        pass

class NotificationFactory:
    _registry: dict[NotificationType, type[Notification]] = {
        NotificationType.EMAIL: EmailNotification,
        NotificationType.SMS: SMSNotification,
    }
    
    @classmethod
    def create(cls, notification_type: NotificationType) -> Notification:
        if notification_type not in cls._registry:
            raise ValueError(f"Unknown notification type: {notification_type}")
        return cls._registry[notification_type]()
    
    @classmethod
    def register(cls, type_: NotificationType, impl: type[Notification]) -> None:
        cls._registry[type_] = impl

5. Strategy Pattern

from abc import ABC, abstractmethod
from typing import Protocol

# Strategy interface
class PricingStrategy(Protocol):
    def calculate(self, base_price: float) -> float: ...

# Concrete strategies
class RegularPricing:
    def calculate(self, base_price: float) -> float:
        return base_price

class PremiumDiscount:
    def __init__(self, discount_percent: float = 10):
        self._discount = discount_percent / 100
    
    def calculate(self, base_price: float) -> float:
        return base_price * (1 - self._discount)

class BulkDiscount:
    def __init__(self, quantity: int, threshold: int = 10):
        self._quantity = quantity
        self._threshold = threshold
    
    def calculate(self, base_price: float) -> float:
        if self._quantity >= self._threshold:
            return base_price * 0.85  # 15% off
        return base_price

# Context
class PriceCalculator:
    def __init__(self, strategy: PricingStrategy):
        self._strategy = strategy
    
    def calculate(self, base_price: float) -> float:
        return self._strategy.calculate(base_price)

# Usage
calculator = PriceCalculator(PremiumDiscount(discount_percent=15))
final_price = calculator.calculate(100.0)  # 85.0

6. Decorator Pattern (Composition)

from abc import ABC, abstractmethod
from functools import wraps
import time
import logging

logger = logging.getLogger(__name__)

# Base interface
class DataFetcher(ABC):
    @abstractmethod
    async def fetch(self, key: str) -> dict: ...

# Core implementation
class APIDataFetcher(DataFetcher):
    async def fetch(self, key: str) -> dict:
        # Fetch from API
        return {"data": key}

# Decorator: Caching
class CachedFetcher(DataFetcher):
    def __init__(self, fetcher: DataFetcher, cache: dict[str, dict]):
        self._fetcher = fetcher
        self._cache = cache
    
    async def fetch(self, key: str) -> dict:
        if key in self._cache:
            return self._cache[key]
        result = await self._fetcher.fetch(key)
        self._cache[key] = result
        return result

# Decorator: Logging
class LoggedFetcher(DataFetcher):
    def __init__(self, fetcher: DataFetcher):
        self._fetcher = fetcher
    
    async def fetch(self, key: str) -> dict:
        logger.info(f"Fetching {key}")
        start = time.monotonic()
        result = await self._fetcher.fetch(key)
        elapsed = time.monotonic() - start
        logger.info(f"Fetched {key} in {elapsed:.2f}s")
        return result

# Decorator: Retry
class RetryFetcher(DataFetcher):
    def __init__(self, fetcher: DataFetcher, max_retries: int = 3):
        self._fetcher = fetcher
        self._max_retries = max_retries
    
    async def fetch(self, key: str) -> dict:
        for attempt in range(self._max_retries):
            try:
                return await self._fetcher.fetch(key)
            except Exception as e:
                if attempt == self._max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

# Compose decorators
cache: dict[str, dict] = {}
fetcher = LoggedFetcher(
    CachedFetcher(
        RetryFetcher(
            APIDataFetcher()
        ),
        cache
    )
)

7. Builder Pattern

from dataclasses import dataclass, field
from typing import Self

@dataclass
class HTTPRequest:
    method: str
    url: str
    headers: dict[str, str] = field(default_factory=dict)
    body: bytes | None = None
    timeout: int = 30

class HTTPRequestBuilder:
    def __init__(self):
        self._method = "GET"
        self._url = ""
        self._headers: dict[str, str] = {}
        self._body: bytes | None = None
        self._timeout = 30
    
    def method(self, method: str) -> Self:
        self._method = method
        return self
    
    def url(self, url: str) -> Self:
        self._url = url
        return self
    
    def header(self, key: str, value: str) -> Self:
        self._headers[key] = value
        return self
    
    def json_body(self, data: dict) -> Self:
        import json
        self._body = json.dumps(data).encode()
        self._headers["Content-Type"] = "application/json"
        return self
    
    def timeout(self, seconds: int) -> Self:
        self._timeout = seconds
        return self
    
    def build(self) -> HTTPRequest:
        if not self._url:
            raise ValueError("URL is required")
        return HTTPRequest(
            method=self._method,
            url=self._url,
            headers=self._headers,
            body=self._body,
            timeout=self._timeout,
        )

# Usage
request = (
    HTTPRequestBuilder()
    .method("POST")
    .url("https://api.example.com/users")
    .header("Authorization", "Bearer token")
    .json_body({"name": "John"})
    .timeout(60)
    .build()
)

8. Unit of Work Pattern

from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
from typing import AsyncIterator

class UnitOfWork(ABC):
    users: UserRepository
    orders: OrderRepository
    
    @abstractmethod
    async def commit(self) -> None: ...
    
    @abstractmethod
    async def rollback(self) -> None: ...

class SQLAlchemyUnitOfWork(UnitOfWork):
    def __init__(self, session_factory):
        self._session_factory = session_factory
    
    async def __aenter__(self) -> "SQLAlchemyUnitOfWork":
        self._session = self._session_factory()
        self.users = SQLAlchemyUserRepository(self._session)
        self.orders = SQLAlchemyOrderRepository(self._session)
        return self
    
    async def __aexit__(self, *args) -> None:
        await self.rollback()
        await self._session.close()
    
    async def commit(self) -> None:
        await self._session.commit()
    
    async def rollback(self) -> None:
        await self._session.rollback()

# Usage in use case
async def create_order_with_uow(uow: UnitOfWork, request: CreateOrderRequest):
    async with uow:
        user = await uow.users.get_by_id(request.user_id)
        order = Order(user_id=user.id, ...)
        await uow.orders.save(order)
        await uow.commit()

9. No Magic Values Rule

NEVER hardcode literal strings, numbers, or tuples inline in logic.

Think of it like a recipe -- you don't write "add 2.5" without saying "2.5 cups of flour". A number or string without a name is meaningless to the next developer (or your future self).

# BAD -- magic strings scattered in logic
if text.lower() in ("approve", "approved", "yes", "ok"):
    ...
say(text=f":x: Sorry <@{user_id}>, something went wrong.")

# GOOD -- named constants at module top
APPROVAL_KEYWORDS: frozenset[str] = frozenset({"approve", "approved", "yes", "ok"})
_MSG_ERROR_GENERIC: str = (
    ":x: Sorry <@{user_id}>, something went wrong. "
    "I've logged it for debugging."
)

if text.lower() in APPROVAL_KEYWORDS:
    ...
say(text=_MSG_ERROR_GENERIC.format(user_id=user_id))

Rules:

Visibility:

• _prefixed = private to the module
• No prefix = public (importable by tests / other modules)

Quick Reference: When to Use What