Python × DynamoDB 実践ガイド：NoSQLデータベースの基本から応用まで

はじめに

WebアプリケーションやAPIサービスを開発する際、データの保存先として適切なデータベースを選択することは非常に重要です。従来のリレーショナルデータベース（RDB）が長く主流でしたが、クラウドネイティブな環境では、スケーラビリティとパフォーマンスに優れたNoSQLデータベースが急速に普及しています。

本記事では、AWS DynamoDBとPythonを組み合わせた実践的な開発方法について解説します。単なる基本的な使い方だけでなく、アーキテクチャ設計のベストプラクティスやパフォーマンスチューニングまで幅広くカバーしています。

サンプル実装はこちら

目次

1. NoSQLとは何か？RDBとの違い
2. DynamoDBの基本概念
3. Pythonでの接続方法：boto3ライブラリ
4. ローカル開発環境の構築
5. データモデリングのベストプラクティス
6. リポジトリパターンの実装
7. FastAPIとの統合
8. ユニットテストの書き方
9. パフォーマンスチューニングとコスト最適化
10. 参考文献

1. NoSQLとは何か？RDBとの違い

NoSQLの特徴

NoSQLデータベース（Not Only SQL）は、従来のリレーショナルデータベースとは異なるアプローチでデータを格納・取得します。主な特徴は以下の通りです：

• スキーマレス設計: テーブル構造を事前に厳密に定義する必要がない
• 水平スケーリング: サーバーを追加するだけで簡単に容量を増やせる
• 分散アーキテクチャ: データを複数のノードに分散して保存
• 高可用性: レプリケーションによる障害耐性
• 大量のデータ処理: 膨大な読み書きリクエストに対応

RDBとの比較

機能	リレーショナルDB	NoSQL (DynamoDB)
データモデル	テーブル、行、列（正規化）	キーと値、ドキュメント（非正規化）
スキーマ	固定	柔軟
トランザクション	ACID準拠	結果整合性（場合によりACID）
スケーリング	垂直（より強力なハードウェア）	水平（より多くのノード）
クエリ言語	SQL	専用API
結合操作	サポート	限定的または非サポート
ユースケース	複雑な関係、トランザクション	大規模データ、高スループット

使い分けのポイント

• RDBが適している場合:

  • 複雑なクエリや結合が必要
  • トランザクション整合性が重要
  • データ構造が予測可能で変化が少ない

• NoSQLが適している場合:

  • スケーラビリティが最優先
  • スキーマが頻繁に変更される
  • 読み取り/書き込みのパフォーマンスが重要
  • 大量のデータを扱う

2. DynamoDBの基本概念

DynamoDBとは

Amazon DynamoDBは、AWSが提供するフルマネージド型のNoSQLデータベースサービスです。どんなに大規模なアプリケーションでも、一貫した低レイテンシーでデータにアクセスできるように設計されています。

主な特徴：

• サーバーレス（プロビジョニング不要）
• 自動スケーリング
• マルチリージョンレプリケーション
• 自動バックアップ
• きめ細かなアクセス制御

基本的な構成要素

DynamoDBにおける重要な概念を理解しましょう：

テーブル、アイテム、属性

• テーブル: データの集合（RDBのテーブルに相当）
• アイテム: テーブル内の個々のデータ（RDBの行に相当）
• 属性: アイテム内の要素（RDBの列に相当）

プライマリキー

DynamoDBのテーブルには、以下のいずれかのプライマリキー構造が必要です：

• 単一キー（パーティションキーのみ）:

  • 一意の識別子として機能
  • 内部ハッシュ関数によりデータ分散の決定に使用

• 複合キー（パーティションキー + ソートキー）:

  • パーティションキー：データの分散を決定
  • ソートキー：同じパーティション内でのデータの順序付け

セカンダリインデックス

クエリの柔軟性を高めるための追加インデックス：

