【アーキテクチャパターン】 Pythonでリポジトリパターンを書いてみる

この文章はAIを使用して作成しています。

1. なぜリポジトリパターンが必要なのか

従来の課題：ビジネスロジックとデータアクセスの混在

リポジトリパターンを使わない場合、ビジネスロジックの中にデータアクセスのコードが直接書き込まれがちです。

# ❌ パターンを使わない場合
import sqlite3

def register_user(user_id: int, name: str, email: str) -> None:
    conn = sqlite3.connect("app.db")
    row = conn.execute("SELECT id FROM users WHERE id = ?", (user_id,)).fetchone()
    if row:
        raise ValueError(f"ユーザー {user_id} は既に存在します")
    conn.execute("INSERT INTO users VALUES (?, ?, ?)", (user_id, name, email))
    conn.commit()

このコードには次の3つの問題があります。

1. 変更に弱い DBをSQLiteからPostgreSQLに変えたい場合、register_user を直接書き直さなければなりません。データアクセスの都合がビジネスロジックに波及してしまいます。

2. テストが難しい 関数の中でDBコネクションを直接生成しているため、テスト時にも実際のDBが必要になります。テストのたびにDBのセットアップ・クリーンアップが必要になり、テストが遅く・不安定になります。

3. 同じSQLが散らばる アプリが大きくなるにつれ、似たようなSQLがサービス層・コントローラー層などあちこちに重複して書かれていきます。修正漏れやバグの温床になります。

解決策：リポジトリパターン

リポジトリパターンは「データの保存・取得をどう行うか 」をビジネスロジックから切り離し、専用のクラス（リポジトリ）に閉じ込める設計パターンです。

ポイントは2段階の分離です。

• 抽象リポジトリ （インターフェース）をドメイン層に置き、「何ができるか」だけを定義する
• 具体的な実装 （SQLite、インメモリ等）をインフラ層に隔離し、「どうやるか」を閉じ込める

# ✅ リポジトリパターンを使う場合
class UserService:
    def __init__(self, repo: UserRepository):  # 抽象に依存
        self.repo = repo

    def register(self, user_id: int, name: str, email: str) -> None:
        if self.repo.find_by_id(user_id):
            raise ValueError(f"ユーザー {user_id} は既に存在します")
        self.repo.save(User(id=user_id, name=name, email=email))

UserService は UserRepository というインターフェースにしか依存していません。裏側がSQLiteなのかインメモリなのかを知らなくてよいため、次のことが可能になります。

• DBを別の種類に乗り換えても、UserService には一切手を加えなくてよい
• テスト時は InMemoryUserRepository に差し替えるだけで、DBなしで高速にテストできる
• データアクセスのコードがリポジトリクラスに集約されるため、SQLの重複が起きない

---

2. 全体構造を掴む

3つの層と役割

リポジトリパターンはアプリケーションを3つの層に分けて考えます。

層	役割	依存してよいもの
domain（ドメイン層）	ビジネスの概念を定義する。エンティティと抽象リポジトリを置く。	なし（最も内側）
service（サービス層）	ビジネスロジックを実装する。抽象リポジトリを通じてデータを扱う。	domain層のみ
infrastructure（インフラ層）	データの実際の保存・取得を担う。DBやAPIとのやりとりを閉じ込める。	domain層のみ

infrastructure は domain を知っていますが、domain は infrastructure を知りません。この一方向の依存 がパターンの核心です。内側の層（domain）が外側の層（infrastructure）に依存してしまうと、DBを変えるたびにビジネスロジックの修正が必要になってしまいます。

                  ┌─────────────────────────┐
                  │       domain 層          │
                  │  User（エンティティ）     │
                  │  UserRepository（抽象）   │
                  └────────────┬────────────┘
                               │ 依存
          ┌────────────────────┼───────────────────┐
          │                   │                   │
          ▼                   ▼                   ▼
  ┌───────────────┐  ┌────────────────┐  ┌───────────────────┐
  │  service 層   │  │infrastructure層│  │     main.py        │
  │  UserService  │  │ InMemory実装   │  │ どの実装を使うか    │
  │               │  │ SQLite実装     │  │ ここだけが知っている │
  └───────────────┘  └────────────────┘  └───────────────────┘

