สร้างระบบ Python ขนาดใหญ่แบบยั่งยืนด้วย Clean Architecture (พร้อมตัวอย่างและแผนภาพ)

ทำไมต้อง Clean Architecture?

เมื่อซอฟต์แวร์เติบโตขึ้น ความซับซ้อนก็เพิ่มขึ้นตาม
ถ้าโค้ดผูกติดกับเฟรมเวิร์กหรือฐานข้อมูล จะทดสอบหรือขยายระบบยาก
Clean Architecture ช่วยให้โค้ดธุรกิจ (business logic) แยกขาดจากเฟรมเวิร์ก, ฐานข้อมูล, หรือ UI

“โค้ดธุรกิจของคุณควรทดสอบได้ และแก้ไขได้ โดยไม่ต้องยุ่งกับเฟรมเวิร์กหรือฐานข้อมูลเลย”

หลักการสำคัญ

โค้ดธุรกิจของคุณต้องทดสอบได้ โดยไม่ผูกกับเฟรมเวิร์กใด ๆ
ถ้าคุณรัน logic หลัก (entities, use cases) ด้วย Python เพียว ๆ ได้ ถือว่าถูกทาง

การจัดโครงสร้างโค้ดแบบ Clean Architecture

แนวคิดนี้แบ่งโค้ดเป็นชั้น (layer) ชัดเจน เพื่อให้ดูแลและทดสอบง่าย

Entities: คลาสธุรกิจ (เช่น Car, Rental)
Use Cases: logic สำหรับงานแต่ละอย่าง (เช่น "เช่ารถ", "แสดงรถทั้งหมด")
Repositories/Interfaces: ชั้นเชื่อมต่อข้อมูล/บริการ (abstract)
Frameworks/Drivers: Django, Flask, Database, UI

แผนภาพคลาส (UML Class Diagram)

classDiagram
    class Car {
        +int car_id
        +str make
        +str model
        +bool is_available
    }
    class Rental {
        +int rental_id
        +int car_id
        +int user_id
        +str start_date
        +str end_date
    }
    class CarRepository {
        +save(car)
        +get_by_id(car_id)
        +get_all()
        +delete(car_id)
    }
    class RentalRepository {
        +save(rental)
        +get_by_id(rental_id)
        +get_all()
        +delete(rental_id)
    }
    class CarCRUDUseCase {
        -car_repository
        +create_car(car_id, make, model)
        +get_car(car_id)
        +list_cars()
        +update_car(car_id, make, model, is_available)
        +delete_car(car_id)
    }
    class RentCarUseCase {
        -car_repository
        -rental_repository
        +execute(car_id, user_id, start_date, end_date)
    }
    class DjangoCarRepository {
        +save(car)
        +get_by_id(car_id)
        +get_all()
        +delete(car_id)
    }
    class MongoCarRepository {
        +save(car)
        +get_by_id(car_id)
        +get_all()
        +delete(car_id)
    }
    class DjangoRentalRepository
    class MongoRentalRepository

    CarCRUDUseCase --> CarRepository
    RentCarUseCase --> CarRepository
    RentCarUseCase --> RentalRepository
    DjangoCarRepository --|> CarRepository
    MongoCarRepository --|> CarRepository
    DjangoRentalRepository --|> RentalRepository
    MongoRentalRepository --|> RentalRepository

แผนภาพ Flow: ขั้นตอนการทำงานของคลาส

ยกตัวอย่าง “เช่ารถ” – ดูว่า Request เดินทางผ่านแต่ละชั้นอย่างไร

sequenceDiagram
    participant User
    participant UI as UI/View (Django/Flask)
    participant UC as RentCarUseCase
    participant CarRepo as CarRepository
    participant RentalRepo as RentalRepository
    participant DB as DB/Framework

    User->>UI: กรอกฟอร์มเช่ารถ (car_id, user_id, วันที่)
    UI->>UC: execute(car_id, user_id, start_date, end_date)
    UC->>CarRepo: get_by_id(car_id)
    CarRepo->>DB: Query car by id
    CarRepo-->>UC: Car entity
    UC->>CarRepo: save(car) (mark as unavailable)
    CarRepo->>DB: Update car
    UC->>RentalRepo: save(rental)
    RentalRepo->>DB: Insert rental record
    RentalRepo-->>UC: Rental entity
    UC-->>UI: ส่ง Rental กลับ
    UI-->>User: แสดงผลลัพธ์ (สำเร็จ/ไม่สำเร็จ)

