MongoDB Atlas 接続と操作 完全ガイド

目次

• プロジェクト概要
• 環境設定
• コアクラス解説
• データベース操作
• テスト例
• ベストプラクティス

プロジェクト概要

本稿では、PythonベースのMongoDB Atlas接続マネージャーを介紹します。このツールは接続管理、CRUD操作、インデックス管理、集計クエリを含む完全なデータベース操作機能を提供します。

主な機能

• 自動接続管理
• 完全なCRUD操作
• インデックスの作成与管理
• 集計パイプライン対応
• エラー処理とログ記録
• 型安全なレスポンスオブジェクト

環境設定

1. 依存パッケージのインストール

pip install pymongo dnspython python-dotenv pydantic

2. 環境変数の設定

.envファイルを作成します：

MONGODB_URI=mongodb+srv://username:password@cluster.mongodb.net/?retryWrites=true&w=majority
LOG_LEVEL=INFO

3. プロジェクト構成

mongo_atlas_mcp/
├── __init__.py
├── database.py      # コアデータベースマネージャー
└── models.py        # データモデル定義

コアクラス解説

MongoAtlasManagerクラス

本システムのコアクラスであり、MongoDB Atlasの接続管理とすべてのデータベース操作を担当します。

初期化と接続

class MongoAtlasManager:
    def __init__(self):
        """MongoDB Atlasマネージャーの初期化"""
        self.client: Optional[MongoClient] = None
        self._connect()
    
    def _connect(self) -> None:
        """MongoDB Atlasへの接続"""
        try:
            mongodb_uri = os.getenv('MONGODB_URI')
            if not mongodb_uri:
                raise ValueError("MONGODB_URI環境変数が未設定です")
            
            self.client = MongoClient(mongodb_uri)
            # 接続テスト
            self.client.admin.command('ping')
            logger.info("MongoDB Atlasへの接続に成功しました")
            
        except Exception as e:
            logger.error(f"MongoDB Atlasへの接続に失敗しました: {str(e)}")
            raise

接続管理メソッド

def get_database(self, database_name: str) -> Database:
    """データベースオブジェクトの取得"""
    if not self.client:
        raise ConnectionError("MongoDBクライアントが接続されていません")
    return self.client[database_name]

def get_collection(self, database_name: str, collection_name: str) -> Collection:
    """コレクションオブジェクトの取得"""
    database = self.get_database(database_name)
    return database[collection_name]

データベース操作

1. データベース管理

全データベースの一覧表示

def list_databases(self) -> MongoResponse:
    """全データベースの一覧表示"""
    try:
        databases = []
        for db_name in self.client.list_database_names():
            database_info = DatabaseInfo(
                name=db_name,
                size_on_disk=0,  # 権限問題を回避
                empty=False
            )
            databases.append(database_info.model_dump())
        
        return MongoResponse(
            success=True,
            data=databases,
            count=len(databases),
            message="データベースの一覧表示に成功しました"
        )
    except Exception as e:
        logger.error(f"データベースの一覧表示に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=str(e),
            message="データベースの一覧表示に失敗しました"
        )

コレクションの一覧表示

def list_collections(self, database_name: str) -> MongoResponse:
    """指定データベースの全コレクションの一覧表示"""
    try:
        database = self.get_database(database_name)
        collections = []
        
        for collection_name in database.list_collection_names():
            collection_info = CollectionInfo(
                name=collection_name,
                count=0,  # 権限問題を回避
                size=0,
                avg_obj_size=0
            )
            collections.append(collection_info.model_dump())
        
        return MongoResponse(
            success=True,
            data=collections,
            count=len(collections),
            message="コレクションの一覧表示に成功しました"
        )
    except Exception as e:
        logger.error(f"コレクションの一覧表示に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=str(e),
            message="コレクションの一覧表示に失敗しました"
        )

2. ドキュメント操作

ドキュメントの検索

def find_documents(self, database_name: str, collection_name: str, 
                  filter_dict: Dict[str, Any] = None, 
                  projection: Dict[str, Any] = None,
                  sort: List[tuple] = None,
                  limit: int = None,
                  skip: int = 0) -> MongoResponse:
    """ドキュメントの検索"""
    try:
        collection = self.get_collection(database_name, collection_name)
        
        cursor = collection.find(
            filter=filter_dict or {},
            projection=projection
        )
        
        if sort:
            cursor = cursor.sort(sort)
        if skip:
            cursor = cursor.skip(skip)
        if limit:
            cursor = cursor.limit(limit)
        
        documents = list(cursor)
        
        # ドキュメントのシリアライズ、ObjectIdを文字列に変換
        serialized_documents = []
        for doc in documents:
            serialized_doc = {}
            for key, value in doc.items():
                if key == '_id':
                    serialized_doc[key] = str(value)
                else:
                    serialized_doc[key] = value
            serialized_documents.append(serialized_doc)
        
        return MongoResponse(
            success=True,
            data=serialized_documents,
            count=len(serialized_documents)
        )
    except PyMongoError as e:
        logger.error(f"ドキュメントの検索に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=f"ドキュメントの検索に失敗しました: {str(e)}"
        )