ディレクトリ構成

層ごとにディレクトリを分けることで、依存関係がフォルダ構造として視覚化されます。

my_app/
├── domain/
│   ├── user.py                        # エンティティ
│   └── user_repository.py             # 抽象リポジトリ
├── infrastructure/
│   ├── in_memory_user_repository.py   # インメモリ実装（テスト・開発用）
│   └── sqlite_user_repository.py      # SQLite実装（本番用）
├── service/
│   └── user_service.py                # ビジネスロジック
└── main.py                            # エントリーポイント（実装の選択）

---

3. 実装を読む

domain層：ビジネスの概念を定義する

domain/user.py — エンティティ

アプリケーションの中心となるデータ構造です。データベースや通信の都合に依存せず、純粋にビジネス上の「ユーザー」という概念だけを表します。他のどのモジュールにも依存しない、最も内側の層です。

from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str
    email: str

domain/user_repository.py — 抽象リポジトリ

「ユーザーをどう保存・取得するか」のインターフェース（契約）だけを定義します。具体的な実装（SQLなのかAPIなのか）はここには一切登場しません。domain 層に置くことで、ビジネスロジックは「どんなデータソースでも同じように扱える」という前提で書けるようになります。

from abc import ABC, abstractmethod
from domain.user import User

class UserRepository(ABC):
    @abstractmethod
    def find_by_id(self, user_id: int) -> User | None:
        pass

    @abstractmethod
    def save(self, user: User) -> None:
        pass

    @abstractmethod
    def find_all(self) -> list[User]:
        pass

infrastructure層：実装の詳細を閉じ込める

同じ UserRepository インターフェースを、異なる技術で実装します。どちらを使っても service 層からは同じように扱えます。

infrastructure/in_memory_user_repository.py — インメモリ実装

辞書をストレージとして使うシンプルな実装です。DBのセットアップが不要なので、テストや開発初期に重宝します。UserRepository の契約さえ守っていれば、内部実装は何でも構いません。

from domain.user import User
from domain.user_repository import UserRepository

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self._store: dict[int, User] = {}

    def find_by_id(self, user_id: int) -> User | None:
        return self._store.get(user_id)

    def save(self, user: User) -> None:
        self._store[user.id] = user

    def find_all(self) -> list[User]:
        return list(self._store.values())

infrastructure/sqlite_user_repository.py — SQLite実装

同じインターフェースを今度はSQLiteで実装します。SQL文やコネクション管理といった「インフラの詳細」がこのクラスの中だけに閉じ込められているのがポイントです。将来PostgreSQLやMongoDBに乗り換えたとしても、このファイルを差し替えるだけで済みます。

import sqlite3
from domain.user import User
from domain.user_repository import UserRepository

class SQLiteUserRepository(UserRepository):
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)"
        )

    def find_by_id(self, user_id: int) -> User | None:
        row = self.conn.execute(
            "SELECT id, name, email FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        return User(*row) if row else None

    def save(self, user: User) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO users VALUES (?, ?, ?)",
            (user.id, user.name, user.email)
        )
        self.conn.commit()

    def find_all(self) -> list[User]:
        rows = self.conn.execute("SELECT id, name, email FROM users").fetchall()
        return [User(*row) for row in rows]

service層：ビジネスロジックを書く

service/user_service.py

「ユーザー登録時に重複チェックをする」といったアプリケーション固有のルールをここに書きます。UserRepository という抽象にだけ依存しているため、裏側がSQLiteかインメモリかをまったく意識しません。これがリポジトリパターンの最大の恩恵です。

