Complete CRUD Implementation — From the Data Store to the API Endpoint

Q: Repository は必ずクラスでなければなりませんか?

いいえ, 関数でも構いません。ただし, クラスは DB セッションステートをカプセル化し, メソッド間でセッションを共有しやすいため, クラスアプローチを推奨します。

Q: flush と commit の違いは何ですか?

flush は SQL をデータベースに送信しますが, トランザクションはコミットしません。自動採番 ID の取得やデータベース制約チェックのトリガーができます。 commit はトランザクションをコミットし, 変更を永続化します。 yield 依存では flush の後に自動 commit することをお勧めします。

Q: N+1 問題はループ内でのみ発生しますか?

主に関連属性の反復処理時に発生しますが, Pydantic のシリアライズ時にもトリガーされる可能性があります — response_model が関連フィールドにアクセスする場合です。必ずプリロードしてください。

Q: bulk_create にサイズ制限はありますか?

フレームワーク側の制限はありませんが, PostgreSQL は1操作あたりのバインドパラメータ数を 32,767 に制限しています。数千件の挿入は問題ありませんが, 数万件の場合はバッチに分割することをお勧めします。

Q: オフセットページネーションは大規模データセットでどれくらい遅いですか?

OFFSET 1000000 はまず最初の100万行をスキャンしてからスキップするため, クエリ時間はオフセットに比例します。数百万件のデータセットではカーソルページネーションを推奨します。

Q: トランザクション内で部分失敗を処理するにはどうすればよいですか?

yield に関連するロールバックは例外発生時に自動的にトリガーされます。部分的なロールバックが必要な場合は, セーブポイント ( await session.begin_nested() )を使用してください。

1. 学ぶ内容

Repository Pattern:ProductRepository/PriceRepository のカプセル化
ページネーションとソート:limit/offset と order_by の Pydantic クエリモデル
関連データのプリロード:selectinload vs joinedload — パフォーマンスのトレードオフ
トランザクション管理:クロステーブル操作の一貫性の確保
Alice シナリオ:一括価格インポート API — 1リクエストで1,000件の価格レコードを挿入するトランザクション処理

2. Alice のリアルストーリー

(1) ペインポイント:SQL がルーティング関数に散在

Alice の PriceTracker コードでは, SQL クエリがルートハンドラ関数内に直接書かれており, 20のエンドポイントに30箇所の重複した select(Product).where(...) ロジックがあります。すべての商品クエリに「ソフトデリートフィルター」条件を追加する必要が生じたとき, Alice は15箇所を修正しなければならず, 2箇所見落としたことで削除済みの商品が結果に表示されてしまいました。

(2) リポジトリパターンの解決策

データベース操作を Repository クラス内にカプセル化し, ルート関数は Repository のメソッドのみを呼び出すようにします。グローバルフィルターを追加するには, Repository 内の1箇所を変更するだけで済みます。

PYTHON

class ProductRepository:
    async def get_all(self, db: AsyncSession, filters: ProductFilter) -> list[Product]:
        stmt = select(Product).where(Product.deleted == False)
        # Pydantic モデルからフィルターを追加
        if filters.category:
            stmt = stmt.where(Product.category == filters.category)
        result = await db.execute(stmt)
        return result.scalars().all()

(3) 成果

SQL ロジックが Repository に集中し, ルーティング関数がより簡潔になりました (15行から5行に)。グローバルフィルター条件は1箇所の変更でシステム全体に反映されるため, フロントエンドの Bob は削除済み商品を見ることはありません。

3. レイヤードアーキテクチャとリポジトリパターン

(1) 4層アーキテクチャ

flowchart TD
    Router[Router Layer] -->|呼び出し| Service[Service Layer]
    Service -->|呼び出し| Repository[Repository Layer]
    Repository -->|クエリ| SQLAlchemy[SQLAlchemy ORM]
    SQLAlchemy -->|SQL| Database[(PostgreSQL)]
    
    style Router fill:#e3f2fd
    style Service fill:#fff3e0
    style Repository fill:#e8f5e9
    style SQLAlchemy fill:#fce4ec

