Clean Architecture Example - Python User Service

gistfile1.txt

# ==========================================
# ENTITIES LAYER - Enterprise Business Rules
# ==========================================

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List
import re

@dataclass
class User:
    """Core business entity representing a user"""
    id: Optional[int]
    email: str
    name: str
    
    def __post_init__(self):
        """Validate business rules"""
        if not self.is_valid_email(self.email):
            raise ValueError("Invalid email format")
        if len(self.name.strip()) < 2:
            raise ValueError("Name must be at least 2 characters")
    
    def is_valid_email(self, email: str) -> bool:
        """Enterprise business rule for email validation"""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return re.match(pattern, email) is not None
    
    def get_display_name(self) -> str:
        """Business rule for how names are displayed"""
        return self.name.title()


# ==========================================
# USE CASES LAYER - Application Business Rules
# ==========================================

class UserRepository(ABC):
    """Abstract repository interface (port)"""
    
    @abstractmethod
    def save(self, user: User) -> User:
        pass
    
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[User]:
        pass
    
    @abstractmethod
    def find_by_email(self, email: str) -> Optional[User]:
        pass
    
    @abstractmethod
    def find_all(self) -> List[User]:
        pass


class UserService:
    """Service class containing all user-related business logic"""
    
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    def create_user(self, email: str, name: str) -> User:
        """Create a new user with business rule validation"""
        # Check if user already exists
        existing_user = self.user_repository.find_by_email(email)
        if existing_user:
            raise ValueError(f"User with email {email} already exists")
        
        # Create new user (entity validation happens automatically)
        user = User(id=None, email=email, name=name)
        
        # Save and return
        return self.user_repository.save(user)
    
    def get_user(self, user_id: int) -> Optional[User]:
        """Retrieve a user by ID"""
        return self.user_repository.find_by_id(user_id)
    
    def list_users(self) -> List[User]:
        """Get all users"""
        return self.user_repository.find_all()


# ==========================================
# INTERFACE ADAPTERS LAYER - Controllers, Presenters, Gateways
# ==========================================

class InMemoryUserRepository(UserRepository):
    """Concrete implementation of UserRepository (adapter)"""
    
    def __init__(self):
        self._users: List[User] = []
        self._next_id = 1
    
    def save(self, user: User) -> User:
        if user.id is None:
            # Create new user
            user.id = self._next_id
            self._next_id += 1
            self._users.append(user)
        else:
            # Update existing user
            for i, existing_user in enumerate(self._users):
                if existing_user.id == user.id:
                    self._users[i] = user
                    break
        return user
    
    def find_by_id(self, user_id: int) -> Optional[User]:
        for user in self._users:
            if user.id == user_id:
                return user
        return None
    
    def find_by_email(self, email: str) -> Optional[User]:
        for user in self._users:
            if user.email == email:
                return user
        return None
    
    def find_all(self) -> List[User]:
        return self._users.copy()


class UserController:
    """HTTP controller (adapter) that handles web requests"""
    
    def __init__(self, user_service: UserService):
        self.user_service = user_service
    
    def create_user(self, email: str, name: str) -> dict:
        """Handle HTTP POST /users"""
        try:
            user = self.user_service.create_user(email, name)
            return {
                "success": True,
                "data": {
                    "id": user.id,
                    "email": user.email,
                    "name": user.get_display_name()
                }
            }
        except ValueError as e:
            return {"success": False, "error": str(e)}
    
    def get_user(self, user_id: int) -> dict:
        """Handle HTTP GET /users/{id}"""
        user = self.user_service.get_user(user_id)
        if user:
            return {
                "success": True,
                "data": {
                    "id": user.id,
                    "email": user.email,
                    "name": user.get_display_name()
                }
            }
        return {"success": False, "error": "User not found"}
    
    def list_users(self) -> dict:
        """Handle HTTP GET /users"""
        users = self.user_service.list_users()
        return {
            "success": True,
            "data": [
                {
                    "id": user.id,
                    "email": user.email,
                    "name": user.get_display_name()
                }
                for user in users
            ]
        }


# ==========================================
# FRAMEWORKS & DRIVERS LAYER - Main/Composition Root
# ==========================================

class Application:
    """Main application class that wires everything together"""
    
    def __init__(self):
        # Infrastructure layer
        self.user_repository = InMemoryUserRepository()
        
        # Use cases layer
        self.user_service = UserService(self.user_repository)
        
        # Controllers layer
        self.user_controller = UserController(self.user_service)


# ==========================================
# EXAMPLE USAGE
# ==========================================

def main():
    """Example of how the application would be used"""
    app = Application()
    
    print("=== Clean Architecture Demo ===\n")
    
    # Create users
    print("Creating users...")
    result1 = app.user_controller.create_user("[email protected]", "john doe")
    print(f"Create John: {result1}")
    
    result2 = app.user_controller.create_user("[email protected]", "jane smith")
    print(f"Create Jane: {result2}")
    
    # Try to create duplicate user
    result3 = app.user_controller.create_user("[email protected]", "john duplicate")
    print(f"Create duplicate John: {result3}")
    
    # Try invalid email
    result4 = app.user_controller.create_user("invalid-email", "invalid user")
    print(f"Create invalid user: {result4}")
    
    print("\n" + "="*50 + "\n")
    
    # List all users
    print("Listing all users...")
    all_users = app.user_controller.list_users()
    print(f"All users: {all_users}")
    
    print("\n" + "="*50 + "\n")
    
    # Get specific user
    print("Getting specific user...")
    user_1 = app.user_controller.get_user(1)
    print(f"User 1: {user_1}")
    
    # Get non-existent user
    user_999 = app.user_controller.get_user(999)
    print(f"User 999: {user_999}")


if __name__ == "__main__":
    main()