from domain.user import User
from domain.user_repository import UserRepository

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def register(self, user_id: int, name: str, email: str) -> None:
        if self.repo.find_by_id(user_id):
            raise ValueError(f"ユーザー {user_id} は既に存在します")
        self.repo.save(User(id=user_id, name=name, email=email))

    def get_user(self, user_id: int) -> User:
        user = self.repo.find_by_id(user_id)
        if not user:
            raise ValueError(f"ユーザー {user_id} が見つかりません")
        return user

---

4. 組み立てる — main.py

どの実装を使うかを決める唯一の場所です。repo = ... の1行を差し替えるだけで動作環境を切り替えられます。UserService 自身はこの選択を知らなくてよいため、ビジネスロジックに一切手を加えずに環境を変えられます。

from infrastructure.in_memory_user_repository import InMemoryUserRepository
from infrastructure.sqlite_user_repository import SQLiteUserRepository
from service.user_service import UserService

# テスト・開発時
repo = InMemoryUserRepository()

# 本番時（1行差し替えるだけ）
# repo = SQLiteUserRepository("prod.db")

service = UserService(repo=repo)

service.register(1, "田中太郎", "tanaka@example.com")
service.register(2, "鈴木花子", "suzuki@example.com")

user = service.get_user(1)
print(user)

all_users = repo.find_all()
print(all_users)

main.py だけが全体を知っています。service 層も infrastructure 層も、互いの存在を直接知ることなく、domain 層の抽象を介して連携しています。

---

5. まとめ

リポジトリパターンが解決する3つの課題と、その手段を整理します。

課題	解決手段
DBを変えるとビジネスロジックまで変更が必要	抽象リポジトリに依存させることで、実装の差し替えをビジネスロジックから隔離する
テストに実DBが必要で遅い・不安定	InMemoryUserRepository に差し替えるだけでDBなしテストが可能になる
SQLが各所に散らばり保守しにくい	データアクセスのコードをリポジトリクラスに集約する

設計の核心は依存の方向を一方向に保つこと です。domain 層は何も知らなくてよく、infrastructure 層だけが domain 層を知る。この原則を守ることで、アプリケーションの内側（ビジネスロジック）を外側の都合（DBの種類、通信方式）から守り続けることができます。

---

補足：Webアプリに発展させたら main.py はどう変わる？

今回のサンプルでは main.py がインスタンスの組み立てとサービスの呼び出しを兼ねていますが、Webアプリに発展させると役割が分離されます。

変更前（今回のサンプル）

my_app/
├── domain/
├── infrastructure/
├── service/
└── main.py   ← サービスの直接呼び出しを兼ねている

変更後（Webアプリ）

my_app/
├── domain/
├── infrastructure/
├── service/
├── controller/
│   └── user_controller.py   ← HTTPリクエストを受け取りServiceを呼ぶ
├── dependencies.py           ← DIの組み立てを集約する
└── main.py                  ← サーバー起動のみ

コントローラーはHTTPリクエストを受け取り、サービスに処理を委譲して、レスポンスを返す専用の層です。main.py はサーバーの起動だけに専念し、リクエストの処理には関与しなくなります。

DIの組み立ては dependencies.py に集約します。エンドポイントが増えてもDB接続の設定変更がここ1か所に収まります。FastAPIでは Depends() を使った依存性注入によって、リクエストのたびに get_user_service() が呼び出され、返ってきた UserService インスタンスがコントローラーの引数に自動的に注入されます。

dependencies.py

def get_user_service() -> UserService:
    repo = SQLiteUserRepository("prod.db")
    return UserService(repo=repo)

controller/user_controller.py （FastAPIの例）

@router.post("/users")
def create_user(body: CreateUserRequest, service: UserService = Depends(get_user_service)):
    service.register(body.id, body.name, body.email)

main.py

app = FastAPI()
app.include_router(user_controller.router)

domain / service / infrastructure の3層はそのまま変わりません。Webアプリへの発展は、main.py にコントローラー層を加えるだけで済みます。