본문으로 건너뛰기
Advertisement

MongoDB 연동

MongoDB는 문서 지향 NoSQL 데이터베이스입니다. 유연한 스키마와 수평 확장이 특징입니다.

설치​

pip install pymongo           # 동기
pip install motor             # 비동기 (asyncio 기반)
pip install "pymongo[srv]"    # MongoDB Atlas (DNS SRV) 연결 시

PyMongo — 동기 클라이언트​

from pymongo import MongoClient, ASCENDING, DESCENDING
from pymongo.collection import Collection
from bson import ObjectId
from datetime import datetime

# 연결
client = MongoClient("mongodb://localhost:27017/")
db = client["myapp"]
products_col: Collection = db["products"]

# Atlas 연결
# client = MongoClient("mongodb+srv://user:pass@cluster.mongodb.net/")

기본 CRUD​

# CREATE
result = products_col.insert_one({
    "name": "Python Book",
    "price": 35000,
    "stock": 100,
    "tags": ["python", "programming"],
    "category": {"id": 1, "name": "Books"},
    "created_at": datetime.utcnow(),
})
product_id = result.inserted_id  # ObjectId

# 여러 문서 삽입
result = products_col.insert_many([
    {"name": "FastAPI Guide", "price": 28000},
    {"name": "Django Pro", "price": 32000},
])


# READ
# 단일 조회
product = products_col.find_one({"_id": ObjectId(product_id)})
product = products_col.find_one({"name": "Python Book"})

# 복수 조회
all_products = list(products_col.find({"stock": {"$gt": 0}}))

# 필드 선택 (projection)
names_only = list(products_col.find({}, {"name": 1, "price": 1, "_id": 0}))

# 정렬 + 페이지네이션
page = list(
    products_col.find({"is_active": True})
    .sort("created_at", DESCENDING)
    .skip(20)
    .limit(20)
)

# 카운트
total = products_col.count_documents({"is_active": True})


# UPDATE
# 단일 수정
products_col.update_one(
    {"_id": ObjectId(product_id)},
    {"$set": {"price": 38000, "updated_at": datetime.utcnow()}}
)

# 숫자 증감
products_col.update_one(
    {"_id": ObjectId(product_id)},
    {"$inc": {"stock": -1}}
)

# 배열 연산
products_col.update_one(
    {"_id": ObjectId(product_id)},
    {"$push": {"tags": "bestseller"}}  # 배열에 추가
)
products_col.update_one(
    {"_id": ObjectId(product_id)},
    {"$pull": {"tags": "draft"}}       # 배열에서 제거
)

# 다중 수정
products_col.update_many(
    {"category.name": "Books"},
    {"$set": {"discount": 0.1}}
)


# DELETE
products_col.delete_one({"_id": ObjectId(product_id)})
products_col.delete_many({"is_active": False, "stock": 0})

쿼리 연산자​

# 비교
products_col.find({"price": {"$gte": 10000, "$lte": 50000}})
products_col.find({"stock": {"$gt": 0}})
products_col.find({"category.name": {"$ne": "Electronics"}})

# 논리
from pymongo import collection

# AND (기본)
products_col.find({"is_active": True, "stock": {"$gt": 0}})

# OR
products_col.find({"$or": [{"price": {"$lt": 10000}}, {"tags": "free"}]})

# NOT
products_col.find({"price": {"$not": {"$gt": 100000}}})

# IN / NIN
products_col.find({"tags": {"$in": ["python", "backend"]}})
products_col.find({"category.name": {"$nin": ["Deleted", "Hidden"]}})

# 정규식
import re
products_col.find({"name": {"$regex": re.compile("python", re.IGNORECASE)}})

# 배열 쿼리
products_col.find({"tags": "python"})          # 배열에 값 포함
products_col.find({"tags": {"$all": ["python", "api"]}})  # 모두 포함
products_col.find({"tags": {"$size": 3}})      # 배열 길이

집계 파이프라인​

# 카테고리별 통계
pipeline = [
    {"$match": {"is_active": True}},               # 필터
    {"$group": {
        "_id": "$category.name",                    # 그룹 기준
        "count": {"$sum": 1},                       # 개수
        "avg_price": {"$avg": "$price"},            # 평균
        "total_stock": {"$sum": "$stock"},          # 합계
        "max_price": {"$max": "$price"},            # 최대
    }},
    {"$sort": {"count": -1}},                       # 정렬
    {"$limit": 10},                                 # 상위 10개
]

results = list(products_col.aggregate(pipeline))


# 조인 ($lookup)
pipeline = [
    {"$lookup": {
        "from": "reviews",
        "localField": "_id",
        "foreignField": "product_id",
        "as": "reviews",
    }},
    {"$addFields": {
        "review_count": {"$size": "$reviews"},
        "avg_rating": {"$avg": "$reviews.rating"},
    }},
    {"$project": {"reviews": 0}},  # reviews 배열 제외
]


# 언와인드 (배열 → 개별 문서)
pipeline = [
    {"$unwind": "$tags"},          # tags 배열 펼치기
    {"$group": {"_id": "$tags", "count": {"$sum": 1}}},
    {"$sort": {"count": -1}},
]

인덱스 관리​

# 단일 인덱스
products_col.create_index("name")
products_col.create_index([("price", ASCENDING)])

# 복합 인덱스
products_col.create_index([("category.name", ASCENDING), ("is_active", ASCENDING)])

# 유니크 인덱스
products_col.create_index("sku", unique=True)

# TTL 인덱스 (자동 만료)
logs_col.create_index("created_at", expireAfterSeconds=86400)  # 24시간 후 삭제

# 텍스트 인덱스 (전문 검색)
products_col.create_index([("name", "text"), ("description", "text")])
results = list(products_col.find({"$text": {"$search": "python api"}}))

# 인덱스 목록 확인
print(list(products_col.list_indexes()))

Motor — 비동기 클라이언트​

import motor.motor_asyncio

client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017/")
db = client["myapp"]
products_col = db["products"]


async def get_product(product_id: str) -> dict | None:
    return await products_col.find_one({"_id": ObjectId(product_id)})


async def get_products(skip: int = 0, limit: int = 20) -> list[dict]:
    cursor = products_col.find({"is_active": True}).skip(skip).limit(limit)
    return await cursor.to_list(length=limit)


async def create_product(data: dict) -> str:
    data["created_at"] = datetime.utcnow()
    result = await products_col.insert_one(data)
    return str(result.inserted_id)


# 비동기 집계
async def get_category_stats() -> list[dict]:
    pipeline = [
        {"$group": {"_id": "$category.name", "count": {"$sum": 1}}},
        {"$sort": {"count": -1}},
    ]
    cursor = products_col.aggregate(pipeline)
    return await cursor.to_list(length=None)

정리​

개념설명
Collection테이블에 해당하는 문서 컨테이너
Document행에 해당하는 JSON 문서
$set, $inc업데이트 연산자
$match, $group집계 파이프라인 스테이지
$lookup컬렉션 조인
create_index()성능 최적화 인덱스
Motorasyncio 기반 비동기 드라이버

MongoDB는 비정형 데이터, 빠른 개발, 수평 확장이 필요한 서비스에 적합합니다.

Advertisement
Advertisement