Split View: 소프트웨어 아키텍처 & 디자인 패턴 완전 정복: SOLID, Clean Architecture, AI 시스템 설계까지

# BAD: multiple responsibilities in one class
class UserManager:
    def create_user(self, data): ...
    def send_welcome_email(self, user): ...
    def save_to_database(self, user): ...

# GOOD: each responsibility in its own class
class UserRepository:
    def save(self, user): ...

class EmailService:
    def send_welcome(self, user): ...

class UserFactory:
    def create(self, data): ...

from abc import ABC, abstractmethod

class Discount(ABC):
    @abstractmethod
    def apply(self, price: float) -> float: ...

class NoDiscount(Discount):
    def apply(self, price: float) -> float:
        return price

class PercentDiscount(Discount):
    def __init__(self, percent: float):
        self.percent = percent
    def apply(self, price: float) -> float:
        return price * (1 - self.percent / 100)

class VIPDiscount(Discount):
    def apply(self, price: float) -> float:
        return price * 0.7

# New discount policies can be added without modifying existing code
class Order:
    def __init__(self, discount: Discount):
        self.discount = discount

    def final_price(self, base: float) -> float:
        return self.discount.apply(base)

class Bird:
    def fly(self) -> str:
        return "Flying"

# LSP violation: Penguin inherits Bird but cannot fly
class Penguin(Bird):
    def fly(self):
        raise NotImplementedError("Penguins cannot fly")

# GOOD: separate interfaces based on capabilities
class FlyingBird(ABC):
    @abstractmethod
    def fly(self) -> str: ...

class SwimmingBird(ABC):
    @abstractmethod
    def swim(self) -> str: ...

class Eagle(FlyingBird):
    def fly(self) -> str:
        return "Eagle soars through the sky"

class Penguin(SwimmingBird):
    def swim(self) -> str:
        return "Penguin swims gracefully"

# BAD: one bloated interface
class Machine(ABC):
    @abstractmethod
    def print(self): ...
    @abstractmethod
    def scan(self): ...
    @abstractmethod
    def fax(self): ...

# GOOD: split into smaller, focused interfaces
class Printable(ABC):
    @abstractmethod
    def print(self): ...

class Scannable(ABC):
    @abstractmethod
    def scan(self): ...

class MultiFunctionPrinter(Printable, Scannable):
    def print(self): print("Printing...")
    def scan(self): print("Scanning...")

class SimplePrinter(Printable):
    def print(self): print("Simple print...")

# BAD: high-level depends directly on low-level
class MySQLDatabase:
    def query(self, sql: str): ...

class UserService:
    def __init__(self):
        self.db = MySQLDatabase()  # depends on a concrete class

# GOOD: depend on abstractions (dependency injection)
class DatabasePort(ABC):
    @abstractmethod
    def find_user(self, user_id: str) -> dict: ...

class MySQLAdapter(DatabasePort):
    def find_user(self, user_id: str) -> dict:
        # MySQL implementation
        return {}

class MongoAdapter(DatabasePort):
    def find_user(self, user_id: str) -> dict:
        # MongoDB implementation
        return {}

class UserService:
    def __init__(self, db: DatabasePort):
        self.db = db  # depends on the abstraction

# Usage
service = UserService(db=MySQLAdapter())

class LLMProvider(ABC):
    @abstractmethod
    def complete(self, prompt: str) -> str: ...

class OpenAIProvider(LLMProvider):
    def complete(self, prompt: str) -> str:
        return f"OpenAI response: {prompt}"

class AnthropicProvider(LLMProvider):
    def complete(self, prompt: str) -> str:
        return f"Anthropic response: {prompt}"

class LLMFactory:
    _registry = {
        "openai": OpenAIProvider,
        "anthropic": AnthropicProvider,
    }

    @classmethod
    def create(cls, provider: str) -> LLMProvider:
        klass = cls._registry.get(provider)
        if not klass:
            raise ValueError(f"Unknown provider: {provider}")
        return klass()

