[Python MongoDB] pymongo - CRUD 작업

소개

MongoDB는 강력한 NoSQL 데이터베이스이며, pymongo는 Python에서 MongoDB를 사용할 수 있게 해주는 공식 드라이버입니다. 이번 글에서는 pymongo를 사용하여 데이터를 다루는 다양한 방법을 자세히 알아보겠습니다.

MongoDB 연결 설정

from pymongo import MongoClient
from pymongo.collection import Collection
from pymongo.database import Database
from typing import Optional, Dict, List

class MongoDBManager:
    def __init__(self, uri: str, db_name: str):
        """
        MongoDB 연결 관리자

        Args:
            uri: MongoDB 연결 문자열
            db_name: 데이터베이스 이름

        Example:
            manager = MongoDBManager("mongodb://localhost:27017", "my_db")
        """
        self.client = MongoClient(uri)
        self.db: Database = self.client[db_name]

    def get_collection(self, collection_name: str) -> Collection:
        """특정 컬렉션 가져오기"""
        return self.db[collection_name]

    def close(self):
        """연결 종료"""
        self.client.close()

문서 삽입 (Insert)

단일 문서 삽입

from datetime import datetime

def insert_user(collection: Collection, user_data: Dict) -> str:
    """
    사용자 정보를 데이터베이스에 삽입

    Args:
        collection: MongoDB 컬렉션
        user_data: 삽입할 사용자 데이터

    Returns:
        생성된 문서의 ID

    Example:
        user = {
            "name": "John Doe",
            "email": "john@example.com",
            "created_at": datetime.now()
        }
        user_id = insert_user(db.users, user)
    """
    result = collection.insert_one(user_data)
    return str(result.inserted_id)

# 사용 예시
db = MongoDBManager("mongodb://localhost:27017", "my_db")
users = db.get_collection("users")

new_user = {
    "name": "John Doe",
    "email": "john@example.com",
    "age": 30,
    "created_at": datetime.now()
}

user_id = insert_user(users, new_user)
print(f"Created user with ID: {user_id}")

다중 문서 삽입

def insert_many_users(collection: Collection, users_data: List[Dict]) -> List[str]:
    """
    여러 사용자 정보를 한 번에 삽입

    Args:
        collection: MongoDB 컬렉션
        users_data: 삽입할 사용자 데이터 리스트

    Returns:
        생성된 문서들의 ID 리스트

    Example:
        users = [
            {"name": "John", "email": "john@example.com"},
            {"name": "Jane", "email": "jane@example.com"}
        ]
        user_ids = insert_many_users(db.users, users)
    """
    result = collection.insert_many(users_data)
    return [str(id) for id in result.inserted_ids]

문서 조회 (Find)

단일 문서 조회

from bson.objectid import ObjectId

def find_user_by_id(collection: Collection, user_id: str) -> Optional[Dict]:
    """
    ID로 사용자 조회

    Example:
        user = find_user_by_id(db.users, "507f1f77bcf86cd799439011")
    """
    return collection.find_one({"_id": ObjectId(user_id)})

def find_user_by_email(collection: Collection, email: str) -> Optional[Dict]:
    """
    이메일로 사용자 조회

    Example:
        user = find_user_by_email(db.users, "john@example.com")
    """
    return collection.find_one({"email": email})

다중 문서 조회

def find_users(
    collection: Collection,
    query: Dict = None,
    sort_by: str = None,
    limit: int = None
) -> List[Dict]:
    """
    조건에 맞는 사용자들 조회

    Example:
        # 30세 이상 사용자 조회
        users = find_users(db.users, {"age": {"$gte": 30}})

        # 이름으로 정렬하여 조회
        users = find_users(db.users, sort_by="name")
    """
    cursor = collection.find(query or {})

    if sort_by:
        cursor = cursor.sort(sort_by)

    if limit:
        cursor = cursor.limit(limit)

    return list(cursor)

문서 수정 (Update)

단일 문서 수정

def update_user(
    collection: Collection,
    user_id: str,
    update_data: Dict
) -> bool:
    """
    사용자 정보 수정

    Example:
        success = update_user(
            db.users,
            "507f1f77bcf86cd799439011",
            {"$set": {"age": 31}}
        )
    """
    result = collection.update_one(
        {"_id": ObjectId(user_id)},
        {"$set": update_data}
    )
    return result.modified_count > 0