• グローバルセカンダリインデックス (GSI):

  • プライマリキーとは異なるパーティションキーとソートキーを持つ
  • テーブル全体にアクセス可能

• ローカルセカンダリインデックス (LSI):

  • 同じパーティションキーを使用するが、異なるソートキーを持つ
  • テーブル作成時にのみ定義可能

読み取り整合性モデル

DynamoDBは2種類の読み取り整合性を提供します：

• 結果整合性のある読み込み:

  • レイテンシーが低く、スループットが高い
  • 最新のデータが反映されていない可能性がある

• 強力な整合性のある読み込み:

  • 最新のデータを確実に取得
  • より多くのリソースを消費

3. Pythonでの接続方法：boto3ライブラリ

boto3の概要

boto3はAWSの公式Python SDKで、DynamoDBを含む様々なAWSサービスにアクセスするための統一されたインターフェースを提供します。

import boto3

# DynamoDBリソースの初期化
dynamodb = boto3.resource(
    'dynamodb',
    region_name='ap-northeast-1',
    aws_access_key_id='YOUR_ACCESS_KEY',  # 本番環境では環境変数を使用
    aws_secret_access_key='YOUR_SECRET_KEY'
)

# テーブルの参照
table = dynamodb.Table('YourTableName')

基本的なCRUD操作

項目の作成（Create）

def create_item(table, item_data):
    response = table.put_item(Item=item_data)
    return response

項目の読み取り（Read）

def get_item(table, partition_key_value, sort_key_value=None):
    key = {'partition_key_name': partition_key_value}
    
    if sort_key_value:
        key['sort_key_name'] = sort_key_value
        
    response = table.get_item(Key=key)
    return response.get('Item')

項目の更新（Update）

def update_item(table, partition_key_value, sort_key_value=None, update_expression=None, expression_values=None):
    key = {'partition_key_name': partition_key_value}
    
    if sort_key_value:
        key['sort_key_name'] = sort_key_value
        
    response = table.update_item(
        Key=key,
        UpdateExpression=update_expression,
        ExpressionAttributeValues=expression_values,
        ReturnValues="UPDATED_NEW"
    )
    return response

項目の削除（Delete）

def delete_item(table, partition_key_value, sort_key_value=None):
    key = {'partition_key_name': partition_key_value}
    
    if sort_key_value:
        key['sort_key_name'] = sort_key_value
        
    response = table.delete_item(Key=key)
    return response

クエリとスキャン

クエリ操作

パーティションキーを指定して効率的に検索：

def query_items(table, partition_key_value, sort_key_condition=None):
    from boto3.dynamodb.conditions import Key
    
    key_condition = Key('partition_key_name').eq(partition_key_value)
    
    if sort_key_condition:
        key_condition = key_condition & sort_key_condition
        
    response = table.query(KeyConditionExpression=key_condition)
    return response['Items']

スキャン操作

テーブル全体を検索（大規模テーブルでは非効率）：

def scan_items(table, filter_expression=None):
    params = {}
    
    if filter_expression:
        params['FilterExpression'] = filter_expression
        
    response = table.scan(**params)
    return response['Items']

4. ローカル開発環境の構築

DynamoDBをローカルで実行できるため、開発とテストが容易になります。

DynamoDB Localの設定

Docker Composeを使用したローカル環境構築：

# docker-compose.yml
version: '3'
services:
  dynamodb-local:
    image: amazon/dynamodb-local
    ports:
      - "8000:8000"
    command: "-jar DynamoDBLocal.jar -sharedDb -dbPath /home/dynamodblocal/data/"
    volumes:
      - "./dynamodb_data:/home/dynamodblocal/data"
    
  dynamodb-admin:
    image: aaronshaf/dynamodb-admin
    ports:
      - "8001:8001"
    environment:
      - DYNAMO_ENDPOINT=http://dynamodb-local:8000
    depends_on:
      - dynamodb-local