import threading

class ConfigManager:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._config = {}
        return cls._instance

    def set(self, key: str, value):
        self._config[key] = value

    def get(self, key: str):
        return self._config.get(key)

class EventBus:
    def __init__(self):
        self._subscribers: dict[str, list] = {}

    def subscribe(self, event: str, handler):
        self._subscribers.setdefault(event, []).append(handler)

    def publish(self, event: str, data=None):
        for handler in self._subscribers.get(event, []):
            handler(data)

# Usage
bus = EventBus()
bus.subscribe("user.created", lambda d: print(f"Send welcome email: {d}"))
bus.subscribe("user.created", lambda d: print(f"Log analytics event: {d}"))
bus.publish("user.created", {"id": "u1", "email": "user@example.com"})

class SortStrategy(ABC):
    @abstractmethod
    def sort(self, data: list) -> list: ...

class QuickSort(SortStrategy):
    def sort(self, data: list) -> list:
        return sorted(data)

class MergeSort(SortStrategy):
    def sort(self, data: list) -> list:
        return sorted(data, key=lambda x: x)

class DataProcessor:
    def __init__(self, strategy: SortStrategy):
        self._strategy = strategy

    def set_strategy(self, strategy: SortStrategy):
        self._strategy = strategy

    def process(self, data: list) -> list:
        return self._strategy.sort(data)

import time
import functools

def retry(max_attempts: int = 3):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(2 ** attempt)
        return wrapper
    return decorator

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__} took {time.time() - start:.2f}s")
        return result
    return wrapper

@retry(max_attempts=3)
@timed
def call_llm_api(prompt: str) -> str:
    return "response"

class Command(ABC):
    @abstractmethod
    def execute(self): ...
    @abstractmethod
    def undo(self): ...

class CreatePostCommand(Command):
    def __init__(self, repo, post_data: dict):
        self.repo = repo
        self.post_data = post_data
        self.created_id = None

    def execute(self):
        self.created_id = self.repo.create(self.post_data)

    def undo(self):
        if self.created_id:
            self.repo.delete(self.created_id)

class CommandHistory:
    def __init__(self):
        self._history: list[Command] = []

    def execute(self, cmd: Command):
        cmd.execute()
        self._history.append(cmd)

    def undo_last(self):
        if self._history:
            self._history.pop().undo()

Outer Layer → Interface Adapters → Use Cases → Domain Entities

# Domain Entity (innermost layer)
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Article:
    id: str
    title: str
    content: str
    author_id: str
    created_at: datetime = field(default_factory=datetime.utcnow)

    def publish(self):
        if not self.title or not self.content:
            raise ValueError("Title and content are required")

# Use Case (orchestrates domain logic)
class CreateArticleUseCase:
    def __init__(self, repo: "ArticleRepository", event_bus: EventBus):
        self.repo = repo
        self.event_bus = event_bus

    def execute(self, title: str, content: str, author_id: str) -> Article:
        import uuid
        article = Article(
            id=str(uuid.uuid4()),
            title=title,
            content=content,
            author_id=author_id,
        )
        article.publish()
        self.repo.save(article)
        self.event_bus.publish("article.created", {"id": article.id})
        return article

# Interface Adapter (outermost layer)
class ArticleController:
    def __init__(self, use_case: CreateArticleUseCase):
        self.use_case = use_case

    def handle_create(self, request: dict) -> dict:
        article = self.use_case.execute(
            title=request["title"],
            content=request["content"],
            author_id=request["author_id"],
        )
        return {"id": article.id, "title": article.title}

# Port (interface definition)
class ArticleRepository(ABC):
    @abstractmethod
    def save(self, article: Article): ...
    @abstractmethod
    def find_by_id(self, id: str) -> Article: ...

class NotificationPort(ABC):
    @abstractmethod
    def notify(self, message: str): ...