ตัวอย่างโค้ดแต่ละชั้น (Layer)

1. Entities (Business Object)

class Car:
    def __init__(self, car_id, make, model, is_available=True):
        self.car_id = car_id
        self.make = make
        self.model = model
        self.is_available = is_available

class Rental:
    def __init__(self, rental_id, car_id, user_id, start_date, end_date):
        self.rental_id = rental_id
        self.car_id = car_id
        self.user_id = user_id
        self.start_date = start_date
        self.end_date = end_date

2. Use Cases

class CarCRUDUseCase:
    def __init__(self, car_repository):
        self.car_repository = car_repository

    def create_car(self, car_id, make, model):
        car = Car(car_id=car_id, make=make, model=model, is_available=True)
        self.car_repository.save(car)
        return car

    def get_car(self, car_id):
        return self.car_repository.get_by_id(car_id)

    def list_cars(self):
        return self.car_repository.get_all()

    def update_car(self, car_id, make, model, is_available):
        car = Car(car_id=car_id, make=make, model=model, is_available=is_available)
        self.car_repository.save(car)
        return car

    def delete_car(self, car_id):
        self.car_repository.delete(car_id)

class RentCarUseCase:
    def __init__(self, car_repository, rental_repository):
        self.car_repository = car_repository
        self.rental_repository = rental_repository

    def execute(self, car_id, user_id, start_date, end_date):
        car = self.car_repository.get_by_id(car_id)
        if not car or not car.is_available:
            raise Exception("Car not available")
        car.is_available = False
        self.car_repository.save(car)
        rental = Rental(
            rental_id=None, car_id=car_id, user_id=user_id,
            start_date=start_date, end_date=end_date
        )
        self.rental_repository.save(rental)
        return rental

3. Repository Interface และ Implementation

class CarRepository:
    def save(self, car): ...
    def get_by_id(self, car_id): ...
    def get_all(self): ...
    def delete(self, car_id): ...

Django ORM ตัวอย่าง:

from entities.car import Car
from myapp.models import CarModel

class DjangoCarRepository(CarRepository):
    def save(self, car):
        obj, _ = CarModel.objects.update_or_create(
            pk=car.car_id,
            defaults={'make': car.make, 'model': car.model, 'is_available': car.is_available}
        )
        return car

    def get_by_id(self, car_id):
        try:
            obj = CarModel.objects.get(pk=car_id)
            return Car(car_id=obj.pk, make=obj.make, model=obj.model, is_available=obj.is_available)
        except CarModel.DoesNotExist:
            return None

    def get_all(self):
        return [
            Car(
                car_id=obj.pk,
                make=obj.make,
                model=obj.model,
                is_available=obj.is_available
            )
            for obj in CarModel.objects.all()
        ]

    def delete(self, car_id):
        CarModel.objects.filter(pk=car_id).delete()

MongoDB ตัวอย่าง:

class MongoCarRepository(CarRepository):
    def __init__(self, db):
        self.collection = db['cars']

    def save(self, car):
        self.collection.update_one(
            {'car_id': car.car_id},
            {'$set': {'make': car.make, 'model': car.model, 'is_available': car.is_available}},
            upsert=True
        )
        return car

    def get_by_id(self, car_id):
        doc = self.collection.find_one({'car_id': car_id})
        if not doc:
            return None
        return Car(
            car_id=doc['car_id'],
            make=doc['make'],
            model=doc['model'],
            is_available=doc.get('is_available', True)
        )

    def get_all(self):
        return [
            Car(
                car_id=doc['car_id'],
                make=doc['make'],
                model=doc['model'],
                is_available=doc.get('is_available', True)
            )
            for doc in self.collection.find()
        ]

    def delete(self, car_id):
        self.collection.delete_one({'car_id': car_id})