起動コマンド：

docker-compose up -d

ローカルDynamoDBへの接続

# ローカルDynamoDBへの接続設定
dynamodb = boto3.resource(
    'dynamodb',
    endpoint_url='http://localhost:8000',  # ローカルエンドポイント
    region_name='us-west-2',
    aws_access_key_id='dummy',  # ローカルでは任意の値
    aws_secret_access_key='dummy'
)

テーブル作成スクリプト

アプリケーション起動時にテーブルを自動作成：

def create_table_if_not_exists(dynamodb, table_name):
    # 既存のテーブルリストを取得
    existing_tables = [t.name for t in dynamodb.tables.all()]
    
    if table_name not in existing_tables:
        print(f"Creating table: {table_name}")
        
        # テーブル作成
        table = dynamodb.create_table(
            TableName=table_name,
            KeySchema=[
                {'AttributeName': 'id', 'KeyType': 'HASH'},  # パーティションキー
                {'AttributeName': 'timestamp', 'KeyType': 'RANGE'}  # ソートキー
            ],
            AttributeDefinitions=[
                {'AttributeName': 'id', 'AttributeType': 'S'},
                {'AttributeName': 'timestamp', 'AttributeType': 'S'}
            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 5,
                'WriteCapacityUnits': 5
            }
        )
        
        # テーブルが作成されるまで待機
        table.meta.client.get_waiter('table_exists').wait(TableName=table_name)
        print(f"Table created: {table_name}")
        return table
    else:
        print(f"Table already exists: {table_name}")
        return dynamodb.Table(table_name)

5. データモデリングのベストプラクティス

NoSQLデータモデリングの考え方

NoSQLデータモデリングはRDBとは根本的に異なるアプローチが必要です：

• クエリ駆動設計: 必要なクエリパターンを先に決めてからデータモデルを設計
• 非正規化: 読み取り効率化のためにデータを複製する
• コレクション思考: 関連データをまとめて格納
• 階層構造: JSONのネスト構造を活用

アクセスパターンの特定

データモデリングの最初のステップは、アプリケーションがどのようにデータにアクセスするかを明確にすることです：

1. 必要なクエリを全てリストアップする
2. 各クエリの頻度と重要度を評価する
3. 読み取りと書き込みの比率を分析する

単一テーブル設計

DynamoDBでは、複数のエンティティタイプを1つのテーブルに格納する設計が一般的です：

• 利点:

  • 結合操作が不要
  • トランザクション処理の簡素化
  • クエリの効率化

• 実装方法:

  • パーティションキーにエンティティタイプを含める
  • ソートキーを複合キーとして使用

# 単一テーブル設計例
PK                | SK                | その他の属性
------------------|-------------------|------------
USER#1001         | PROFILE          | name, email, ...
USER#1001         | ORDER#12345      | amount, date, ...
PRODUCT#5678      | INFO             | title, price, ...
ORDER#12345       | ITEM#5678        | quantity, ...

GSIとLSIの効果的な使用

インデックスを活用して多様なアクセスパターンをサポート：

• GSIのユースケース:

  • 別の属性でのクエリ
  • インバースインデックス（SKとPKを入れ替える）
  • スパースインデックス（条件付きインデックス）

• LSIのユースケース:

  • 同じパーティション内での別のソート順

複合ソートキーとクエリパターン

複合ソートキーは多様なクエリニーズに対応する強力な手法です：

# 階層型ソートキー例
USER#1001 | ORDER#2022-05-01#12345 | ...

このデータは以下のようなクエリに対応できます：

• 特定日付の注文：SK begins_with "ORDER#2022-05-01"
• 日付範囲の注文：SK between "ORDER#2022-05-01" and "ORDER#2022-05-31"
• 注文ID指定：SK = "ORDER#2022-05-01#12345"

6. リポジトリパターンの実装

アプリケーションコードとデータアクセスロジックを分離するためのリポジトリパターンを実装しましょう。

