Приведи пример использования MongoDB

Пример использования MongoDB

MongoDB — это NoSQL база данных, основанная на документах. Вот практический пример её использования.

Когда использовать MongoDB?

MongoDB хороша для:

• Гибкая схема — структура документов может отличаться
• Вложенные данные — эффективно хранит nested структуры (JSON-like)
• Быстрое прототипирование — не нужно создавать миграции
• Масштабируемость — sharding встроен в архитектуру

MongoDB плохо подходит для:

• ACID транзакции (в современных версиях есть, но сложно)
• Очень связанные данные (тогда лучше SQL)
• Простые CRUD операции (SQL быстрее)

Пример 1: Блог с комментариями

from pymongo import MongoClient
from datetime import datetime

# Подключение
client = MongoClient('mongodb://localhost:27017/')
db = client['blog_db']
posts_collection = db['posts']
users_collection = db['users']

# Пользователь
user = {
    "_id": 1,
    "username": "alice",
    "email": "alice@example.com",
    "created_at": datetime.now(),
    "profile": {
        "bio": "I love Python",
        "avatar_url": "https://...",
        "followers_count": 100
    }
}
users_collection.insert_one(user)

# Блог-пост с вложенными комментариями
post = {
    "_id": 1,
    "title": "MongoDB для Python разработчиков",
    "author_id": 1,
    "content": "MongoDB это документ-ориентированная база...",
    "created_at": datetime.now(),
    "tags": ["mongodb", "python", "database"],
    "views_count": 1500,
    "comments": [
        {
            "user_id": 2,
            "username": "bob",
            "text": "Отличная статья!",
            "likes": 10,
            "created_at": datetime.now()
        },
        {
            "user_id": 3,
            "username": "charlie",
            "text": "А как с производительностью?",
            "likes": 5,
            "created_at": datetime.now()
        }
    ]
}
posts_collection.insert_one(post)

Заметь: комментарии хранятся внутри документа поста, а не в отдельной таблице (как в SQL). Это удобно для этого сценария.

Пример 2: CRUD операции

# CREATE — уже сделали выше

# READ — получить пост
post = posts_collection.find_one({"_id": 1})
print(post["title"])  # MongoDB для Python разработчиков

# READ — найти все посты пользователя 1
my_posts = posts_collection.find({"author_id": 1})
for post in my_posts:
    print(post["title"])

# READ с фильтром — посты с тегом "python"
python_posts = posts_collection.find({"tags": "python"})
for post in python_posts:
    print(post["title"])

# UPDATE — добавить лайк к посту
posts_collection.update_one(
    {"_id": 1},
    {"$inc": {"views_count": 1}}  # Увеличить на 1
)

# UPDATE — добавить комментарий к посту
posts_collection.update_one(
    {"_id": 1},
    {
        "$push": {
            "comments": {
                "user_id": 4,
                "username": "diana",
                "text": "Спасибо!",
                "likes": 0,
                "created_at": datetime.now()
            }
        }
    }
)

# DELETE — удалить пост
posts_collection.delete_one({"_id": 1})

# DELETE — удалить все посты пользователя
posts_collection.delete_many({"author_id": 1})

Пример 3: Сложные запросы (Aggregation Pipeline)

# Найти топ 5 авторов по количеству комментариев на их постах
pipeline = [
    # Этап 1: распаковать комментарии (flatten)
    {
        "$unwind": "$comments"
    },
    # Этап 2: группировать по автору
    {
        "$group": {
            "_id": "$author_id",
            "total_comments": {"$sum": 1},
            "posts_count": {"$first": "$title"}
        }
    },
    # Этап 3: сортировать по количеству комментариев
    {
        "$sort": {"total_comments": -1}
    },
    # Этап 4: ограничить топ 5
    {
        "$limit": 5
    }
]

top_authors = posts_collection.aggregate(pipeline)
for author in top_authors:
    print(f"Author {author['_id']}: {author['total_comments']} комментариев")

Пример 4: Интеграция с FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
from bson import ObjectId

app = FastAPI()

# Pydantic модель для валидации
class CommentCreate(BaseModel):
    user_id: int
    username: str
    text: str

class PostCreate(BaseModel):
    title: str
    content: str
    author_id: int
    tags: List[str]

# Создать пост
@app.post("/posts")
def create_post(post: PostCreate):
    post_data = post.dict()
    post_data["created_at"] = datetime.now()
    post_data["comments"] = []
    post_data["views_count"] = 0
    
    result = posts_collection.insert_one(post_data)
    return {"id": str(result.inserted_id), "message": "Post created"}

# Получить пост
@app.get("/posts/{post_id}")
def get_post(post_id: str):
    try:
        post = posts_collection.find_one({"_id": ObjectId(post_id)})
        if not post:
            raise HTTPException(status_code=404, detail="Post not found")
        post["_id"] = str(post["_id"])
        return post
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

# Добавить комментарий
@app.post("/posts/{post_id}/comments")
def add_comment(post_id: str, comment: CommentCreate):
    try:
        result = posts_collection.update_one(
            {"_id": ObjectId(post_id)},
            {
                "$push": {
                    "comments": {
                        **comment.dict(),
                        "likes": 0,
                        "created_at": datetime.now()
                    }
                }
            }
        )
        if result.matched_count == 0:
            raise HTTPException(status_code=404, detail="Post not found")
        return {"message": "Comment added"}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

# Получить посты пользователя
@app.get("/users/{user_id}/posts")
def get_user_posts(user_id: int):
    posts = list(posts_collection.find({"author_id": user_id}))
    for post in posts:
        post["_id"] = str(post["_id"])
    return posts

Пример 5: Индексы для производительности

# Создать индекс на поле author_id
posts_collection.create_index("author_id")

# Создать индекс на tags (для поиска по тегам)
posts_collection.create_index("tags")

# Создать составной индекс
posts_collection.create_index([("author_id", 1), ("created_at", -1)])

# Текстовый индекс для поиска по содержимому
posts_collection.create_index([("title", "text"), ("content", "text")])

# Использование текстового индекса
results = posts_collection.find(
    {"$text": {"$search": "python mongodb"}}
)

Пример 6: Транзакции (новые версии MongoDB)

from pymongo.errors import OperationFailure

# Простая транзакция
try:
    with client.start_session() as session:
        with session.start_transaction():
            # Уменьшить баланс пользователя A
            users_collection.update_one(
                {"_id": 1},
                {"$inc": {"balance": -100}},
                session=session
            )
            # Увеличить баланс пользователя B
            users_collection.update_one(
                {"_id": 2},
                {"$inc": {"balance": 100}},
                session=session
            )
        # Если всё OK — commit, если ошибка — rollback
except OperationFailure:
    print("Transaction failed")

Пример 7: Работа со временем жизни документов (TTL)

# Хранить токены восстановления пароля, которые протухают через 1 час
reset_tokens_collection = db['reset_tokens']

# Создать TTL индекс (документы удалятся автоматически)
reset_tokens_collection.create_index("created_at", expireAfterSeconds=3600)

# Сохранить токен
reset_tokens_collection.insert_one({
    "user_id": 1,
    "token": "abc123def456",
    "created_at": datetime.now()
})

# Через 1 час этот документ автоматически удалится

Когда выбрать MongoDB vs SQL?

Вывод: MongoDB отличный выбор для приложений с гибкой структурой данных, вложенными объектами и высокой нагрузкой на запись.