4. ตัวอย่าง Django View + Template

from django.shortcuts import render
from use_cases.car_crud import CarCRUDUseCase
from frameworks.django_repositories import DjangoCarRepository

def car_list_view(request):
    use_case = CarCRUDUseCase(DjangoCarRepository())
    cars = use_case.list_cars()
    return render(request, "car_list.html", {"cars": cars})

HTML Template (car_list.html):

<!DOCTYPE html>
<html>
<head>
    <title>Car List</title>
</head>
<body>
    <h1>Available Cars</h1>
    <table border="1">
        <tr>
            <th>ID</th>
            <th>Make</th>
            <th>Model</th>
            <th>Available</th>
        </tr>
        {% for car in cars %}
        <tr>
            <td>{{ car.car_id }}</td>
            <td>{{ car.make }}</td>
            <td>{{ car.model }}</td>
            <td>{{ car.is_available|yesno:"Yes,No" }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>

5. ตัวอย่าง Flask API (MongoDB)

from flask import Flask, request, jsonify
from pymongo import MongoClient
from use_cases.car_crud import CarCRUDUseCase
from frameworks.mongo_car_repository import MongoCarRepository

app = Flask(__name__)
client = MongoClient("mongodb://localhost:27017")
db = client["car_rental"]
car_repo = MongoCarRepository(db)
car_crud = CarCRUDUseCase(car_repo)

@app.route('/cars', methods=['POST'])
def create_car():
    data = request.json
    car = car_crud.create_car(data['car_id'], data['make'], data['model'])
    return jsonify({"car_id": car.car_id, "make": car.make, "model": car.model, "is_available": car.is_available}), 201

@app.route('/cars', methods=['GET'])
def list_cars():
    cars = car_crud.list_cars()
    return jsonify([
        {"car_id": c.car_id, "make": c.make, "model": c.model, "is_available": c.is_available}
        for c in cars
    ])

6. Unit Test (Pure Python ไม่ใช้ Framework)

import unittest
from entities.car import Car
from use_cases.car_crud import CarCRUDUseCase

class InMemoryCarRepository:
    def __init__(self):
        self._cars = {}
    def save(self, car): self._cars[car.car_id] = car; return car
    def get_by_id(self, car_id): return self._cars.get(car_id)
    def get_all(self): return list(self._cars.values())
    def delete(self, car_id): self._cars.pop(car_id, None)

class TestCarCRUDUseCase(unittest.TestCase):
    def setUp(self):
        self.repo = InMemoryCarRepository()
        self.usecase = CarCRUDUseCase(self.repo)

    def test_create_car(self):
        car = self.usecase.create_car(1, 'Honda', 'Civic')
        self.assertEqual(car.car_id, 1)
        self.assertEqual(car.make, 'Honda')

    def test_list_cars(self):
        self.usecase.create_car(2, 'Toyota', 'Yaris')
        cars = self.usecase.list_cars()
        self.assertEqual(len(cars), 1)
        self.assertEqual(cars[0].model, 'Yaris')

if __name__ == '__main__':
    unittest.main()

สรุปสำคัญ

UI (Flask/Django/HTML): แค่รับ input, เรียก use case, ส่งผลลัพธ์ให้ template/response
Business logic ทดสอบได้จริง ไม่ผูกกับ framework หรือฐานข้อมูล
Repositories เป็นตัวกลาง storage จะเปลี่ยน Mongo, SQL หรือ In-memory test ก็ได้
Template มีหน้าที่แค่แสดงข้อมูลเท่านั้น ไม่มี logic ธุรกิจปะปน

ถ้าคุณสามารถรัน logic หลัก + เทสต์ โดยไม่ต้องพึ่ง web server หรือฐานข้อมูล แสดงว่าคุณสร้างระบบที่ยืดหยุ่นและพร้อมต่อยอดในอนาคตแล้ว!

ต้องการเจาะลึกแต่ละ layer, form, หรือ repo ตัวอย่างครบชุด DM มาได้เลยครับ!