プロジェクト設計 — PriceTracker フルスタックアーキテクチャ青写真

プロジェクト設計は建物を建てる前の青写真のようなもの - 青図面なしで建設を始めると, 3階に到達した時点で基礎が足りないことに気づき, 取り壊してやり直すしかなくなります。良い青写真は描いて終わりではなく, 建設全体を通じての指針となります。

1. 学ぶ内容

• 要件分析:SaaS価格追跡のコアユースケース (ユーザー, 製品, 価格, サブスクリプション, 通知)
• ドメイン駆動設計:集約ルート, 値オブジェクト, ドメインイベントの特定
• 技術選定の意思決定:FastAPI + PostgreSQL + Redis + Celery + WebSocketを選択した理由
• APIバージョニング戦略:URLベースとヘッダーベースのトレードオフ
• アーキテクチャ青写真:Charlieのデプロイトポロジー - 負荷分散, データベース読み書き分離, キャッシュ層

2. Aliceの実話

(1) 悩み:機能が増えるほどコードが散乱

PriceTrackerはシンプルなAPIから始まりましたが, 機能が増えるにつれ統一された設計がなく - 製品と価格の境界が曖昧, サブスクリプションロジックが10のエンドポイントに散在, 通知システムがビジネスロジックと密結合していました。新しい機能を追加するたびに5つのファイルを修正し, 回帰バグが頻発していました。

(2) ドメイン駆動設計による解決策

ドメイン駆動設計 (DDD)はビジネスの視点からコアドメインと境界を特定します:Product集約, Price集約, User集約, Subscription集約 - 各集約は明確な境界と責任を持ち, 集約間の通信はイベントを通じて行われ, 相互依存を排除します。

(3) 成果

新機能の追加時には該当する集約内のコードのみを修正すればよく, 変更はシステム全体に影響しなくなりました。サブスクリプションロジックはSubscription集約に集約され, 通知はPriceUpdatedイベントでトリガーされ, 完全に疎結合になりました。

3. 要件分析

(1) コアユースケース

(1) ▶ サンプル:ユースケースを優先度順にソートするスクリプト

# scripts/prioritize_use_cases.py
use_cases = [
    {"role": "User", "action": "register_login", "priority": "P0", "effort": 2},
    {"role": "User", "action": "query_price", "priority": "P0", "effort": 1},
    {"role": "User", "action": "subscribe_alert", "priority": "P1", "effort": 3},
    {"role": "Admin", "action": "bulk_import", "priority": "P0", "effort": 5},
]

# 優先度でソート, 同じ優先度内では工数の昇順
order = {"P0": 0, "P1": 1, "P2": 2}
sorted_cases = sorted(use_cases, key=lambda x: (order[x["priority"]], x["effort"]))
for uc in sorted_cases:
    print(f"[{uc['priority']}] {uc['role']}: {uc['action']} (effort={uc['effort']}d)")

出力:

# 実行成功

(2) 非機能要件

4. ドメインモデル設計

(1) ER図

