Clean Architecture

Clean Architecture là gì?

Clean Architecture là một mô hình kiến trúc phần mềm do Robert C. Martin (Uncle Bob) đề xuất. Mục tiêu là tạo ra hệ thống:

Độc lập với framework: Không bị phụ thuộc vào framework cụ thể
Dễ kiểm thử: Business logic có thể test mà không cần UI, Database, Web Server
Độc lập với UI: UI có thể thay đổi mà không ảnh hưởng business logic
Độc lập với Database: Có thể đổi database dễ dàng
Độc lập với bất kỳ external agency nào: Business logic không biết gì về thế giới bên ngoài

Kiến trúc tầng (Layered Architecture)

Clean Architecture chia hệ thống thành các tầng đồng tâm:


┌────────────────────────────────────────┐
│   Frameworks & Drivers                 │  ← Tầng ngoài cùng
│   (Web, DB, UI, External Services)    │
├────────────────────────────────────────┤
│   Interface Adapters                   │  ← Controllers, Presenters, Gateways
│   (Controllers, Presenters, Gateways) │
├────────────────────────────────────────┤
│   Use Cases (Application Logic)       │  ← Business rules cụ thể cho app
├────────────────────────────────────────┤
│   Entities (Domain/Business Logic)    │  ← Tầng trong cùng - Business rules
└────────────────────────────────────────┘

Dependency Rule (Quy tắc phụ thuộc)

Dependencies chỉ được trỏ vào trong, không bao giờ trỏ ra ngoài.

Tầng trong không biết gì về tầng ngoài
Business logic không phụ thuộc vào UI, Database, Framework
Code ở tầng trong không được import code từ tầng ngoài

Các tầng trong Clean Architecture

1. Entities (Domain Layer) - Tầng trong cùng

Entities chứa business logic thuần túy, không phụ thuộc vào bất cứ thứ gì.


from dataclasses import dataclass
from typing import List
from datetime import datetime
 
@dataclass
class OrderItem:
    product_id: str
    product_name: str
    price: float
    quantity: int
 
    def get_total(self) -> float:
        return self.price * self.quantity
 
class Order:
    """
    Entity Order - Business logic thuần túy.
    Không biết gì về database, web framework, etc.
    """
    def __init__(self, customer_id: str, items: List[OrderItem]):
        self.id = None
        self.customer_id = customer_id
        self.items = items
        self.created_at = datetime.now()
        self.status = "pending"
 
    def calculate_total(self) -> float:
        """Tính tổng giá trị đơn hàng."""
        return sum(item.get_total() for item in self.items)
 
    def apply_discount(self, discount_percent: float) -> None:
        """Áp dụng discount cho đơn hàng."""
        if not 0 <= discount_percent <= 100:
            raise ValueError("Discount must be between 0 and 100")
 
        total = self.calculate_total()
        discount_amount = total * (discount_percent / 100)
        self.discount = discount_amount
 
    def can_be_cancelled(self) -> bool:
        """Kiểm tra đơn hàng có thể hủy không."""
        return self.status in ["pending", "confirmed"]
 
    def cancel(self) -> None:
        """Hủy đơn hàng."""
        if not self.can_be_cancelled():
            raise ValueError(f"Cannot cancel order with status {self.status}")
        self.status = "cancelled"

Đặc điểm:

Không import gì từ tầng ngoài
Business rules thuần túy
Có thể test mà không cần database, framework

2. Use Cases (Application Layer)

Use Cases chứa business logic cụ thể cho application. Điều phối các Entities.


from abc import ABC, abstractmethod
from typing import Optional
 
# Interfaces (Ports) - Abstractions cho tầng ngoài
class OrderRepository(ABC):
    """Interface cho việc lưu trữ Order."""
    @abstractmethod
    def save(self, order: Order) -> Order:
        pass
 
    @abstractmethod
    def find_by_id(self, order_id: str) -> Optional[Order]:
        pass
 
class PaymentGateway(ABC):
    """Interface cho payment processing."""
    @abstractmethod
    def process_payment(self, order: Order) -> bool:
        pass
 
class EmailService(ABC):
    """Interface cho email service."""
    @abstractmethod
    def send_order_confirmation(self, order: Order) -> None:
        pass
 