ドキュメントの挿入

def insert_document(self, database_name: str, collection_name: str, 
                   document: Dict[str, Any]) -> MongoResponse:
    """ドキュメントの挿入"""
    try:
        collection = self.get_collection(database_name, collection_name)
        result = collection.insert_one(document)
        
        return MongoResponse(
            success=True,
            data={"inserted_id": str(result.inserted_id)},
            count=1
        )
    except PyMongoError as e:
        logger.error(f"ドキュメントの挿入に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=f"ドキュメントの挿入に失敗しました: {str(e)}"
        )

ドキュメントの更新

def update_document(self, database_name: str, collection_name: str,
                   filter_dict: Dict[str, Any], update_dict: Dict[str, Any],
                   upsert: bool = False, multi: bool = False) -> MongoResponse:
    """ドキュメントの更新"""
    try:
        collection = self.get_collection(database_name, collection_name)
        
        if multi:
            result = collection.update_many(
                filter_dict, update_dict, upsert=upsert
            )
        else:
            result = collection.update_one(
                filter_dict, update_dict, upsert=upsert
            )
        
        return MongoResponse(
            success=True,
            data={
                "matched_count": result.matched_count,
                "modified_count": result.modified_count,
                "upserted_id": str(result.upserted_id) if result.upserted_id else None
            },
            count=result.modified_count
        )
    except PyMongoError as e:
        logger.error(f"ドキュメントの更新に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=f"ドキュメントの更新に失敗しました: {str(e)}"
        )

ドキュメントの削除

def delete_document(self, database_name: str, collection_name: str,
                   filter_dict: Dict[str, Any], multi: bool = False) -> MongoResponse:
    """ドキュメントの削除"""
    try:
        collection = self.get_collection(database_name, collection_name)
        
        if multi:
            result = collection.delete_many(filter_dict)
        else:
            result = collection.delete_one(filter_dict)
        
        return MongoResponse(
            success=True,
            data={"deleted_count": result.deleted_count},
            count=result.deleted_count
        )
    except PyMongoError as e:
        logger.error(f"ドキュメントの削除に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=f"ドキュメントの削除に失敗しました: {str(e)}"
        )

3. 集計操作

集計パイプラインの実行

def aggregate(self, database_name: str, collection_name: str,
              pipeline: List[Dict[str, Any]]) -> MongoResponse:
    """集計パイプラインの実行"""
    try:
        collection = self.get_collection(database_name, collection_name)
        cursor = collection.aggregate(pipeline)
        results = list(cursor)
        
        return MongoResponse(
            success=True,
            data=results,
            count=len(results)
        )
    except PyMongoError as e:
        logger.error(f"集計パイプラインの実行に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=f"集計パイプラインの実行に失敗しました: {str(e)}"
        )

4. インデックス管理

インデックスの作成

def create_index(self, database_name: str, collection_name: str,
                 keys: List[tuple], name: str = None,
                 unique: bool = False, sparse: bool = False,
                 background: bool = True) -> MongoResponse:
    """インデックスの作成"""
    try:
        collection = self.get_collection(database_name, collection_name)
        
        index_options = {
            "unique": unique,
            "sparse": sparse,
            "background": background
        }
        
        if name:
            index_options["name"] = name
        
        result = collection.create_index(keys, **index_options)
        
        return MongoResponse(
            success=True,
            data={"index_name": result},
            count=1
        )
    except PyMongoError as e:
        logger.error(f"インデックスの作成に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=f"インデックスの作成に失敗しました: {str(e)}"
        )

インデックスの一覧表示

def list_indexes(self, database_name: str, collection_name: str) -> MongoResponse:
    """コレクションの全インデックスを表示"""
    try:
        collection = self.get_collection(database_name, collection_name)
        indexes = []
        
        for index_info in collection.list_indexes():
            index_data = IndexInfo(
                name=index_info["name"],
                key=list(index_info["key"]),
                unique=index_info.get("unique", False),
                sparse=index_info.get("sparse", False),
                background=index_info.get("background", True)
            )
            indexes.append(index_data.model_dump())
        
        return MongoResponse(
            success=True,
            data=indexes,
            count=len(indexes)
        )
    except PyMongoError as e:
        logger.error(f"インデックスの一覧表示に失敗しました: {str(e)}")
        return MongoResponse(
            success=False,
            error=f"インデックスの一覧表示に失敗しました: {str(e)}"
        )

テスト例

1. 基本接続テスト

from mongo_atlas_mcp.database import MongoAtlasManager

# マネージャーインスタンスの作成
manager = MongoAtlasManager()

