Clean Architecture

+-------------------------------------+
|  Infrastructure (Frameworks, DB)   |  <- Outer layer
+-------------------------------------+
|  Adapters (Controllers, Repos)     |
+-------------------------------------+
|  Use Cases (Application Logic)     |
+-------------------------------------+
|  Domain (Entities, Value Objects)  |  <- Inner layer
+-------------------------------------+

myapp/
+-- domain/                    # Inner layer - no external deps
|   +-- entities/             # Business entities
|   +-- value_objects/        # Immutable value objects
|   +-- events/               # Domain events
|   +-- repositories/         # Abstract repository interfaces (ports)
+-- use_cases/                # Application layer
+-- adapters/                 # Interface adapters
|   +-- repositories/         # Repository implementations
|   +-- controllers/          # API controllers
+-- infrastructure/           # Framework & external concerns
|   +-- database.py          # Database configuration
|   +-- container.py         # Dependency injection container
|   +-- config.py            # Application settings
+-- main.py                  # Application entry point

# domain/value_objects/email.py
from dataclasses import dataclass
import re

@dataclass(frozen=True)
class Email:
    value: str
    def __post_init__(self):
        if not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', self.value):
            raise ValueError(f"Invalid email: {self.value}")
    def __str__(self) -> str:
        return self.value


# domain/entities/user.py
from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID, uuid4
from domain.value_objects.email import Email

@dataclass
class User:
    email: Email
    name: str
    id: UUID = field(default_factory=uuid4)
    is_active: bool = True
    created_at: datetime = field(default_factory=datetime.utcnow)

    def deactivate(self) -> None:
        self.is_active = False

    def can_login(self) -> bool:
        return self.is_active

    @classmethod
    def create(cls, email: Email, name: str) -> "User":
        return cls(email=email, name=name)

# domain/repositories/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
from uuid import UUID
from domain.entities.user import User
from domain.value_objects.email import Email

class IUserRepository(ABC):
    @abstractmethod
    async def find_by_id(self, user_id: UUID) -> Optional[User]: ...
    @abstractmethod
    async def find_by_email(self, email: Email) -> Optional[User]: ...
    @abstractmethod
    async def save(self, user: User) -> User: ...
    @abstractmethod
    async def delete(self, user_id: UUID) -> bool: ...

# use_cases/create_user.py
from dataclasses import dataclass
from typing import Optional
from uuid import UUID
from domain.entities.user import User
from domain.value_objects.email import Email
from domain.repositories.user_repository import IUserRepository

@dataclass
class CreateUserRequest:
    email: str
    name: str

@dataclass
class CreateUserResponse:
    user_id: Optional[UUID]
    success: bool
    error_message: Optional[str] = None

class CreateUserUseCase:
    def __init__(self, user_repository: IUserRepository):
        self._user_repository = user_repository

    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
        try:
            email = Email(request.email)
        except ValueError as e:
            return CreateUserResponse(None, False, str(e))
        if await self._user_repository.find_by_email(email):
            return CreateUserResponse(None, False, "Email already registered")
        user = User.create(email=email, name=request.name)
        saved = await self._user_repository.save(user)
        return CreateUserResponse(saved.id, True)

# adapters/repositories/sqlalchemy_user_repository.py
from typing import Optional
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from domain.entities.user import User
from domain.value_objects.email import Email
from domain.repositories.user_repository import IUserRepository

class SQLAlchemyUserRepository(IUserRepository):
    def __init__(self, session: AsyncSession):
        self._session = session

    async def find_by_id(self, user_id: UUID) -> Optional[User]:
        result = await self._session.execute(
            select(UserModel).where(UserModel.id == user_id)
        )
        row = result.scalar_one_or_none()
        return self._to_entity(row) if row else None

    async def find_by_email(self, email: Email) -> Optional[User]:
        result = await self._session.execute(
            select(UserModel).where(UserModel.email == str(email))
        )
        row = result.scalar_one_or_none()
        return self._to_entity(row) if row else None

    async def save(self, user: User) -> User:
        model = UserModel(
            id=user.id, email=str(user.email), name=user.name,
            is_active=user.is_active, created_at=user.created_at
        )
        self._session.add(model)
        await self._session.commit()
        return user

    def _to_entity(self, model) -> User:
        return User(
            id=model.id, email=Email(model.email), name=model.name,
            is_active=model.is_active, created_at=model.created_at
        )

# infrastructure/container.py
from dependency_injector import containers, providers
from adapters.repositories.sqlalchemy_user_repository import SQLAlchemyUserRepository
from use_cases.create_user import CreateUserUseCase
from infrastructure.database import get_session

class Container(containers.DeclarativeContainer):
    db_session = providers.Factory(get_session)
    user_repository = providers.Factory(SQLAlchemyUserRepository, session=db_session)
    create_user_use_case = providers.Factory(
        CreateUserUseCase, user_repository=user_repository
    )

# adapters/controllers/user_controller.py
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, EmailStr
from use_cases.create_user import CreateUserUseCase, CreateUserRequest
from infrastructure.container import Container
from dependency_injector.wiring import inject, Provide

router = APIRouter(prefix="/users", tags=["users"])

class CreateUserInput(BaseModel):
    email: EmailStr
    name: str

@router.post("/", status_code=status.HTTP_201_CREATED)
@inject
async def create_user(
    data: CreateUserInput,
    use_case: CreateUserUseCase = Depends(Provide[Container.create_user_use_case])
):
    request = CreateUserRequest(email=data.email, name=data.name)
    response = await use_case.execute(request)
    if not response.success:
        raise HTTPException(status_code=400, detail=response.error_message)
    return {"id": str(response.user_id)}

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from adapters.controllers import user_controller
from infrastructure.container import Container
from infrastructure.database import init_db

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield

def create_app() -> FastAPI:
    container = Container()
    container.wire(modules=[user_controller])
    app = FastAPI(title="Clean Architecture API", lifespan=lifespan)
    app.container = container
    app.include_router(user_controller.router)
    return app

app = create_app()

# tests/unit/test_create_user_use_case.py
import pytest
from unittest.mock import AsyncMock
from use_cases.create_user import CreateUserUseCase, CreateUserRequest
from domain.entities.user import User
from domain.value_objects.email import Email

@pytest.fixture
def mock_repository():
    return AsyncMock()

@pytest.fixture
def use_case(mock_repository):
    return CreateUserUseCase(user_repository=mock_repository)

@pytest.mark.asyncio
async def test_create_user_success(use_case, mock_repository):
    mock_repository.find_by_email.return_value = None
    mock_repository.save.return_value = User(
        email=Email("test@example.com"), name="Test User"
    )
    request = CreateUserRequest(email="test@example.com", name="Test User")
    response = await use_case.execute(request)
    assert response.success is True
    assert response.user_id is not None

@pytest.mark.asyncio
async def test_create_user_duplicate_email(use_case, mock_repository):
    mock_repository.find_by_email.return_value = AsyncMock()
    request = CreateUserRequest(email="test@example.com", name="Test User")
    response = await use_case.execute(request)
    assert response.success is False
    assert "already registered" in response.error_message