# Use Case
class CreateOrderUseCase:
    """
    Use Case: Tạo đơn hàng mới.
    Điều phối các entities và gọi các services.
    """
    def __init__(
        self,
        order_repository: OrderRepository,
        payment_gateway: PaymentGateway,
        email_service: EmailService
    ):
        self.order_repository = order_repository
        self.payment_gateway = payment_gateway
        self.email_service = email_service
 
    def execute(self, customer_id: str, items: List[OrderItem], discount: float = 0) -> Order:
        """
        Thực thi use case tạo đơn hàng.
 
        Args:
            customer_id: ID của khách hàng
            items: Danh sách items trong đơn hàng
            discount: Discount percent (0-100)
 
        Returns:
            Order đã được tạo
 
        Raises:
            ValueError: Nếu items rỗng hoặc payment failed
        """
        # 1. Validate
        if not items:
            raise ValueError("Order must have at least one item")
 
        # 2. Tạo Order entity
        order = Order(customer_id, items)
 
        # 3. Áp dụng discount
        if discount > 0:
            order.apply_discount(discount)
 
        # 4. Lưu vào database (thông qua repository)
        order = self.order_repository.save(order)
 
        # 5. Process payment
        payment_success = self.payment_gateway.process_payment(order)
        if not payment_success:
            order.status = "payment_failed"
            self.order_repository.save(order)
            raise ValueError("Payment processing failed")
 
        # 6. Update status
        order.status = "confirmed"
        order = self.order_repository.save(order)
 
        # 7. Send confirmation email
        self.email_service.send_order_confirmation(order)
 
        return order
 
class CancelOrderUseCase:
    """Use Case: Hủy đơn hàng."""
    def __init__(
        self,
        order_repository: OrderRepository,
        email_service: EmailService
    ):
        self.order_repository = order_repository
        self.email_service = email_service
 
    def execute(self, order_id: str) -> Order:
        """Hủy đơn hàng."""
        # 1. Tìm order
        order = self.order_repository.find_by_id(order_id)
        if not order:
            raise ValueError(f"Order {order_id} not found")
 
        # 2. Cancel order (business logic trong entity)
        order.cancel()
 
        # 3. Lưu
        order = self.order_repository.save(order)
 
        # 4. Send email
        self.email_service.send_order_confirmation(order)
 
        return order

Đặc điểm:

Điều phối các Entities
Định nghĩa interfaces (ports) cho các services bên ngoài
Không biết implementation cụ thể của database, email service, etc.

3. Interface Adapters (Adapters Layer)

Interface Adapters chuyển đổi data giữa Use Cases và External Services.


from typing import Dict, Any
 
# DTO (Data Transfer Object)
class CreateOrderRequest:
    """DTO cho request tạo order."""
    def __init__(self, customer_id: str, items: List[Dict[str, Any]], discount: float = 0):
        self.customer_id = customer_id
        self.items = items
        self.discount = discount
 
    def to_domain_items(self) -> List[OrderItem]:
        """Chuyển đổi từ DTO sang Domain OrderItem."""
        return [
            OrderItem(
                product_id=item['product_id'],
                product_name=item['product_name'],
                price=item['price'],
                quantity=item['quantity']
            )
            for item in self.items
        ]
 
class OrderResponse:
    """DTO cho response."""
    def __init__(self, order: Order):
        self.id = order.id
        self.customer_id = order.customer_id
        self.total = order.calculate_total()
        self.status = order.status
        self.created_at = order.created_at.isoformat()
 
    def to_dict(self) -> Dict[str, Any]:
        """Chuyển sang dictionary để trả về JSON."""
        return {
            'id': self.id,
            'customer_id': self.customer_id,
            'total': self.total,
            'status': self.status,
            'created_at': self.created_at
        }
 
