Как работал с MongoDB?

Как я работал с MongoDB

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

Основные операции и примеры

from pymongo import MongoClient
from pymongo.errors import DuplicateKeyError
import json
from bson.objectid import ObjectId

class MongoDBManager:
    def __init__(self, connection_string):
        self.client = MongoClient(connection_string)
        self.db = self.client['my_database']
        self.users = self.db['users']
        self.posts = self.db['posts']
    
    # CRUD операции
    def create_user(self, name, email, age):
        """Создание документа"""
        result = self.users.insert_one({
            'name': name,
            'email': email,
            'age': age,
            'created_at': datetime.now(UTC)
        })
        return result.inserted_id
    
    def read_user(self, user_id):
        """Чтение по ID"""
        return self.users.find_one({'_id': ObjectId(user_id)})
    
    def find_by_email(self, email):
        """Поиск с условием"""
        return self.users.find_one({'email': email})
    
    def find_all_active_users(self):
        """Получить с фильтром"""
        return list(self.users.find({'status': 'active'}))
    
    def update_user(self, user_id, update_data):
        """Обновление документа"""
        result = self.users.update_one(
            {'_id': ObjectId(user_id)},
            {'$set': update_data}
        )
        return result.modified_count
    
    def delete_user(self, user_id):
        """Удаление документа"""
        result = self.users.delete_one({'_id': ObjectId(user_id)})
        return result.deleted_count

Операторы MongoDB и продвинутые запросы

def advanced_queries(self):
    # Арифметические операции
    self.users.update_one(
        {'_id': ObjectId('...')},
        {'$inc': {'age': 1}}  # инкремент
    )
    
    # Массивы
    self.users.update_one(
        {'_id': ObjectId('...')},
        {'$push': {'tags': 'python'}}  # добавить в массив
    )
    
    # Получить только нужные поля
    self.users.find(
        {},
        {'name': 1, 'email': 1, '_id': 0}  # projection
    )
    
    # Условия
    users_over_30 = self.users.find({
        'age': {'$gt': 30},  # greater than
        'status': {'$in': ['active', 'premium']}
    })
    
    # Сортировка и лимит
    self.users.find({}).sort('age', -1).limit(10)
    
    # Агрегация (pipeline)
    self.users.aggregate([
        {'$match': {'age': {'$gt': 25}}},
        {'$group': {
            '_id': '$status',
            'count': {'$sum': 1},
            'avg_age': {'$avg': '$age'}
        }},
        {'$sort': {'count': -1}}
    ])

Индексирование для оптимизации

def setup_indexes(self):
    # Уникальный индекс на email
    self.users.create_index('email', unique=True)
    
    # Составной индекс
    self.users.create_index([('status', 1), ('created_at', -1)])
    
    # Текстовый индекс для полнотекстового поиска
    self.posts.create_index([('title', 'text'), ('content', 'text')])
    
    # Индекс с опциями
    self.users.create_index(
        'email',
        unique=True,
        sparse=True  # индекс только для документов с полем
    )

Полнотекстовый поиск

def search_posts(self, query):
    """Полнотекстовый поиск"""
    return list(self.posts.find(
        {'$text': {'$search': query}},
        {'score': {'$meta': 'textScore'}}
    ).sort([('score', {'$meta': 'textScore'})]).limit(20))

def autocomplete_search(self, prefix):
    """Автодополнение"""
    return list(self.posts.find(
        {'title': {'$regex': f'^{prefix}', '$options': 'i'}},
        {'title': 1}
    ).limit(10))

Транзакции и валидация

def transfer_credits(self, from_user_id, to_user_id, amount):
    """Транзакция"""
    with self.client.start_session() as session:
        with session.start_transaction():
            self.users.update_one(
                {'_id': ObjectId(from_user_id)},
                {'$inc': {'credits': -amount}},
                session=session
            )
            self.users.update_one(
                {'_id': ObjectId(to_user_id)},
                {'$inc': {'credits': amount}},
                session=session
            )

# Валидация схемы
def setup_schema_validation(self):
    self.db.create_collection('users', validator={
        '$jsonSchema': {
            'bsonType': 'object',
            'required': ['email', 'name'],
            'properties': {
                'email': {'bsonType': 'string', 'pattern': '^[\\w-\\.]+@[\\w-]+\\.[a-z]+$'},
                'name': {'bsonType': 'string'},
                'age': {'bsonType': 'int', 'minimum': 0, 'maximum': 150}
            }
        }
    })

Типичные вызовы в production

• Индексирование — без индексов на часто используемые поля производительность падает в 100+ раз
• Batch операции — использование insert_many() вместо цикла с insert_one()
• Connection pooling — правильная настройка пула соединений
• Мониторинг — отслеживание медленных запросов с explain()

Когда я выбираю MongoDB вместо SQL

✅ Когда схема данных не строгая и может меняться ✅ Для документов с вложенной структурой ✅ Когда нужна горизонтальная масштабируемость (sharding) ✅ Для быстрого прототипирования

❌ Когда нужны сложные JOIN'ы (есть $lookup, но он не оптимален) ❌ Когда требуется ACID гарантия для многих операций ❌ Для денег и финансовых операций

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