レベル	責務	例
Router	パラメータバリデーション, レスポンスフォーマット	`@app.get("/products")`
Service	ビジネスロジック, トランザクション調整	`ProductService.create_with_price()`
Repository	データアクセス, SQL ラッパー	`ProductRepository.get_by_id()`
ORM	オブジェクトリレーショナルマッピング	`select(Product).where(...)`

(1) ▶ サンプル:ProductRepository

PYTHON

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession

class ProductRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, product_id: int) -> Product | None:
        stmt = select(Product).where(Product.id == product_id)
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def get_all(
        self,
        category: str | None = None,
        skip: int = 0,
        limit: int = 20,
    ) -> list[Product]:
        stmt = select(Product).offset(skip).limit(limit)
        if category:
            stmt = stmt.where(Product.category == category)
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def create(self, product: ProductCreate) -> Product:
        db_product = Product(**product.model_dump())
        self.db.add(db_product)
        await self.db.flush()
        await self.db.refresh(db_product)
        return db_product

    async def update(self, product_id: int, data: dict) -> Product | None:
        stmt = (
            update(Product)
            .where(Product.id == product_id)
            .values(**data)
            .returning(Product)
        )
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def delete(self, product_id: int) -> bool:
        stmt = delete(Product).where(Product.id == product_id)
        result = await self.db.execute(stmt)
        return result.rowcount > 0

TEXT

# 関数定義成功

(2) ▶ サンプル:PriceRepository

PYTHON

class PriceRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_product(
        self,
        product_id: int,
        min_price: float | None = None,
        max_price: float | None = None,
        skip: int = 0,
        limit: int = 50,
    ) -> list[Price]:
        stmt = (
            select(Price)
            .where(Price.product_id == product_id)
            .order_by(Price.recorded_at.desc())
            .offset(skip)
            .limit(limit)
        )
        if min_price is not None:
            stmt = stmt.where(Price.price >= min_price)
        if max_price is not None:
            stmt = stmt.where(Price.price <= max_price)
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def bulk_create(self, prices: list[PriceCreate]) -> list[Price]:
        db_prices = [Price(**p.model_dump()) for p in prices]
        self.db.add_all(db_prices)
        await self.db.flush()
        return db_prices

TEXT

# 関数定義成功

4. ページネーションとソート

(1) Pydantic クエリモデル

ページネーションパラメータを Pydantic モデルにカプセル化し, 各エンドポイントでクエリパラメータを繰り返し宣言するのを防ぎます。

(1) ▶ サンプル:ページネーションクエリモデル

PYTHON

from pydantic import BaseModel, Field
from typing import Optional

class PaginationParams(BaseModel):
    skip: int = Field(0, ge=0, description="スキップするレコード数")
    limit: int = Field(20, ge=1, le=100, description="返す最大レコード数")

class ProductFilter(PaginationParams):
    category: Optional[str] = Field(None, max_length=100)
    min_price: Optional[float] = Field(None, ge=0, description="最低価格 (USD)")
    max_price: Optional[float] = Field(None, ge=0, description="最高価格 (USD)")
    sort_by: str = Field("name", pattern="^(name|base_price|created_at)$")
    sort_order: str = Field("asc", pattern="^(asc|desc)$")

TEXT

# 実行成功

(2) ▶ サンプル:ページネーションモデルを使用するエンドポイント

PYTHON

from fastapi import FastAPI, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

@app.get("/products", response_model=list[ProductResponse])
async def list_products(
    category: Optional[str] = Query(None),
    min_price: Optional[float] = Query(None, ge=0),
    max_price: Optional[float] = Query(None, ge=0),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
):
    repo = ProductRepository(db)
    return await repo.get_all(
        category=category,
        min_price=min_price,
        max_price=max_price,
        skip=skip,
        limit=limit,
    )

TEXT

# 関数定義成功

(2) ページネーション戦略の比較