# Controller
class OrderController:
    """
    Controller - Xử lý HTTP requests.
    Adapter giữa Web Framework và Use Cases.
    """
    def __init__(self, create_order_use_case: CreateOrderUseCase):
        self.create_order_use_case = create_order_use_case
 
    def create_order(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Handle POST /orders request.
 
        Args:
            request_data: Data từ HTTP request body
 
        Returns:
            Response dictionary
        """
        try:
            # 1. Parse request
            request = CreateOrderRequest(
                customer_id=request_data['customer_id'],
                items=request_data['items'],
                discount=request_data.get('discount', 0)
            )
 
            # 2. Convert to domain objects
            items = request.to_domain_items()
 
            # 3. Execute use case
            order = self.create_order_use_case.execute(
                request.customer_id,
                items,
                request.discount
            )
 
            # 4. Convert to response
            response = OrderResponse(order)
 
            return {
                'success': True,
                'data': response.to_dict()
            }
 
        except ValueError as e:
            return {
                'success': False,
                'error': str(e)
            }

4. Frameworks & Drivers (Infrastructure Layer) - Tầng ngoài cùng

Frameworks & Drivers chứa implementations cụ thể.


import sqlite3
from typing import Optional
 
# Database Implementation
class SQLiteOrderRepository(OrderRepository):
    """
    Implementation cụ thể của OrderRepository sử dụng SQLite.
    """
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_db()
 
    def _init_db(self):
        """Khởi tạo database schema."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS orders (
                    id TEXT PRIMARY KEY,
                    customer_id TEXT,
                    total REAL,
                    status TEXT,
                    created_at TEXT
                )
            """)
 
    def save(self, order: Order) -> Order:
        """Lưu order vào SQLite."""
        with sqlite3.connect(self.db_path) as conn:
            if order.id is None:
                # Generate ID
                import uuid
                order.id = str(uuid.uuid4())
 
            conn.execute(
                """
                INSERT OR REPLACE INTO orders (id, customer_id, total, status, created_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    order.id,
                    order.customer_id,
                    order.calculate_total(),
                    order.status,
                    order.created_at.isoformat()
                )
            )
            conn.commit()
 
        return order
 
    def find_by_id(self, order_id: str) -> Optional[Order]:
        """Tìm order theo ID."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT customer_id FROM orders WHERE id = ?",
                (order_id,)
            )
            row = cursor.fetchone()
 
            if row:
                # Reconstruct Order entity
                # (Simplified - thực tế cần load items)
                order = Order(row[0], [])
                order.id = order_id
                return order
 
        return None
 
# Payment Gateway Implementation
class StripePaymentGateway(PaymentGateway):
    """Implementation sử dụng Stripe."""
    def __init__(self, api_key: str):
        self.api_key = api_key
 
    def process_payment(self, order: Order) -> bool:
        """Process payment qua Stripe."""
        # Call Stripe API
        print(f"Processing payment via Stripe for order {order.id}")
        # stripe.Charge.create(...)
        return True
 
# Email Service Implementation
class SMTPEmailService(EmailService):
    """Implementation sử dụng SMTP."""
    def __init__(self, smtp_host: str, smtp_port: int):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
 
    def send_order_confirmation(self, order: Order) -> None:
        """Gửi email xác nhận đơn hàng."""
        print(f"Sending email confirmation for order {order.id}")
        # smtplib.SMTP(...).send_message(...)

Dependency Injection & Wiring

Kết nối tất cả các tầng lại với nhau:


# main.py hoặc dependency injection container
 
def create_order_use_case() -> CreateOrderUseCase:
    """Factory function tạo use case với dependencies."""
    # Infrastructure layer
    order_repository = SQLiteOrderRepository("orders.db")
    payment_gateway = StripePaymentGateway("sk_test_...")
    email_service = SMTPEmailService("smtp.gmail.com", 587)
 
    # Use case layer
    use_case = CreateOrderUseCase(
        order_repository,
        payment_gateway,
        email_service
    )
 
    return use_case
 
def create_order_controller() -> OrderController:
    """Factory function tạo controller."""
    use_case = create_order_use_case()
    controller = OrderController(use_case)
    return controller
 
# Web Framework (FastAPI, Flask, etc.)
from flask import Flask, request, jsonify
 
app = Flask(__name__)
order_controller = create_order_controller()
 
@app.route('/orders', methods=['POST'])
def create_order():
    data = request.get_json()
    response = order_controller.create_order(data)
    return jsonify(response)
 
if __name__ == '__main__':
    app.run()

Cấu trúc thư mục


project/
├── domain/                    # Entities layer
│   ├── __init__.py
│   ├── entities.py           # Order, OrderItem entities
│   └── exceptions.py
│
├── application/               # Use Cases layer
│   ├── __init__.py
│   ├── interfaces.py         # OrderRepository, PaymentGateway interfaces
│   ├── use_cases/
│   │   ├── create_order.py
│   │   └── cancel_order.py
│   └── dtos.py               # Data Transfer Objects
│
├── adapters/                  # Interface Adapters layer
│   ├── __init__.py
│   ├── controllers/
│   │   └── order_controller.py
│   └── presenters/
│       └── order_presenter.py
│
├── infrastructure/            # Frameworks & Drivers layer
│   ├── __init__.py
│   ├── database/
│   │   └── sqlite_order_repository.py
│   ├── payment/
│   │   └── stripe_payment_gateway.py
│   └── email/
│       └── smtp_email_service.py
│
└── main.py                    # Application entry point, DI setup

Lợi ích của Clean Architecture

1. Dễ kiểm thử


# Test Use Case mà không cần database thật
from unittest.mock import Mock
 
def test_create_order_use_case():
    # Arrange - Mock các dependencies
    mock_repository = Mock(spec=OrderRepository)
    mock_payment = Mock(spec=PaymentGateway)
    mock_email = Mock(spec=EmailService)
 
    mock_payment.process_payment.return_value = True
 
    use_case = CreateOrderUseCase(mock_repository, mock_payment, mock_email)
 
    items = [OrderItem("P1", "Product 1", 100, 2)]
 
    # Act
    order = use_case.execute("C123", items)
 
    # Assert
    assert order.status == "confirmed"
    mock_repository.save.assert_called()
    mock_payment.process_payment.assert_called_once()
    mock_email.send_order_confirmation.assert_called_once()

2. Dễ thay đổi implementation


# Đổi từ SQLite sang PostgreSQL
postgres_repository = PostgreSQLOrderRepository("postgresql://...")
 
# Đổi từ Stripe sang PayPal
paypal_gateway = PayPalPaymentGateway(client_id="...")
 
# Đổi từ SMTP sang SendGrid
sendgrid_service = SendGridEmailService(api_key="...")
 
# Use case không cần thay đổi!
use_case = CreateOrderUseCase(
    postgres_repository,
    paypal_gateway,
    sendgrid_service
)

3. Độc lập với framework


# Dễ dàng đổi từ Flask sang FastAPI
from fastapi import FastAPI
 
app = FastAPI()
 
@app.post("/orders")
async def create_order(data: dict):
    response = order_controller.create_order(data)
    return response
 
# Business logic (Use Cases, Entities) không đổi!

So sánh với kiến trúc khác

Kiến trúc	Ưu điểm	Nhược điểm
Monolithic	Đơn giản, dễ bắt đầu	Khó scale, tight coupling
Layered (3-tier)	Rõ ràng, phổ biến	Database-centric, khó test
Clean Architecture	Loose coupling, testable, flexible	Phức tạp, nhiều boilerplate
Hexagonal	Tương tự Clean Architecture	Phức tạp
Microservices	Scale tốt, độc lập	Rất phức tạp, distributed system

Khi nào nên dùng Clean Architecture?

✅ Nên dùng khi:

Dự án lớn, phức tạp
Nhiều business rules
Dự án dài hạn, cần bảo trì lâu
Team lớn
Cần thay đổi framework/database trong tương lai
Yêu cầu test coverage cao

⚠️ Có thể overkill cho:

Prototype, POC
CRUD đơn giản
Script nhỏ
Dự án ngắn hạn
Team nhỏ, một người

Best Practices

Dependency Rule: Luôn tuân thủ quy tắc dependencies chỉ trỏ vào trong
Interface Segregation: Tách interfaces nhỏ, cụ thể
Dependency Injection: Inject dependencies qua constructor
DTOs: Sử dụng DTOs để chuyển đổi data giữa các layers
Testing: Viết tests cho Use Cases và Entities
Documentation: Document rõ business rules

Tổng kết

Clean Architecture giúp:

✅ Tách biệt concerns: Business logic độc lập với technical details
✅ Testable: Dễ test từng layer
✅ Flexible: Dễ thay đổi implementation
✅ Maintainable: Dễ bảo trì và mở rộng
✅ Framework independent: Không phụ thuộc framework

Trade-offs:

❌ Phức tạp hơn
❌ Nhiều boilerplate code
❌ Learning curve cao
❌ Có thể overkill cho project nhỏ

Clean Architecture là một công cụ mạnh mẽ cho các dự án lớn. Hãy cân nhắc kỹ trước khi áp dụng!