# 接続テスト
print("✅ 接続成功")

2. データベース操作テスト

# 全データベースの一覧表示
result = manager.list_databases()
print(f"データベース一覧: {result.data}")

# 指定データベースのコレクション一覧表示
result = manager.list_collections("test_db")
print(f"コレクション一覧: {result.data}")

3. ドキュメント操作テスト

# ドキュメントの挿入
document = {
    "name": "田中太郎",
    "age": 30,
    "email": "tanaka@example.com",
    "city": "東京",
    "hobbies": ["読書", "水泳", "コーディング"],
    "created_at": "2024-01-15T10:30:00Z",
    "is_active": True,
    "score": 87.5
}

result = manager.insert_document("test_db", "users", document)
print(f"挿入結果: {result.data}")

# ドキュメントの検索
result = manager.find_documents("test_db", "users")
print(f"検索結果: {result.data}")

# ドキュメントの更新
filter_dict = {"name": "田中太郎"}
update_dict = {"$set": {"age": 31, "score": 88.0}}
result = manager.update_document("test_db", "users", filter_dict, update_dict)
print(f"更新結果: {result.data}")

# ドキュメントの削除
result = manager.delete_document("test_db", "users", {"name": "田中太郎"})
print(f"削除結果: {result.data}")

4. 集計操作テスト

# 集計パイプライン例
pipeline = [
    {"$match": {"age": {"$gte": 25}}},
    {"$group": {"_id": "$city", "count": {"$sum": 1}, "avg_age": {"$avg": "$age"}}},
    {"$sort": {"count": -1}}
]

result = manager.aggregate("test_db", "users", pipeline)
print(f"集計結果: {result.data}")

5. インデックス操作テスト

# インデックスの作成
keys = [("email", 1)]
result = manager.create_index("test_db", "users", keys, name="email_index", unique=True)
print(f"インデックス作成結果: {result.data}")

# インデックスの一覧表示
result = manager.list_indexes("test_db", "users")
print(f"インデックス一覧: {result.data}")

テスト出力例

データベース一覧出力

{
  "success": true,
  "data": [
    {
      "name": "sample_mflix",
      "size_on_disk": 0,
      "empty": false
    },
    {
      "name": "test_db",
      "size_on_disk": 0,
      "empty": false
    },
    {
      "name": "admin",
      "size_on_disk": 0,
      "empty": false
    },
    {
      "name": "local",
      "size_on_disk": 0,
      "empty": false
    }
  ],
  "count": 4,
  "message": "データベースの一覧表示に成功しました"
}

コレクション一覧出力

{
  "success": true,
  "data": [
    {
      "name": "users",
      "count": 0,
      "size": 0,
      "avg_obj_size": 0
    },
    {
      "name": "products",
      "count": 0,
      "size": 0,
      "avg_obj_size": 0
    }
  ],
  "count": 2,
  "message": "コレクションの一覧表示に成功しました"
}

ドキュメント挿入出力

{
  "success": true,
  "data": {
    "inserted_id": "6891e7622b3504a53dbe9e71"
  },
  "count": 1
}

ドキュメント検索出力

{
  "success": true,
  "data": [
    {
      "_id": "6891e7622b3504a53dbe9e71",
      "name": "田中太郎",
      "age": 30,
      "email": "tanaka@example.com",
      "city": "東京",
      "hobbies": ["読書", "水泳", "コーディング"],
      "created_at": "2024-01-15T10:30:00Z",
      "is_active": true,
      "score": 87.5
    }
  ],
  "count": 1
}

ベストプラクティス

1. 接続管理

• 環境変数で接続文字列を管理する
• 接続プール管理を実装する
• 接続再試行メカニズムを追加する

2. エラー処理

• すべてのデータベース操作をtry-catchでラップする
• 詳細なエラーログを記録する
• 統一されたエラーレスポンスFormatを返す

3. パフォーマンス最適化

• インデックスを使用してクエリパフォーマンスを向上させる
• Projectionを効果的に使用してデータ転送を削減する
• ページネーションを実装してメモリオーバーフローを防ぐ

4. セキュリティ

• 環境変数に機密情報を保存する
• ユーザー権限検証を実装する
• SQLインジェクションを防ぐ（MongoDBは天然で対応）

5. 保守性

• 型注釈を使用してコードの可読性を向上させる
• 統一されたレスポンスFormatを実装する
• 詳細なドキュメントコメントを追加する

まとめ

本MongoDB Atlas接続マネージャーは、以下を含む完全なデータベース操作機能を提供します：

• 自動接続管理
• 完全なCRUD操作
• 集計クエリ対応
• インデックス管理機能
• 統一されたエラー処理
• 型安全なレスポンス

適切な设计与実装により、PythonプロジェクトでMongoDB Atlasを轻松に統合し、高效なデータ存储と検索功能を実現できます。