リポジトリクラスの基本構造

import boto3
import logging
from typing import Dict, List, Optional

class DynamoDBRepository:
    """DynamoDBリポジトリクラス - データアクセス層"""
    
    def __init__(self, 
                 table_name: str,
                 region_name: str = "us-west-2",
                 endpoint_url: Optional[str] = None):
        """初期化"""
        self.logger = logging.getLogger(__name__)
        self.table_name = table_name
        
        # DynamoDBリソースの初期化
        kwargs = {
            'region_name': region_name
        }
        
        # ローカル開発用エンドポイントの設定
        if endpoint_url:
            kwargs['endpoint_url'] = endpoint_url
            kwargs['aws_access_key_id'] = 'dummy'
            kwargs['aws_secret_access_key'] = 'dummy'
            
        self.dynamodb = boto3.resource('dynamodb', **kwargs)
        
        # テーブルを初期化
        self.table = self._create_table_if_not_exists()
    
    def _create_table_if_not_exists(self):
        """テーブルが存在しない場合は作成する"""
        # 実装は省略
        pass
        
    # 以下、具体的なCRUD操作を実装

インターフェースと実装の分離

コードの柔軟性とテスト容易性を高めるため、インターフェースを定義することをお勧めします：

from abc import ABC, abstractmethod
from typing import Dict, List, Optional

class RepositoryInterface(ABC):
    """リポジトリインターフェース"""
    
    @abstractmethod
    def save_item(self, item: Dict) -> Dict:
        """アイテムを保存する"""
        pass
    
    @abstractmethod
    def get_item(self, partition_key: str, sort_key: Optional[str] = None) -> Optional[Dict]:
        """アイテムを取得する"""
        pass
    
    @abstractmethod
    def query_items(self, partition_key: str) -> List[Dict]:
        """パーティションキーに基づいてアイテムを取得する"""
        pass
    
    # その他の必要なメソッド

テスト用モックリポジトリ

単体テスト用のモック実装：

class MockRepository(RepositoryInterface):
    """テスト用モックリポジトリ"""
    
    def __init__(self):
        self.items = {}  # インメモリストレージ
    
    def save_item(self, item: Dict) -> Dict:
        key = f"{item['pk']}#{item.get('sk', '')}"
        self.items[key] = item
        return item
    
    def get_item(self, partition_key: str, sort_key: Optional[str] = None) -> Optional[Dict]:
        key = f"{partition_key}#{sort_key or ''}"
        return self.items.get(key)
    
    def query_items(self, partition_key: str) -> List[Dict]:
        return [item for key, item in self.items.items() if key.startswith(f"{partition_key}#")]

7. FastAPIとの統合

DynamoDBリポジトリをFastAPIアプリケーションと統合する方法です。

依存性注入の設定

from fastapi import Depends, FastAPI
from typing import Annotated
from .repository import DynamoDBRepository, RepositoryInterface

app = FastAPI()

# リポジトリのインスタンスを取得する関数
def get_repository() -> RepositoryInterface:
    # 環境変数に基づいて本番またはローカルリポジトリを返す
    return DynamoDBRepository(
        table_name="YourTableName",
        endpoint_url="http://localhost:8000"  # ローカル開発時のみ
    )

# エンドポイントで依存性を使用
@app.get("/items/{item_id}")
async def read_item(
    item_id: str, 
    repo: Annotated[RepositoryInterface, Depends(get_repository)]
):
    item = repo.get_item(partition_key=item_id)
    if item is None:
        return {"error": "Item not found"}
    return item

サービス層の実装

ビジネスロジックとデータアクセスを分離：

from typing import Dict, List, Optional

