2025-03-03

[MtkN1]

Today I Learned ...

[MtkN1]

ドメイン駆動設計

ドメイン駆動設計を学習するためのプロンプトセットアップ用プロンプト。

Crypto 系 WebSocket API から取得したデータを永続化するスクリプトを常日頃書いているが、それにドメイン駆動設計 (DDD) を適用しようとしている。 しかし全くの素人なのでかみ砕きながら学んでいる。 そこで Copilot に DDD プロジェクトをセットアップしてももらうことで、その実際のディレクトリツリー構成からも学びを得ることができる (ただし実際の実装は使い物にならないことが多い)。 以下がその学習用セットアッププロンプト:

あなたはドメイン駆動設計 (DDD) のプロとして「WebSocket クライアントで取得したデータをデータベースに保存するバッチスクリプト」の Python プロジェクトを立ち上げてください。

テスト (tests) と実装 (src) を提供してください。 実装のライブラリとしては aiohttp (ClientSession) と SQLAlchemy を使います。

DDD は「エリック・エヴァンスのドメイン駆動設計」に従ってください。

[MtkN1]

./
├── src/
│   ├── application/
│   │   ├── __init__.py*
│   │   └── services.py*
│   ├── domain/
│   │   ├── __init__.py*
│   │   ├── entities.py*
│   │   ├── repositories.py*
│   │   └── value_objects.py*
│   ├── infrastructure/
│   │   ├── persistence/
│   │   │   ├── __init__.py*
│   │   │   ├── database.py*
│   │   │   ├── models.py*
│   │   │   └── repositories.py*
│   │   ├── websocket/
│   │   │   ├── __init__.py*
│   │   │   └── client.py*
│   │   └── __init__.py*
│   ├── interface/
│   │   ├── __init__.py*
│   │   └── cli.py*
│   └── main.py*
├── tests/
│   ├── application/
│   │   └── test_services.py*
│   ├── domain/
│   │   └── test_entities.py*
│   ├── infrastructure/
│   │   ├── persistence/
│   │   │   └── test_repositories.py*
│   │   └── websocket/
│   │       └── test_client.py*
│   ├── interface/
│   │   └── test_cli.py*
│   └── __init__.py*
├── .gitignore*
├── pyproject.toml*
├── README.md*
└── requirements.txt*

[MtkN1]