戦略	実装	利点	欠点
オフセットページネーション	`OFFSET n LIMIT m`	シンプル, ページジャンプ対応	大きなオフセットでパフォーマンス低下
カーソルページネーション	`WHERE id > cursor LIMIT m`	大規模データセットで良好	ページジャンプ非対応
キーセットページネーション	`WHERE created_at < last LIMIT m`	ソートフィールドでのページネーションが良好	順序付きフィールドが必要

5. 関連データのプリロード

(1) N+1 クエリ問題

(1) ▶ サンプル:N+1 問題のデモ

PYTHON

# 悪い例:N+1 クエリ - 1つのプロダクトクエリ + Nつの価格クエリ
stmt = select(Product)
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
    # 各アクセスが個別のクエリをトリガー!
    print(len(p.prices))  # N クエリ

TEXT

# 実行成功

(2) ▶ サンプル:selectinload プリロード

PYTHON

from sqlalchemy.orm import selectinload, joinedload

# 良い例:合計2クエリ - products + prices (SELECT IN を使用)
stmt = select(Product).options(selectinload(Product.prices))
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
    print(len(p.prices))  # 追加クエリなし

TEXT

# 実行成功

(2) selectinload vs joinedload

側面	selectinload	joinedload
SQL 戦略	SELECT ... WHERE id IN (...)	LEFT OUTER JOIN
クエリ数	2クエリ	1クエリ
データ量	正確 (重複行なし)	JOIN で重複行が発生する可能性
ユースケース	1対多 (コレクション)	多対1/1対1
パフォーマンス	大規模データセットで有利	リレーションが少ないデータセットで有利
重複削除	不要	`unique()` が必要

(3) ▶ サンプル:価格をプリロードした商品クエリ

PYTHON

async def get_product_with_prices(
    self, product_id: int
) -> Product | None:
    stmt = (
        select(Product)
        .options(selectinload(Product.prices))
        .where(Product.id == product_id)
    )
    result = await self.db.execute(stmt)
    return result.scalar_one_or_none()

TEXT

# 関数定義成功

6. トランザクション管理

(1) クロステーブルトランザクション

(1) ▶ サンプル:一括価格インポートトランザクション

PYTHON

from sqlalchemy.ext.asyncio import AsyncSession

class PriceService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.price_repo = PriceRepository(db)
        self.product_repo = ProductRepository(db)

    async def bulk_import_prices(
        self,
        product_id: int,
        prices: list[PriceCreate],
    ) -> list[Price]:
        # 商品の存在を確認
        product = await self.product_repo.get_by_id(product_id)
        if not product:
            raise HTTPException(status_code=404, detail="Product not found")

        # すべての価格が同じ商品を参照していることを確認
        for p in prices:
            p.product_id = product_id

        # トランザクション内で一括挿入 (DB セッションがコミット/ロールバックを処理)
        db_prices = await self.price_repo.bulk_create(prices)

        # 商品の基本価格を最新に更新
        latest_price = db_prices[-1]
        await self.product_repo.update(
            product_id,
            {"base_price": latest_price.price},
        )

        return db_prices

TEXT

# 関数定義成功

(2) ▶ サンプル:一括インポートのエンドポイント

PYTHON

@app.post("/products/{product_id}/prices/bulk", response_model=list[PriceResponse])
async def bulk_import(
    product_id: int = Path(gt=0),
    prices: list[PriceCreate] = ...,
    db: AsyncSession = Depends(get_db),
    user: dict = Depends(get_current_user),
):
    # サブスクリプションチェック:Free ユーザーは1000価格まで制限
    if user["subscription"] == "free" and len(prices) > 1000:
        raise HTTPException(
            status_code=403,
            detail="Free plan limited to 1000 prices per import. Upgrade to Pro.",
        )
    
    service = PriceService(db)
    return await service.bulk_import_prices(product_id, prices)

TEXT

# 関数定義成功

❓ よくある質問

Q Repository は必ずクラスでなければなりませんか?

A いいえ, 関数でも構いません。ただし, クラスは DB セッションステートをカプセル化し, メソッド間でセッションを共有しやすいため, クラスアプローチを推奨します。

Q flush と commit の違いは何ですか?