# 증가/감소 연산
def increment_user_age(collection: Collection, user_id: str) -> bool:
    """
    사용자 나이 1 증가
    """
    result = collection.update_one(
        {"_id": ObjectId(user_id)},
        {"$inc": {"age": 1}}
    )
    return result.modified_count > 0

다중 문서 수정

def update_many_users(
    collection: Collection,
    query: Dict,
    update_data: Dict
) -> int:
    """
    조건에 맞는 여러 사용자 정보 수정

    Returns:
        수정된 문서 수

    Example:
        # 30세 이상 사용자의 상태를 'active'로 변경
        modified_count = update_many_users(
            db.users,
            {"age": {"$gte": 30}},
            {"$set": {"status": "active"}}
        )
    """
    result = collection.update_many(query, {"$set": update_data})
    return result.modified_count

문서 삭제 (Delete)

def delete_user(collection: Collection, user_id: str) -> bool:
    """
    사용자 삭제

    Example:
        success = delete_user(db.users, "507f1f77bcf86cd799439011")
    """
    result = collection.delete_one({"_id": ObjectId(user_id)})
    return result.deleted_count > 0

def delete_many_users(collection: Collection, query: Dict) -> int:
    """
    조건에 맞는 여러 사용자 삭제

    Example:
        # 비활성 사용자 모두 삭제
        deleted_count = delete_many_users(
            db.users,
            {"status": "inactive"}
        )
    """
    result = collection.delete_many(query)
    return result.deleted_count

고급 쿼리 작업

집계 (Aggregation)

def get_user_stats(collection: Collection) -> Dict:
    """
    사용자 통계 정보 조회

    Example:
        stats = get_user_stats(db.users)
        # {"total_users": 100, "avg_age": 35.5}
    """
    pipeline = [
        {
            "$group": {
                "_id": None,
                "total_users": {"$sum": 1},
                "avg_age": {"$avg": "$age"}
            }
        }
    ]

    result = list(collection.aggregate(pipeline))
    return result[0] if result else {}

인덱스 관리

def ensure_indexes(collection: Collection):
    """
    필요한 인덱스 생성
    """
    # 이메일 unique 인덱스
    collection.create_index("email", unique=True)
    # 나이 + 이름 복합 인덱스
    collection.create_index([("age", 1), ("name", 1)])

배치 작업 처리

from typing import List

def batch_process_users(
    collection: Collection,
    operations: List[Dict],
    batch_size: int = 1000
):
    """
    대량의 사용자 정보 일괄 처리

    Example:
        operations = [
            {
                "type": "update",
                "query": {"_id": ObjectId("...")},
                "update": {"$set": {"status": "active"}}
            },
            {
                "type": "insert",
                "document": {"name": "New User", "email": "new@example.com"}
            }
        ]
        batch_process_users(db.users, operations)
    """
    for i in range(0, len(operations), batch_size):
        batch = operations[i:i + batch_size]
        bulk_operations = []

        for op in batch:
            if op["type"] == "update":
                bulk_operations.append(
                    UpdateOne(op["query"], op["update"])
                )
            elif op["type"] == "insert":
                bulk_operations.append(
                    InsertOne(op["document"])
                )

        if bulk_operations:
            collection.bulk_write(bulk_operations, ordered=False)

예외 처리

from pymongo.errors import (
    DuplicateKeyError,
    ConnectionError,
    OperationFailure
)

def safe_insert_user(collection: Collection, user_data: Dict) -> Optional[str]:
    """
    예외 처리가 포함된 안전한 사용자 삽입
    """
    try:
        result = collection.insert_one(user_data)
        return str(result.inserted_id)
    except DuplicateKeyError:
        print("이미 존재하는 사용자입니다.")
    except ConnectionError:
        print("데이터베이스 연결 오류")
    except OperationFailure as e:
        print(f"작업 실패: {str(e)}")
    return None

마치며

pymongo를 사용하면 Python에서 MongoDB를 효율적으로 다룰 수 있습니다. 기본적인 CRUD 작업부터 복잡한 집계 작업까지, pymongo는 MongoDB의 모든 기능을 Python에서 사용할 수 있게 해줍니다. 대량의 데이터를 처리할 때는 배치 작업을 활용하고, 적절한 예외 처리를 통해 안정적인 애플리케이션을 구축할 수 있습니다.