OdooでEvent Sourcing駆動によるデータモデルを操作する設計と実装

OdooでEvent Sourcingを導入しスナップショット管理と誤操作・システムエラー記録を行う設計

本記事は、Copilotを使って生成しました。
2025/12/20 ソースコードが実行可能かどうか検証していません

はじめに

OdooはCRUDベースの設計ですが、業務システムでは 「履歴を残し、任意時点の状態を再現したい」 というニーズが強く、Event Sourcingパターンが有効です。本記事では、OdooにEvent Sourcingを導入し、スナップショット管理を行いながら、誤操作とシステムエラーを区別して記録する仕組みを設計・実装する方法をまとめます。

---

Event Sourcingの基本

• イベントストア: すべてのイベントを記録するテーブル
• アグリゲート: イベントを再生して現在の状態を導出する単位
• スナップショット: 最新状態をキャッシュして高速化や外部連携に利用

状態遷移図との違い

• 状態遷移図は「設計図」、Event Sourcingは「実装＋履歴管理」
• Event Sourcingはイベント履歴を永続化し、監査・再現が可能

---

EventStoreモデル

class EventStore(models.Model):
    _name = "event.store"
    _description = "Event Store"

    aggregate_id = fields.Char(required=True)
    model_name = fields.Char(required=True)
    event_type = fields.Char(required=True)
    payload = fields.Json(required=True)
    version = fields.Integer(required=True)
    timestamp = fields.Datetime(default=fields.Datetime.now)

    @api.model
    def create(self, vals):
        record = super(EventStore, self).create(vals)
        try:
            record._update_snapshot()
        except UserError as e:
            # 誤操作エラー
            super(EventStore, self).create({
                "aggregate_id": record.aggregate_id,
                "model_name": record.model_name,
                "event_type": "InvalidOperationAttempted",
                "payload": {
                    "failed_event": record.event_type,
                    "error_message": str(e),
                },
                "version": record.version,
            })
        except Exception as e:
            # システムエラー
            super(EventStore, self).create({
                "aggregate_id": record.aggregate_id,
                "model_name": record.model_name,
                "event_type": "SystemErrorOccurred",
                "payload": {
                    "failed_event": record.event_type,
                    "error_message": str(e),
                },
                "version": record.version,
            })
        return record

    def _update_snapshot(self):
        registry = self.env["snapshot.handler.registry"]
        handler = registry.get_handler(self.model_name)
        if handler:
            handler.apply_event(self)

---

スナップショット更新の仕組み

ハンドラの導入

スナップショット更新ロジックは業務ごとに異なるため、ハンドラに委譲します。

class RentalSnapshotHandler(models.AbstractModel):
    _name = "rental.snapshot.handler"

    def apply_event(self, event):
        snapshot_model = self.env["rental.snapshot"]
        snapshot = snapshot_model.search([("aggregate_id", "=", event.aggregate_id)], limit=1)

        current_status = snapshot.state.get("status") if snapshot else None

        if event.event_type == "DemoCancelled" and current_status != "demo":
            raise UserError("現在デモ状態ではないため、キャンセルできません")

        new_state = {"status": event.event_type, **event.payload}
        if snapshot:
            snapshot.write({"version": event.version, "state": new_state})
        else:
            snapshot_model.create({"aggregate_id": event.aggregate_id, "version": event.version, "state": new_state})

レジストリによる管理

class SnapshotHandlerRegistry(models.AbstractModel):
    _name = "snapshot.handler.registry"
    _description = "Snapshot Handler Registry"

    _handlers = {}

    def register_handler(self, model_name, handler_class):
        self._handlers[model_name] = handler_class

    def get_handler(self, model_name):
        return self._handlers.get(model_name)

    def auto_register_handlers(self):
        for model_name, model_class in self.env.registry.models.items():
            if model_name.endswith(".snapshot.handler"):
                aggregate_type = model_name.split(".")[0]
                self._handlers[aggregate_type] = self.env[model_name]

---

命名規約

• シンプル志向: rental.snapshot.handler
• ドメイン別整理: snapshot.handler.rental
• モジュール階層整理: inventory.snapshot.rental

命名規約を統一することで、自動検出・拡張・廃止が容易になります。

---

不整合イベントの扱い方：拒否 vs 失敗イベント記録

拒否する場合

アグリゲートが状態遷移ルールを持ち、矛盾するイベントは発行できないようにします。

def cancel_demo(self):
    if self.status != "demo":
        raise UserError("現在デモ状態ではないため、キャンセルできません")
    self._apply_event("DemoCancelled", {})

失敗イベントを記録する場合

誤操作も「事実」としてイベントストアに残すことで監査性を高めます。

def cancel_demo(self):
    if self.status != "demo":
        self._apply_event("InvalidOperationAttempted", {"operation": "DemoCancelled"})
        return
    self._apply_event("DemoCancelled", {})

RentalSnapshotHandler.apply_event()内でもDemoCancelledに対して、誤操作エラーの検証をしています。

まとめ

• 拒否: 状態の整合性を優先
• 失敗イベント記録: 監査性を優先
• システム要件に応じて選択、または両方を組み合わせるのがベストプラクティス

---

誤操作エラー vs システムエラー

誤操作エラー (Business Logic Error)

• ユーザー操作が状態遷移ルールに違反している場合
• → InvalidOperationAttempted として記録

システムエラー (Technical Error)

• DB障害、ハンドラのバグ、外部API失敗など
• → SystemErrorOccurred として記録

---

運用イメージ

1. ユーザーがイベントを追加 → EventStoreに記録
2. EventStoreがレジストリを参照 → 該当ハンドラを呼び出し
3. ハンドラがスナップショットを更新
4. 誤操作やシステムエラーは区別してイベントストアに記録
5. ExcelやBIツールでスナップショットや誤操作率を分析可能

---

まとめ

• OdooにEvent Sourcingを導入することで、履歴管理と監査性を強化できる
• スナップショットを表形式で保持すれば、業務ユーザーがExcelで直感的に扱える
• ハンドラ＋レジストリ＋命名規約により、段階的拡張や柔軟な廃止が可能
• 誤操作とシステムエラーを区別して記録することで、監査性と運用性を両立できる