erDiagram
    User ||--o{ Subscription : has
    User ||--o{ Product : creates
    User ||--o{ Alert : sets
    Product ||--o{ Price : has
    Product ||--o{ Alert : watched_by
    Subscription ||--o{ Feature : includes
    
    User {
        int id PK
        string email UK
        string hashed_password
        string role
        datetime created_at
    }
    
    Product {
        int id PK
        string name
        string category
        float base_price
        string sku UK
        int user_id FK
        datetime created_at
    }
    
    Price {
        int id PK
        int product_id FK
        float price
        string currency
        string source
        datetime recorded_at
    }
    
    Subscription {
        int id PK
        int user_id FK
        string plan
        datetime starts_at
        datetime expires_at
    }
    
    Alert {
        int id PK
        int user_id FK
        int product_id FK
        float target_price
        string status
    }

erDiagram
    User ||--o{ Subscription : has
    User ||--o{ Product : creates
    User ||--o{ Alert : sets
    Product ||--o{ Price : has
    Product ||--o{ Alert : watched_by
    Subscription ||--o{ Feature : includes
    
    User {
        int id PK
        string email UK
        string hashed_password
        string role
        datetime created_at
    }
    
    Product {
        int id PK
        string name
        string category
        float base_price
        string sku UK
        int user_id FK
        datetime created_at
    }
    
    Price {
        int id PK
        int product_id FK
        float price
        string currency
        string source
        datetime recorded_at
    }
    
    Subscription {
        int id PK
        int user_id FK
        string plan
        datetime starts_at
        datetime expires_at
    }
    
    Alert {
        int id PK
        int user_id FK
        int product_id FK
        float target_price
        string status
    }

(2) 集約ルートの特定

(1) ▶ サンプル:ドメインイベント定義

# domain/events.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class EventType(Enum):
    PRICE_CHANGED = "price_changed"
    ALERT_TRIGGERED = "alert_triggered"
    USER_SUBSCRIBED = "user_subscribed"

@dataclass
class DomainEvent:
    event_type: EventType
    aggregate_id: int
    occurred_at: datetime = field(default_factory=datetime.utcnow)
    payload: dict = field(default_factory=dict)

# 使用例:価格変動イベント
price_event = DomainEvent(
    event_type=EventType.PRICE_CHANGED,
    aggregate_id=42,
    payload={"old_price": 299.9, "new_price": 249.9, "currency": "CNY"},
)

出力:

# 実行成功

5. 技術選定の意思決定

(1) モデル比較

(2) システムアーキテクチャ概要

flowchart TD
    Client[Client / Bob Frontend] --> Nginx[Nginx Load Balancer]
    Nginx --> API1[FastAPI Pod 1]
    Nginx --> API2[FastAPI Pod 2]
    Nginx --> APIN[FastAPI Pod N]
    
    API1 --> PG_Master[(PostgreSQL Master)]
    API2 --> PG_Master
    APIN --> PG_Master
    
    PG_Master --> PG_Replica[(PostgreSQL Replica)]
    
    API1 --> Redis[(Redis Cluster)]
    API2 --> Redis
    APIN --> Redis
    
    Redis --> CW1[Celery Worker 1]
    Redis --> CW2[Celery Worker 2]
    
    CW1 --> PG_Master
    CW2 --> PG_Master
    
    API1 --> WS[WebSocket Hub]
    API2 --> WS
    
    Prometheus[Prometheus] --> API1
    Grafana[Grafana] --> Prometheus

flowchart TD
    Client[Client / Bob Frontend] --> Nginx[Nginx Load Balancer]
    Nginx --> API1[FastAPI Pod 1]
    Nginx --> API2[FastAPI Pod 2]
    Nginx --> APIN[FastAPI Pod N]
    
    API1 --> PG_Master[(PostgreSQL Master)]
    API2 --> PG_Master
    APIN --> PG_Master
    
    PG_Master --> PG_Replica[(PostgreSQL Replica)]
    
    API1 --> Redis[(Redis Cluster)]
    API2 --> Redis
    APIN --> Redis
    
    Redis --> CW1[Celery Worker 1]
    Redis --> CW2[Celery Worker 2]
    
    CW1 --> PG_Master
    CW2 --> PG_Master
    
    API1 --> WS[WebSocket Hub]
    API2 --> WS
    
    Prometheus[Prometheus] --> API1
    Grafana[Grafana] --> Prometheus

6. APIバージョニング戦略

(1) URLバージョン vs ヘッダーバージョン

(1) ▶ サンプル:URLバージョニングルーティング構造

from fastapi import APIRouter

# V1ルート
v1_router = APIRouter(prefix="/api/v1", tags=["v1"])
v1_products = APIRouter(prefix="/products", tags=["products"])
v1_prices = APIRouter(prefix="/prices", tags=["prices"])
v1_auth = APIRouter(prefix="/auth", tags=["auth"])

# V2ルート (将来用)
v2_router = APIRouter(prefix="/api/v2", tags=["v2"])

# ルーターをマウント
app.include_router(v1_auth)
app.include_router(v1_products, dependencies=[Depends(get_current_user)])
app.include_router(v1_prices, dependencies=[Depends(get_current_user)])
app.include_router(v1_router)

出力:

# 実行成功

❓ よくある質問

📖 まとめ

• 要件分析で機能要件と非機能要件を区分し, 優先度と制約を明確にする
• DDD:集約ルート (User, Product, PriceImport)を特定し, トランザクション境界と一貫性ルールを定義
• ユースケースに基づく技術選定:FastAPI (非同期)+ PostgreSQL (JSONB)+ Redis (データ構造)+ Celery (タスクキュー)
• APIバージョニングはURLプレフィックス/api/v1/を使用, シンプルで直感的, キャッシュフレンドリ
• デプロイトポロジー:Nginxロードバランサー → FastAPI Pods → PostgreSQLマスター・スレーブ → Redisクラスタ → Celery Workers

📝 練習問題

1. 基本問題 (難易度 ⭐):PriceTrackerのシステムアーキテクチャ図 (FastAPI, PostgreSQL, Redis, Celery, Nginxを含む)を作成し, 各コンポーネントの役割とデータフローをラベル付けしてください。ヒント:Mermaid flowchart
2. 応用問題 (難易度 ⭐⭐):PriceTrackerドメインモデルのER図を設計し, 3つの集約ルート (User, Product, PriceImport)を特定し, 各集約に含まれるエンティティとその境界ルールを定義してください。ヒント:Mermaid erDiagram + 集約ルート表
3. チャレンジ (難易度 ⭐⭐⭐):PriceTrackerの完全なアーキテクチャ青写真を作成 - 要件ドキュメント (コアユースケース + 非機能要件), 技術比較表, APIバージョニングルーティング構造のコード, デプロイトポロジー図 (マスター・スレーブデータベース, キャッシュ層, Celeryクラスタを含む)を含めてください。ヒント:このレッスンで扱ったすべての内容を統合。

---|

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

私たちについて