class ItemService:
    """アイテム操作のビジネスロジック層"""
    
    def __init__(self, repository: RepositoryInterface):
        self.repository = repository
    
    def create_item(self, item_data: Dict) -> Dict:
        """アイテムを作成する（ビジネスロジックを適用）"""
        # 入力検証やビジネスルールの適用
        # ...
        
        # リポジトリを使用してデータを保存
        return self.repository.save_item(item_data)
    
    def get_item_details(self, item_id: str) -> Optional[Dict]:
        """アイテムの詳細情報を取得する"""
        item = self.repository.get_item(partition_key=f"ITEM#{item_id}")
        if not item:
            return None
            
        # 関連データの取得と結合
        related_items = self.repository.query_items(f"RELATED#{item_id}")
        
        # 結果の整形
        result = {
            "id": item_id,
            "details": item,
            "related": related_items
        }
        
        return result

完全なFastAPIエンドポイント例

from fastapi import FastAPI, HTTPException, Depends
from typing import Annotated, List

from .repository import RepositoryInterface, get_repository
from .service import ItemService
from .models import ItemCreate, ItemResponse

app = FastAPI()

# サービスのインスタンスを取得する関数
def get_service(
    repo: Annotated[RepositoryInterface, Depends(get_repository)]
) -> ItemService:
    return ItemService(repository=repo)

@app.post("/items/", response_model=ItemResponse)
async def create_item(
    item: ItemCreate,
    service: Annotated[ItemService, Depends(get_service)]
):
    try:
        result = service.create_item(item.dict())
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/items/{item_id}", response_model=ItemResponse)
async def read_item(
    item_id: str,
    service: Annotated[ItemService, Depends(get_service)]
):
    result = service.get_item_details(item_id)
    if result is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return result

8. ユニットテストの書き方

DynamoDBを使用するコードの効果的なテスト方法です。

モックリポジトリを使用したテスト

import unittest
from unittest.mock import MagicMock
from your_app.service import ItemService
from your_app.repository import RepositoryInterface

class TestItemService(unittest.TestCase):
    
    def setUp(self):
        # モックリポジトリを作成
        self.mock_repo = MagicMock(spec=RepositoryInterface)
        # テスト対象のサービスを初期化
        self.service = ItemService(repository=self.mock_repo)
        
    def test_get_item_details_found(self):
        # モックの動作を設定
        self.mock_repo.get_item.return_value = {"name": "Test Item", "price": 100}
        self.mock_repo.query_items.return_value = [{"id": "related1"}, {"id": "related2"}]
        
        # サービスメソッドを実行
        result = self.service.get_item_details("item123")
        
        # アサーション
        self.assertIsNotNone(result)
        self.assertEqual(result["id"], "item123")
        self.assertEqual(len(result["related"]), 2)
        
        # モックが正しく呼び出されたことを検証
        self.mock_repo.get_item.assert_called_once_with(partition_key="ITEM#item123")
        self.mock_repo.query_items.assert_called_once_with("RELATED#item123")
        
    def test_get_item_details_not_found(self):
        # アイテムが見つからない場合
        self.mock_repo.get_item.return_value = None
        
        # サービスメソッドを実行
        result = self.service.get_item_details("nonexistent")
        
        # アサーション
        self.assertIsNone(result)
        
        # モックが正しく呼び出されたことを検証
        self.mock_repo.get_item.assert_called_once()
        self.mock_repo.query_items.assert_not_called()

moto を使用した DynamoDB モック

AWS サービスをモックするための moto ライブラリを使用したテスト：

import boto3
import unittest
from moto import mock_dynamodb

from your_app.repository import DynamoDBRepository