# Adapter (connects to external systems)
class SQLiteArticleRepository(ArticleRepository):
    def save(self, article: Article):
        pass  # SQLite save logic
    def find_by_id(self, id: str) -> Article:
        pass  # SQLite query logic

class SlackNotificationAdapter(NotificationPort):
    def notify(self, message: str):
        pass  # Slack API call

# Command Side
@dataclass
class CreateOrderCommand:
    order_id: str
    user_id: str
    items: list[dict]

# Query Side (separate read model)
@dataclass
class OrderSummaryView:
    order_id: str
    total_price: float
    item_count: int

# Event Sourcing
@dataclass
class OrderCreatedEvent:
    order_id: str
    user_id: str
    items: list[dict]
    timestamp: datetime = field(default_factory=datetime.utcnow)

class OrderAggregate:
    def __init__(self):
        self.events: list = []
        self.state = {}

    def create(self, cmd: CreateOrderCommand):
        event = OrderCreatedEvent(
            order_id=cmd.order_id,
            user_id=cmd.user_id,
            items=cmd.items,
        )
        self._apply(event)
        self.events.append(event)

    def _apply(self, event: OrderCreatedEvent):
        self.state["id"] = event.order_id
        self.state["items"] = event.items
        self.state["total"] = sum(
            i.get("price", 0) * i.get("qty", 1) for i in event.items
        )

from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI(title="API Gateway")

SERVICE_MAP = {
    "users": "http://user-service:8001",
    "orders": "http://order-service:8002",
    "products": "http://product-service:8003",
}