• infrastructure/* に WebSocket やデータベースの実装系を置く
  • つまり websocket/ に aiohttp での WebSocket 接続、persistence/ に SQLAlchemy でのセッションとか
• application/* に処理するロジックを書く
• main.py に最終的な手続きを書く

[MtkN1]

クリーンアーキテクチャによるリファクタリング提案

ご指摘の通り、現在のコードは密結合していてテストが困難です。WebSocketとデータベース処理が直接結合しているため、分離してテストすることが難しい状態です。

クリーンアーキテクチャへのリファクタリング

以下の構造でリファクタリングすることをお勧めします：

tdd-demo/
├── domain/ (ドメインモデルとロジック)
├── adapters/ (外部サービス連携)
├── repositories/ (データ永続化)
├── use_cases/ (アプリケーションロジック)
└── main.py (エントリポイント)

具体的なリファクタリング例

domain/models.py

from datetime import datetime
from decimal import Decimal
from dataclasses import dataclass

@dataclass
class Ticker:
    """ドメインモデルとしてのティッカー"""
    publish_time: datetime
    event_time: datetime
    symbol: str
    price: Decimal
    quantity: Decimal

adapters/websocket.py

import aiohttp
from typing import AsyncGenerator, Protocol
from domain.models import Ticker

class TickerSource(Protocol):
    async def subscribe(self) -> AsyncGenerator[Ticker, None]: ...

class BinanceWebSocketClient:
    """Binance WebSocketクライアント"""
    def __init__(self, symbol: str = "btcusdt"):
        self.url = f"wss://stream.binance.com/ws/{symbol}@ticker"
    
    async def subscribe(self) -> AsyncGenerator[Ticker, None]:
        """WebSocketからティッカー情報を購読"""
        from datetime import datetime, UTC
        from decimal import Decimal
        from main import get_type_adapter, TickerDict
        
        async with aiohttp.ClientSession() as client, client.ws_connect(self.url) as ws:
            async for msg in ws:
                ta = get_type_adapter(TickerDict)
                ticker_dict = ta.validate_json(msg.data)
                publish_time = datetime.now(tz=UTC)
                event_time = datetime.fromtimestamp(ticker_dict["E"] / 1000, tz=UTC)
                
                yield Ticker(
                    publish_time=publish_time,
                    event_time=event_time,
                    symbol=ticker_dict["s"],
                    price=Decimal(ticker_dict["c"]),
                    quantity=Decimal(ticker_dict["Q"])
                )

repositories/ticker_repository.py

from typing import Protocol
from domain.models import Ticker
from sqlalchemy.ext.asyncio import AsyncSession
from main import TickerModel

class TickerRepository(Protocol):
    async def save(self, ticker: Ticker) -> None: ...

class SQLAlchemyTickerRepository:
    def __init__(self, session_factory):
        self.session_factory = session_factory
    
    async def save(self, ticker: Ticker) -> None:
        """ティッカー情報をデータベースに保存"""
        ticker_model = TickerModel(
            publish_time=ticker.publish_time,
            event_time=ticker.event_time,
            symbol=ticker.symbol,
            price=ticker.price,
            quantity=ticker.quantity
        )
        
        async with self.session_factory.begin() as session:
            session.add(ticker_model)

use_cases/ticker_processor.py

from domain.models import Ticker
from adapters.websocket import TickerSource
from repositories.ticker_repository import TickerRepository

class TickerProcessor:
    def __init__(self, source: TickerSource, repository: TickerRepository):
        self.source = source
        self.repository = repository
    
    async def run(self) -> None:
        """ティッカー情報を購読して保存する"""
        async for ticker in self.source.subscribe():
            await self.repository.save(ticker)

main.py

import os
import anyio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from adapters.websocket import BinanceWebSocketClient
from repositories.ticker_repository import SQLAlchemyTickerRepository
from use_cases.ticker_processor import TickerProcessor

# ...existing code...

async def main() -> None:
    """Binance の WebSocket から ticker を受信して DB に保存する。"""
    engine = create_async_engine(os.getenv("DATABASE_URL"))
    try:
        session_factory = async_sessionmaker(engine, expire_on_commit=False)
        
        # 依存性の注入
        ticker_source = BinanceWebSocketClient()
        ticker_repository = SQLAlchemyTickerRepository(session_factory)
        processor = TickerProcessor(ticker_source, ticker_repository)
        
        # ユースケースの実行
        await processor.run()
    finally:
        await engine.dispose()

if __name__ == "__main__":
    anyio.run(main)

テスト設計

リファクタリング後は、各コンポーネントを個別にテストできます：

import pytest
from datetime import datetime, UTC
from decimal import Decimal
from unittest.mock import AsyncMock, MagicMock

from domain.models import Ticker
from use_cases.ticker_processor import TickerProcessor

class MockTickerSource:
    def __init__(self, tickers):
        self.tickers = tickers
    
    async def subscribe(self):
        for ticker in self.tickers:
            yield ticker

class MockTickerRepository:
    def __init__(self):
        self.saved_tickers = []
    
    async def save(self, ticker):
        self.saved_tickers.append(ticker)

@pytest.mark.asyncio
async def test_ticker_processor():
    # テスト用のデータを準備
    now = datetime.now(tz=UTC)
    test_tickers = [
        Ticker(
            publish_time=now,
            event_time=now,
            symbol="BTCUSDT",
            price=Decimal("25000.00"),
            quantity=Decimal("1.5")
        )
    ]
    
    # モックオブジェクトを作成
    source = MockTickerSource(test_tickers)
    repository = MockTickerRepository()
    
    # テスト実行
    processor = TickerProcessor(source, repository)
    await processor.run()
    
    # 検証
    assert len(repository.saved_tickers) == 1
    saved = repository.saved_tickers[0]
    assert saved.symbol == "BTCUSDT"
    assert saved.price == Decimal("25000.00")

統合テスト

最終的に「データベースにデータが保存される」ことを確認するなら：

import pytest
import os
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from datetime import datetime, UTC
from decimal import Decimal

from domain.models import Ticker
from adapters.websocket import TickerSource
from repositories.ticker_repository import SQLAlchemyTickerRepository
from use_cases.ticker_processor import TickerProcessor
from main import TickerModel, Base

class TestWebSocketSource:
    """テスト用のWebSocket代替クラス"""
    def __init__(self, tickers):
        self.tickers = tickers
    
    async def subscribe(self):
        for ticker in self.tickers:
            yield ticker

@pytest.fixture
async def db_session():
    """テスト用のインメモリDBセッション"""
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    yield session_factory
    
    await engine.dispose()

@pytest.mark.asyncio
async def test_ticker_processing_integration(db_session):
    # テストデータ
    now = datetime.now(tz=UTC)
    test_ticker = Ticker(
        publish_time=now,
        event_time=now,
        symbol="BTCUSDT",
        price=Decimal("25000.00"),
        quantity=Decimal("1.5")
    )
    
    # テスト用のソース作成
    source = TestWebSocketSource([test_ticker])
    
    # リポジトリと処理クラスを作成
    repository = SQLAlchemyTickerRepository(db_session)
    processor = TickerProcessor(source, repository)
    
    # 処理を実行
    await processor.run()
    
    # データベースからデータを確認
    async with db_session() as session:
        result = await session.execute(select(TickerModel))
        models = result.scalars().all()
        
        assert len(models) == 1
        assert models[0].symbol == "BTCUSDT"
        assert models[0].price == Decimal("25000.00")
        assert models[0].quantity == Decimal("1.5")

このようなアプローチで、Webソケット通信とデータベース操作を分離し、テスト可能な構造にすることができます。実装詳細に依存せず、テストが可能になります。