class TestDynamoDBRepository(unittest.TestCase):
    
    @mock_dynamodb
    def setUp(self):
        """テスト用のDynamoDBテーブルを作成"""
        self.dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
        
        # テスト用テーブルの作成
        self.table_name = "TestTable"
        self.table = self.dynamodb.create_table(
            TableName=self.table_name,
            KeySchema=[
                {'AttributeName': 'pk', 'KeyType': 'HASH'},
                {'AttributeName': 'sk', 'KeyType': 'RANGE'}
            ],
            AttributeDefinitions=[
                {'AttributeName': 'pk', 'AttributeType': 'S'},
                {'AttributeName': 'sk', 'AttributeType': 'S'}
            ],
            ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
        )
        
        # リポジトリの初期化
        self.repository = DynamoDBRepository(
            table_name=self.table_name,
            region_name='us-west-2'
        )
    
    @mock_dynamodb
    def test_save_and_get_item(self):
        """アイテム保存と取得のテスト"""
        # テストデータ
        test_item = {
            'pk': 'USER#123',
            'sk': 'PROFILE',
            'name': 'Test User',
            'email': '[e-mail]'
        }
        
        # アイテムを保存
        self.repository.save_item(test_item)
        
        # アイテムを取得
        result = self.repository.get_item('USER#123', 'PROFILE')
        
        # 検証
        self.assertIsNotNone(result)
        self.assertEqual(result['name'], 'Test User')
        self.assertEqual(result['email'], '[e-mail]')

9. パフォーマンスチューニングとコスト最適化

DynamoDBを効率的に使用するための重要なヒントです。

プロビジョニングモードの選択

DynamoDBには2つの容量モードがあります：

• オンデマンドモード:

  • 事前にキャパシティを計画する必要がない
  • 使用した分だけ支払い
  • トラフィックの予測が難しい場合に最適

• プロビジョンドモード:

  • 事前に読み書き容量を設定
  • 予測可能なワークロードでコスト効率が高い
  • 自動スケーリングを設定可能

ホットパーティション問題の回避

データアクセスが特定のパーティションに集中すると性能が低下する「ホットパーティション問題」の対策：

• パーティションキーに十分な基数（カーディナリティ）を持たせる
• 合成キーを使用（例：user_123#2023-04-01）
• 書き込みシャーディングの実装（パーティションキーにランダムなサフィックスを追加）

バッチ操作の活用

複数のアイテムを一度に処理してAPI呼び出し回数を削減：

def batch_write_items(table, items):
    """複数アイテムを一括で書き込む"""
    with table.batch_writer() as batch:
        for item in items:
            batch.put_item(Item=item)

DAX (DynamoDB Accelerator) の検討

読み取り頻度の高いワークロードには、インメモリキャッシュのDAXを検討：

• マイクロ秒単位のレスポンスタイム
• アプリケーションコードの変更はほぼ不要
• 読み取りコストの削減

コスト最適化テクニック

• TTL (Time To Live) の活用：不要データの自動削除
• GSIの厳選: 必要最小限のインデックスのみを作成
• 項目サイズの最適化: 属性名の短縮、不要データの削除
• データ圧縮: 大きなテキストやバイナリデータの圧縮

10. 参考文献

• Amazon DynamoDB デベロッパーガイド
• AWS SDK for Python (Boto3) ドキュメント
• The DynamoDB Book by Alex DeBrie
• FastAPI 公式ドキュメント
• DynamoDB のデータモデリングのベストプラクティス
• Rick Houlihan の DynamoDB 単一テーブル設計に関する AWS re:Invent セッション
• AWS Database Blog - DynamoDB の高度なデザインパターン

---

まとめ

Python と DynamoDB を組み合わせることで、スケーラブルで高性能なバックエンドアプリケーションを構築できます。本記事で紹介したリポジトリパターン、データモデリング手法、テスト戦略を活用して、信頼性の高いアプリケーションを開発しましょう。

RDBとは異なる思考法が必要ですが、アクセスパターンを中心に設計することで、DynamoDBの真の力を引き出すことができます。特に大規模なトラフィックや柔軟なスキーマが求められるアプリケーションでは、DynamoDBは優れた選択肢となるでしょう。

実際の開発にお役立ていただければ幸いです。質問やフィードバックがあれば、コメント欄でお待ちしています！