@app.get("/api/users/{user_id}")
async def proxy_user(user_id: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{SERVICE_MAP['users']}/users/{user_id}")
        if resp.status_code == 404:
            raise HTTPException(status_code=404, detail="User not found")
        return resp.json()

class OrderSaga:
    def __init__(self, order_service, payment_service, inventory_service):
        self.order_svc = order_service
        self.payment_svc = payment_service
        self.inventory_svc = inventory_service

    async def execute(self, order_data: dict):
        order_id = None
        payment_id = None
        try:
            # Step 1: Create order
            order_id = await self.order_svc.create(order_data)
            # Step 2: Process payment
            payment_id = await self.payment_svc.charge(order_data["amount"])
            # Step 3: Reserve inventory
            await self.inventory_svc.reserve(order_data["items"])
            return {"status": "success", "order_id": order_id}
        except Exception as e:
            # Compensating transactions (reverse rollback)
            if payment_id:
                await self.payment_svc.refund(payment_id)
            if order_id:
                await self.order_svc.cancel(order_id)
            raise

# BAD
def calc(d, r):
    return d * (1 - r / 100)

# GOOD
def calculate_discounted_price(original_price: float, discount_rate_percent: float) -> float:
    return original_price * (1 - discount_rate_percent / 100)

# BAD: too many responsibilities
def process_user_registration(email, password, name, send_email=True):
    if "@" not in email:
        raise ValueError("Invalid email format")
    hashed = hash(password)
    user = {"email": email, "password": hashed, "name": name}
    db.save(user)
    if send_email:
        mailer.send(email, "Welcome!")
    return user

# GOOD: separated functions
def validate_email(email: str) -> None:
    if "@" not in email:
        raise ValueError("Invalid email format")

def hash_password(raw: str) -> str:
    import hashlib
    return hashlib.sha256(raw.encode()).hexdigest()

def register_user(email: str, password: str, name: str) -> dict:
    validate_email(email)
    return {"email": email, "password": hash_password(password), "name": name}

from dataclasses import dataclass

@dataclass
class RAGConfig:
    embedding_model: str = "text-embedding-3-small"
    llm_model: str = "gpt-4o"
    top_k: int = 5
    chunk_size: int = 512

class RAGPipeline:
    def __init__(self, config: RAGConfig, vector_store, llm_client):
        self.config = config
        self.vector_store = vector_store
        self.llm = llm_client

    def retrieve(self, query: str) -> list[str]:
        # 1. Embed the query
        query_vector = self.llm.embed(query)
        # 2. Search for similar documents
        docs = self.vector_store.search(query_vector, top_k=self.config.top_k)
        return [d["content"] for d in docs]

    def generate(self, query: str, context: list[str]) -> str:
        context_text = "\n\n".join(context)
        prompt = f"""Answer the question using the following context.

Context:
{context_text}

Question: {query}
Answer:"""
        return self.llm.complete(prompt)

    def query(self, user_question: str) -> str:
        context = self.retrieve(user_question)
        return self.generate(user_question, context)

@dataclass
class Tool:
    name: str
    description: str
    func: callable

class ReActAgent:
    """AI Agent using the Reasoning + Acting pattern"""

    def __init__(self, llm, tools: list[Tool]):
        self.llm = llm
        self.tools = {t.name: t for t in tools}

    def _build_system_prompt(self) -> str:
        tool_desc = "\n".join(
            f"- {t.name}: {t.description}" for t in self.tools.values()
        )
        return f"""You are an AI Agent that solves tasks using tools.
Available tools:
{tool_desc}

Format:
Thought: [analyze the current situation]
Action: [tool name to use]
Action Input: [input for the tool]
Observation: [result of the tool execution]
... (repeat)
Final Answer: [final answer]"""

    def run(self, task: str, max_steps: int = 10) -> str:
        messages = [{"role": "user", "content": task}]
        for _ in range(max_steps):
            response = self.llm.chat(messages)
            if "Final Answer:" in response:
                return response.split("Final Answer:")[-1].strip()
            if "Action:" in response:
                action_line = [l for l in response.split("\n") if l.startswith("Action:")]
                if action_line:
                    tool_name = action_line[0].replace("Action:", "").strip()
                    tool = self.tools.get(tool_name)
                    if tool:
                        observation = tool.func(response)
                        messages.append({"role": "assistant", "content": response})
                        messages.append({"role": "user", "content": f"Observation: {observation}"})
        return "Max steps exceeded"

       [E2E Tests]           <- slow and costly, keep few
     [Integration Tests]     <- verify service contracts
  [Unit Tests]               <- fast and cheap, keep many

import pytest

# 1. RED: write a failing test first
def test_calculate_discounted_price_basic():
    assert calculate_discounted_price(100.0, 20.0) == 80.0

def test_calculate_discounted_price_zero_discount():
    assert calculate_discounted_price(100.0, 0.0) == 100.0

def test_calculate_discounted_price_full_discount():
    assert calculate_discounted_price(100.0, 100.0) == 0.0

def test_calculate_discounted_price_invalid_rate():
    with pytest.raises(ValueError):
        calculate_discounted_price(100.0, -10.0)

# 2. GREEN: minimal implementation to pass the tests
def calculate_discounted_price(price: float, discount_rate: float) -> float:
    if discount_rate < 0 or discount_rate > 100:
        raise ValueError("Discount rate must be between 0 and 100")
    return price * (1 - discount_rate / 100)

# 3. REFACTOR: improve quality while keeping tests green

from unittest.mock import MagicMock, patch

class TestUserService:
    def test_create_user_sends_email(self):
        mock_repo = MagicMock()
        mock_email = MagicMock()
        mock_repo.save.return_value = {"id": "u1"}

        service = UserService(repo=mock_repo, email_svc=mock_email)
        service.register("test@example.com", "John Doe")

        mock_repo.save.assert_called_once()
        mock_email.send_welcome.assert_called_once_with("test@example.com")

    def test_create_user_handles_db_error(self):
        mock_repo = MagicMock()
        mock_repo.save.side_effect = Exception("DB connection failed")
        mock_email = MagicMock()

        service = UserService(repo=mock_repo, email_svc=mock_email)
        with pytest.raises(Exception):
            service.register("test@example.com", "John Doe")

        mock_email.send_welcome.assert_not_called()