A flush は SQL をデータベースに送信しますが, トランザクションはコミットしません。自動採番 ID の取得やデータベース制約チェックのトリガーができます。commit はトランザクションをコミットし, 変更を永続化します。yield 依存では flush の後に自動 commit することをお勧めします。

Q N+1 問題はループ内でのみ発生しますか?

A 主に関連属性の反復処理時に発生しますが, Pydantic のシリアライズ時にもトリガーされる可能性があります — response_model が関連フィールドにアクセスする場合です。必ずプリロードしてください。

Q bulk_create にサイズ制限はありますか?

A フレームワーク側の制限はありませんが, PostgreSQL は1操作あたりのバインドパラメータ数を 32,767 に制限しています。数千件の挿入は問題ありませんが, 数万件の場合はバッチに分割することをお勧めします。

Q オフセットページネーションは大規模データセットでどれくらい遅いですか?

A OFFSET 1000000 はまず最初の100万行をスキャンしてからスキップするため, クエリ時間はオフセットに比例します。数百万件のデータセットではカーソルページネーションを推奨します。

Q トランザクション内で部分失敗を処理するにはどうすればよいですか?

A yield に関連するロールバックは例外発生時に自動的にトリガーされます。部分的なロールバックが必要な場合は, セーブポイント (await session.begin_nested())を使用してください。

📖 まとめ

リポジトリパターンは SQL 操作を Repository クラス内にカプセル化し, ルーティング関数は HTTP ロジックに集中できます
ページネーションパラメータを Pydantic モデルとしてカプセル化し, 各エンドポイントでのクエリパラメータの繰り返し宣言を防ぎます
selectinload は1対多のプリロードに適し (重複行なし), joinedload は多対1に適します (単一クエリ)
バッチ操作はトランザクション内で実行し, yield による自動コミット/ロールバック管理に依存します
PriceTracker の一括価格インポートは数千レコードをサポートし, Free ユーザーは1,000レコードに制限されます

📝 練習問題

基本問題 (難易度 ⭐):ProductRepository の get_by_id と get_all メソッドを実装し, エンドポイントに DB セッションを注入して呼び出してください。ヒント:select(Product).where(Product.id == product_id)
応用問題 (難易度 ⭐⭐):ページネーションクエリのサポートを追加し (skip と limit パラメータを使用), selectinload(Product.prices) プリロードを実装し, 価格を含む商品詳細を返してください。ヒント:options(selectinload(...))
チャレンジ問題 (難易度 ⭐⭐⭐):PriceService.bulk_import_prices を実装し, 商品の存在を確認し, 価格を一括挿入し, 商品の base_price を更新してください — すべて1つのトランザクション内で。Free ユーザーは1,000レコードに制限してください。ヒント:bulk_create + update + Depends(get_current_user)

---|

Web-Tutorial 技術チーム

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

私たちについて

404 Not Found

Complete CRUD Implementation — From the Data Store to the API Endpoint

1. 学ぶ内容

2. Alice のリアルストーリー

(1) ペインポイント:SQL がルーティング関数に散在

(2) リポジトリパターンの解決策

(3) 成果

3. レイヤードアーキテクチャとリポジトリパターン

(1) 4層アーキテクチャ

(1) ▶ サンプル:ProductRepository

(2) ▶ サンプル:PriceRepository

4. ページネーションとソート

(1) Pydantic クエリモデル

(1) ▶ サンプル:ページネーションクエリモデル

(2) ▶ サンプル:ページネーションモデルを使用するエンドポイント

(2) ページネーション戦略の比較

5. 関連データのプリロード

(1) N+1 クエリ問題

(1) ▶ サンプル:N+1 問題のデモ

(2) ▶ サンプル:selectinload プリロード

(2) selectinload vs joinedload

(3) ▶ サンプル:価格をプリロードした商品クエリ

6. トランザクション管理

(1) クロステーブルトランザクション

(1) ▶ サンプル:一括価格インポートトランザクション

(2) ▶ サンプル:一括インポートのエンドポイント

❓ よくある質問

📖 まとめ

📝 練習問題