クリーンアーキテクチャで大規模なPythonシステムを構築する(実践サンプル・図解付き)

なぜクリーンアーキテクチャなのか?

システムが大きくなるほど、フレームワーク依存やロジックのスパゲティ化、テスト困難が発生しやすくなります。
クリーンアーキテクチャは、「ビジネスロジック(業務ルール)」をフレームワークやDB、UIから分離して保守性・拡張性を向上させる設計手法です。

「ビジネスロジックはフレームワークやDBから独立し、テストや変更が簡単にできるべきだ」


重要な原則

コアロジックは、どんなフレームワークにも依存せずテスト可能であるべきです。
(エンティティやユースケースはPythonだけでテストできるのが理想)


クリーンアーキテクチャのレイヤ構造

レイヤごとに責務を明確に分けることで、拡張性・テスト容易性・保守性が格段にアップします。

  1. Entities(エンティティ): 業務データを表現する純粋なクラス(例:Car, Rental
  2. Use Cases(ユースケース): 業務処理のロジック(例:「車を借りる」「車一覧取得」など)
  3. Repositories/Interfaces(リポジトリ・インターフェース): データアクセスや外部連携の抽象化
  4. Frameworks/Drivers(フレームワーク/ドライバ): Django, Flask, DB, 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

フロー図(シーケンス図)

「車を借りる」処理の流れを例に、各クラスがどう連携するか示します。

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: 車情報取得
    CarRepo-->>UC: Car entity
    UC->>CarRepo: save(car) (利用不可に更新)
    CarRepo->>DB: 車情報更新
    UC->>RentalRepo: save(rental)
    RentalRepo->>DB: レンタル記録登録
    RentalRepo-->>UC: Rental entity
    UC-->>UI: 結果返却
    UI-->>User: 結果表示(成功/エラー)

各レイヤのサンプルコード(Python)

1. Entities(エンティティ層)

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. リポジトリ実装例

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ビュー + テンプレート例

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テンプレート(car_list.html):

<!DOCTYPE html>
<html>
<head>
    <title>Car List</title>
</head>
<body>
    <h1>利用可能な車一覧</h1>
    <table border="1">
        <tr>
            <th>ID</th>
            <th>メーカー</th>
            <th>モデル</th>
            <th>利用可能</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:"はい,いいえ" }}</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. ユニットテスト例(Pythonのみ、DB/フレームワーク不要)

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): ユーザー入力を受け、ユースケースを呼び出し、テンプレート/レスポンスに結果を渡すだけ
  • ビジネスロジックはテスト可能・再利用可能・フレームワーク非依存
  • リポジトリでストレージを抽象化し、MongoDB/SQL/インメモリどれでも切替OK
  • テンプレートはデータの表示だけ。ビジネスロジックは絶対に混ぜない

コアロジックやテストが、WebサーバーやDBに依存せず動作すれば、将来も安心して発展できるアーキテクチャです。


より詳しく知りたい方は、フォームや完全サンプルリポジトリのリクエストもお気軽にどうぞ!


Get in Touch with us

Chat with Us on LINE

iiitum1984

Speak to Us or Whatsapp

(+66) 83001 0222

Related Posts

Our Products