Conexión a MongoDB

MongoDB es una base de datos NoSQL orientada a documentos que almacena datos en formato similar a JSON (BSON). Para interactuar con MongoDB desde Python, utilizaremos PyMongo, la biblioteca oficial que nos permite conectarnos y realizar operaciones con bases de datos MongoDB.

Antes de comenzar a trabajar con MongoDB, necesitamos establecer una conexión adecuada. Esta conexión será el punto de entrada para todas las operaciones que realizaremos posteriormente con la base de datos.

Instalación de PyMongo

Lo primero que necesitamos es instalar la biblioteca PyMongo. Podemos hacerlo fácilmente utilizando pip:

pip install pymongo

Para proyectos que requieren versiones específicas o entornos aislados, es recomendable usar un entorno virtual:

python -m venv venv
source venv/bin/activate  # En Windows: venv\Scripts\activate
pip install pymongo

Estableciendo una conexión básica

La forma más simple de conectarse a MongoDB es utilizando el cliente de PyMongo:

from pymongo import MongoClient

# Conexión a MongoDB local (por defecto en puerto 27017)
client = MongoClient('localhost', 27017)

# Alternativa usando URI de conexión
client = MongoClient('mongodb://localhost:27017/')

El objeto client es nuestra puerta de entrada a MongoDB y nos permitirá acceder a todas las bases de datos y colecciones disponibles en el servidor.

Conexión con autenticación

En entornos de producción, normalmente necesitaremos proporcionar credenciales de autenticación:

# Conexión con autenticación básica
client = MongoClient('mongodb://usuario:contraseña@localhost:27017/')

# Método alternativo especificando parámetros
client = MongoClient(
    host='localhost',
    port=27017,
    username='usuario',
    password='contraseña',
    authSource='admin'  # Base de datos donde se almacenan las credenciales
)

El parámetro authSource especifica la base de datos donde MongoDB verifica las credenciales del usuario, generalmente es 'admin'.

Conexión a MongoDB Atlas

MongoDB Atlas es el servicio de base de datos en la nube ofrecido por MongoDB. Para conectarnos a una instancia de MongoDB Atlas:

# Conexión a MongoDB Atlas
connection_string = "mongodb+srv://usuario:contraseña@cluster0.mongodb.net/mibasededatos"
client = MongoClient(connection_string)

La cadena de conexión para Atlas sigue el formato mongodb+srv://, que incluye la resolución de DNS para encontrar los servidores del clúster.

Configuración de opciones de conexión

PyMongo permite configurar múltiples opciones para personalizar el comportamiento de la conexión:

# Conexión con opciones avanzadas
client = MongoClient(
    'mongodb://localhost:27017/',
    maxPoolSize=50,                # Tamaño máximo del pool de conexiones
    connectTimeoutMS=5000,         # Tiempo máximo para establecer conexión (ms)
    socketTimeoutMS=30000,         # Tiempo máximo para operaciones de socket (ms)
    serverSelectionTimeoutMS=5000, # Tiempo máximo para seleccionar servidor (ms)
    retryWrites=True               # Reintentar operaciones de escritura fallidas
)

Estas opciones son especialmente útiles en entornos de producción donde necesitamos controlar aspectos como el rendimiento y la resiliencia.

Accediendo a bases de datos y colecciones

Una vez establecida la conexión, podemos acceder a las bases de datos y colecciones:

# Acceder a una base de datos
db = client['mi_base_datos']  # Alternativa: db = client.mi_base_datos

# Acceder a una colección
coleccion = db['mi_coleccion']  # Alternativa: coleccion = db.mi_coleccion

En MongoDB, las bases de datos y colecciones se crean automáticamente cuando insertamos documentos, por lo que no es necesario crearlas explícitamente.

Verificación de conexión

Es una buena práctica verificar que la conexión se ha establecido correctamente:

try:
    # Intenta una operación simple para verificar la conexión
    client.admin.command('ping')
    print("Conexión establecida correctamente")
except Exception as e:
    print(f"Error de conexión: {e}")

Uso de contextos para gestionar conexiones

Python proporciona el gestor de contexto (with) que podemos utilizar para manejar conexiones de manera más segura:

from pymongo import MongoClient
from contextlib import contextmanager

@contextmanager
def get_mongo_client(uri='mongodb://localhost:27017/'):
    """Crea un cliente MongoDB y lo cierra automáticamente al finalizar."""
    client = MongoClient(uri)
    try:
        yield client
    finally:
        client.close()

# Uso del gestor de contexto
with get_mongo_client() as client:
    db = client.mi_base_datos
    # Realizar operaciones con la base de datos
    resultado = db.mi_coleccion.find_one()
    print(resultado)
# Al salir del bloque 'with', la conexión se cierra automáticamente

Este enfoque garantiza que las conexiones se cierren adecuadamente, incluso si ocurren excepciones durante la ejecución.

Conexión con URI completa

Para casos más complejos, podemos utilizar una URI completa que incluya todos los parámetros de conexión:

uri = "mongodb://usuario:contraseña@localhost:27017/mi_base_datos?retryWrites=true&w=majority&authSource=admin&ssl=true"
client = MongoClient(uri)

Los parámetros en la URI permiten configurar aspectos como:

• retryWrites: reintentar operaciones de escritura fallidas
• w: nivel de confirmación de escritura
• authSource: base de datos para autenticación
• ssl: habilitar conexión segura mediante SSL/TLS

Manejo de errores de conexión

Es importante implementar un manejo adecuado de errores para las conexiones:

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError

try:
    client = MongoClient('mongodb://localhost:27017/', serverSelectionTimeoutMS=2000)
    client.admin.command('ismaster')  # Comando para verificar conexión
    print("Servidor MongoDB disponible")
except ConnectionFailure:
    print("No se pudo conectar al servidor MongoDB")
except ServerSelectionTimeoutError:
    print("Tiempo de espera agotado al intentar conectar con MongoDB")

Capturar excepciones específicas nos permite responder adecuadamente a diferentes tipos de problemas de conexión.

Conexiones con pools

Para aplicaciones con alto tráfico, PyMongo gestiona automáticamente un pool de conexiones:

# Configuración de pool de conexiones
client = MongoClient(
    'mongodb://localhost:27017/',
    maxPoolSize=100,      # Máximo de conexiones en el pool
    minPoolSize=10,       # Mínimo de conexiones mantenidas en el pool
    maxIdleTimeMS=45000   # Tiempo máximo que una conexión puede estar inactiva
)

El pool de conexiones mejora el rendimiento al reutilizar conexiones existentes en lugar de crear nuevas para cada operación.

Operaciones con documentos

Una vez establecida la conexión con MongoDB, podemos comenzar a realizar operaciones CRUD (Crear, Leer, Actualizar y Eliminar) con los documentos. En MongoDB, los datos se almacenan como documentos en formato BSON (Binary JSON), lo que permite estructuras de datos flexibles y anidadas.

Inserción de documentos

PyMongo ofrece varios métodos para insertar documentos en una colección. Podemos insertar un solo documento o múltiples documentos a la vez.

Insertar un documento único

Para insertar un único documento, utilizamos el método insert_one():

from pymongo import MongoClient

# Establecer conexión
client = MongoClient('mongodb://localhost:27017/')
db = client['tienda']
coleccion = db['productos']

# Crear un documento
producto = {
    "nombre": "Laptop",
    "marca": "TechPro",
    "precio": 1200.50,
    "stock": 10,
    "categorias": ["electrónica", "computadoras"],
    "especificaciones": {
        "procesador": "Intel i7",
        "memoria": "16GB",
        "almacenamiento": "512GB SSD"
    }
}

# Insertar el documento
resultado = coleccion.insert_one(producto)

# Obtener el ID del documento insertado
print(f"Documento insertado con ID: {resultado.inserted_id}")

El método insert_one() devuelve un objeto InsertOneResult que contiene el inserted_id, un identificador único generado automáticamente por MongoDB si no lo especificamos.

Insertar múltiples documentos

Para insertar varios documentos a la vez, utilizamos insert_many():

# Lista de documentos a insertar
productos = [
    {
        "nombre": "Monitor",
        "marca": "ViewClear",
        "precio": 350.75,
        "stock": 15,
        "categorias": ["electrónica", "periféricos"]
    },
    {
        "nombre": "Teclado",
        "marca": "TypeMaster",
        "precio": 85.99,
        "stock": 30,
        "categorias": ["periféricos", "accesorios"]
    },
    {
        "nombre": "Mouse",
        "marca": "ClickPro",
        "precio": 45.50,
        "stock": 25,
        "categorias": ["periféricos", "accesorios"]
    }
]

# Insertar múltiples documentos
resultado = coleccion.insert_many(productos)

# Obtener los IDs de los documentos insertados
print(f"Documentos insertados: {len(resultado.inserted_ids)}")
print(f"IDs: {resultado.inserted_ids}")

El método insert_many() devuelve un objeto InsertManyResult que contiene una lista de los IDs de todos los documentos insertados.

Lectura de documentos

PyMongo proporciona varios métodos para recuperar documentos de una colección.

Encontrar un documento

Para recuperar un único documento, utilizamos find_one():

# Buscar un documento por un criterio específico
producto = coleccion.find_one({"nombre": "Laptop"})

if producto:
    print(f"Producto encontrado: {producto['nombre']}, Precio: {producto['precio']}")
else:
    print("Producto no encontrado")

El método find_one() devuelve el primer documento que coincide con el criterio de búsqueda o None si no encuentra ninguno.

Encontrar múltiples documentos

Para recuperar múltiples documentos, utilizamos el método find():

# Buscar todos los productos de una categoría
cursor = coleccion.find({"categorias": "periféricos"})

# Iterar sobre los resultados
for producto in cursor:
    print(f"{producto['nombre']} - {producto['marca']} - ${producto['precio']}")

El método find() devuelve un cursor que podemos iterar para acceder a los documentos. El cursor es un objeto iterable que recupera los documentos de forma eficiente, cargándolos por lotes según sea necesario.

Filtrado y proyección

Podemos filtrar los resultados y especificar qué campos queremos recuperar:

# Buscar productos con precio menor a 100 y mostrar solo nombre y precio
cursor = coleccion.find(
    {"precio": {"$lt": 100}},  # Filtro: precio menor que 100
    {"nombre": 1, "precio": 1, "_id": 0}  # Proyección: incluir nombre y precio, excluir _id
)

for producto in cursor:
    print(f"{producto['nombre']}: ${producto['precio']}")

En la proyección, especificamos 1 para incluir un campo y 0 para excluirlo. Por defecto, el campo _id siempre se incluye a menos que lo excluyamos explícitamente.

Ordenación, límite y salto

Podemos ordenar, limitar y saltar resultados:

# Buscar productos ordenados por precio (descendente), 
# saltando los 2 primeros y limitando a 3 resultados
cursor = coleccion.find().sort("precio", -1).skip(2).limit(3)

print("Productos (ordenados por precio descendente, saltando 2, limitando a 3):")
for producto in cursor:
    print(f"{producto['nombre']}: ${producto['precio']}")

• sort(): Ordena los resultados. Usamos 1 para orden ascendente y -1 para descendente.
• skip(): Salta un número específico de documentos.
• limit(): Limita el número de documentos devueltos.

Actualización de documentos

PyMongo ofrece varios métodos para actualizar documentos existentes.

Actualizar un documento

Para actualizar un único documento, utilizamos update_one():

# Actualizar el precio de un producto
resultado = coleccion.update_one(
    {"nombre": "Laptop"},  # Filtro para encontrar el documento
    {"$set": {"precio": 1299.99, "stock": 8}}  # Operación de actualización
)

print(f"Documentos encontrados: {resultado.matched_count}")
print(f"Documentos modificados: {resultado.modified_count}")

El método update_one() actualiza el primer documento que coincide con el filtro. Devuelve un objeto UpdateResult con información sobre la operación.

Actualizar múltiples documentos

Para actualizar varios documentos a la vez, utilizamos update_many():

# Aplicar un descuento del 10% a todos los productos de la categoría "periféricos"
resultado = coleccion.update_many(
    {"categorias": "periféricos"},  # Filtro
    {"$mul": {"precio": 0.9}}  # Multiplicar el precio por 0.9 (10% de descuento)
)

print(f"Documentos actualizados: {resultado.modified_count}")

El operador $mul multiplica el valor del campo por el número especificado.

Operadores de actualización

MongoDB proporciona varios operadores para realizar diferentes tipos de actualizaciones:

# Ejemplos de diferentes operadores de actualización
coleccion.update_one(
    {"nombre": "Monitor"},
    {
        "$set": {"en_oferta": True},  # Establecer un valor
        "$inc": {"stock": -2},  # Incrementar (o decrementar con valor negativo)
        "$push": {"categorias": "ofertas"},  # Añadir un elemento a un array
        "$currentDate": {"ultima_modificacion": True}  # Establecer fecha actual
    }
)

Algunos operadores comunes:

• $set: Establece el valor de un campo
• $inc: Incrementa el valor de un campo
• $push: Añade un elemento a un array
• $pull: Elimina elementos de un array
• $currentDate: Establece el valor a la fecha actual

Upsert (actualizar o insertar)

El parámetro upsert permite crear un documento si no existe ninguno que coincida con el filtro:

# Actualizar un producto o crearlo si no existe
coleccion.update_one(
    {"nombre": "Auriculares"},
    {
        "$set": {
            "marca": "SoundMaster",
            "precio": 129.99,
            "stock": 20,
            "categorias": ["audio", "periféricos"]
        }
    },
    upsert=True  # Crear el documento si no existe
)

Eliminación de documentos

PyMongo proporciona métodos para eliminar documentos de una colección.

Eliminar un documento

Para eliminar un único documento, utilizamos delete_one():

# Eliminar un producto agotado
resultado = coleccion.delete_one({"stock": 0})

print(f"Documentos eliminados: {resultado.deleted_count}")

El método delete_one() elimina el primer documento que coincide con el filtro.

Eliminar múltiples documentos

Para eliminar varios documentos a la vez, utilizamos delete_many():

# Eliminar todos los productos descontinuados
resultado = coleccion.delete_many({"descontinuado": True})

print(f"Documentos eliminados: {resultado.deleted_count}")

Eliminar todos los documentos

Para eliminar todos los documentos de una colección:

# Eliminar todos los documentos (¡usar con precaución!)
resultado = coleccion.delete_many({})
print(f"Se eliminaron todos los documentos: {resultado.deleted_count}")

Operaciones atómicas

MongoDB permite realizar operaciones atómicas con el método find_one_and_update() y similares:

# Decrementar el stock y obtener el documento actualizado en una sola operación
producto = coleccion.find_one_and_update(
    {"nombre": "Laptop", "stock": {"$gt": 0}},  # Filtro: producto con stock > 0
    {"$inc": {"stock": -1}},  # Decrementar stock en 1
    return_document=True  # Devolver el documento después de la actualización
)

if producto:
    print(f"Producto vendido: {producto['nombre']}, Stock restante: {producto['stock']}")
else:
    print("Producto agotado o no encontrado")

Otros métodos similares:

• find_one_and_delete(): Encuentra un documento y lo elimina
• find_one_and_replace(): Encuentra un documento y lo reemplaza

Operaciones en masa (Bulk Operations)

Para operaciones en masa más eficientes, podemos utilizar las operaciones bulk:

from pymongo import UpdateOne, DeleteOne

# Crear una lista de operaciones
operaciones = [
    UpdateOne({"nombre": "Laptop"}, {"$inc": {"stock": 5}}),
    UpdateOne({"nombre": "Monitor"}, {"$set": {"precio": 329.99}}),
    DeleteOne({"nombre": "Producto obsoleto"})
]

# Ejecutar todas las operaciones en una sola solicitud
resultado = coleccion.bulk_write(operaciones)

print(f"Documentos insertados: {resultado.inserted_count}")
print(f"Documentos actualizados: {resultado.modified_count}")
print(f"Documentos eliminados: {resultado.deleted_count}")

Las operaciones bulk son más eficientes cuando necesitamos realizar múltiples operaciones, ya que reducen el número de viajes de ida y vuelta al servidor.

Transacciones

A partir de MongoDB 4.0, podemos utilizar transacciones para garantizar la atomicidad de múltiples operaciones:

# Iniciar una sesión
with client.start_session() as sesion:
    # Iniciar una transacción
    with sesion.start_transaction():
        # Realizar múltiples operaciones dentro de la transacción
        db.productos.update_one(
            {"_id": producto_id}, 
            {"$inc": {"stock": -1}}, 
            session=sesion
        )
        
        db.ventas.insert_one(
            {
                "producto_id": producto_id,
                "cantidad": 1,
                "fecha": datetime.now()
            }, 
            session=sesion
        )
        
        # Si llegamos aquí sin errores, la transacción se confirmará automáticamente
        # Si ocurre una excepción, la transacción se abortará

Las transacciones son especialmente útiles cuando necesitamos garantizar que un conjunto de operaciones se realice como una unidad atómica.

Consultas y filtros

Las consultas y filtros son el núcleo de la interacción con MongoDB, permitiéndonos recuperar exactamente los datos que necesitamos. PyMongo proporciona una API intuitiva que refleja la sintaxis de consulta nativa de MongoDB, facilitando la construcción de consultas desde simples hasta altamente complejas.

Consultas básicas

Para realizar consultas en MongoDB, utilizamos principalmente los métodos find() y find_one(). La diferencia fundamental es que find() devuelve un cursor iterable con todos los documentos coincidentes, mientras que find_one() devuelve un único documento.

from pymongo import MongoClient

# Conexión a la base de datos
client = MongoClient('mongodb://localhost:27017/')
db = client['biblioteca']
coleccion = db['libros']

# Consulta básica - todos los documentos
todos_libros = coleccion.find()

# Consulta con filtro simple
libros_python = coleccion.find({"tema": "Python"})

# Obtener un único documento
primer_libro = coleccion.find_one({"titulo": "Aprende Python en 24 horas"})

Operadores de comparación

MongoDB ofrece varios operadores de comparación que podemos utilizar en nuestras consultas:

# Libros con precio mayor que 20
libros_caros = coleccion.find({"precio": {"$gt": 20}})

# Libros publicados después de 2020
libros_recientes = coleccion.find({"año_publicacion": {"$gte": 2020}})

# Libros que no son de programación
libros_no_programacion = coleccion.find({"tema": {"$ne": "Programación"}})

# Libros con precio entre 15 y 30
rango_precio = coleccion.find({
    "precio": {
        "$gte": 15,
        "$lte": 30
    }
})

Los operadores de comparación más comunes son:

• $eq: Igual a (=)
• $ne: No igual a (≠)
• $gt: Mayor que (>)
• $gte: Mayor o igual que (≥)
• $lt: Menor que (<)
• $lte: Menor o igual que (≤)

Operadores lógicos

Para construir consultas más complejas, podemos combinar condiciones utilizando operadores lógicos:

# Libros de Python O JavaScript
libros_programacion = coleccion.find({
    "$or": [
        {"tema": "Python"},
        {"tema": "JavaScript"}
    ]
})

# Libros de Python que cuestan menos de 25
libros_python_baratos = coleccion.find({
    "$and": [
        {"tema": "Python"},
        {"precio": {"$lt": 25}}
    ]
})

# Forma simplificada del AND (implícito)
libros_python_baratos = coleccion.find({
    "tema": "Python",
    "precio": {"$lt": 25}
})

# Libros que NO son ni de Python ni de JavaScript
otros_libros = coleccion.find({
    "$nor": [
        {"tema": "Python"},
        {"tema": "JavaScript"}
    ]
})

Los operadores lógicos principales son:

• $and: Todas las condiciones deben cumplirse
• $or: Al menos una condición debe cumplirse
• $nor: Ninguna de las condiciones debe cumplirse
• $not: Niega una condición

Consultas en arrays

MongoDB permite realizar consultas sofisticadas en campos de tipo array:

# Libros que tienen exactamente las etiquetas "programación" y "principiantes"
libros_exactos = coleccion.find({
    "etiquetas": ["programación", "principiantes"]
})

# Libros que contienen la etiqueta "programación" (en cualquier posición del array)
libros_programacion = coleccion.find({
    "etiquetas": "programación"
})

# Libros que contienen AMBAS etiquetas (en cualquier orden)
libros_ambas = coleccion.find({
    "etiquetas": {
        "$all": ["programación", "principiantes"]
    }
})

# Libros con exactamente 3 etiquetas
libros_tres_etiquetas = coleccion.find({
    "etiquetas": {
        "$size": 3
    }
})

# Libros donde al menos una etiqueta cumple una condición
libros_etiqueta_larga = coleccion.find({
    "etiquetas": {
        "$elemMatch": {
            "$regex": "^.{10,}$"  # Etiquetas con 10 o más caracteres
        }
    }
})

Consultas en documentos anidados

Para consultar campos dentro de documentos anidados, utilizamos la notación de punto:

# Libros con un editor específico
libros_editor = coleccion.find({
    "detalles_publicacion.editor": "TechBooks"
})

# Libros publicados en España
libros_españa = coleccion.find({
    "detalles_publicacion.pais": "España"
})

# Libros con más de 5 reseñas positivas
libros_populares = coleccion.find({
    "estadisticas.reseñas.positivas": {"$gt": 5}
})

Expresiones regulares

PyMongo permite utilizar expresiones regulares para búsquedas de texto más flexibles:

import re

# Búsqueda con expresión regular (case sensitive)
libros_python = coleccion.find({
    "titulo": {"$regex": "Python"}
})

# Búsqueda insensible a mayúsculas/minúsculas
libros_python = coleccion.find({
    "titulo": {"$regex": "python", "$options": "i"}
})

# Alternativa usando el módulo re de Python
patron = re.compile("python", re.IGNORECASE)
libros_python = coleccion.find({
    "titulo": patron
})

# Títulos que comienzan con "Aprende"
libros_aprende = coleccion.find({
    "titulo": {"$regex": "^Aprende"}
})

# Títulos que terminan con "Python"
libros_python_final = coleccion.find({
    "titulo": {"$regex": "Python$"}
})

Consultas de texto completo

MongoDB ofrece capacidades de búsqueda de texto completo que podemos aprovechar con PyMongo:

# Primero, necesitamos crear un índice de texto (solo una vez)
coleccion.create_index([("titulo", "text"), ("descripcion", "text")])

# Búsqueda de texto
resultados = coleccion.find({
    "$text": {"$search": "python programación"}
})

# Búsqueda de texto con puntuación de relevancia
resultados = coleccion.find(
    {"$text": {"$search": "python programación"}},
    {"score": {"$meta": "textScore"}}
).sort([("score", {"$meta": "textScore"})])

La búsqueda de texto completo es ideal para implementar funcionalidades de búsqueda en aplicaciones donde los usuarios necesitan encontrar documentos basados en su contenido textual.

Proyecciones

Las proyecciones nos permiten especificar qué campos queremos incluir o excluir en los resultados:

# Solo incluir título y precio (y _id por defecto)
libros_resumidos = coleccion.find(
    {"tema": "Python"},
    {"titulo": 1, "precio": 1}
)

# Excluir campos específicos
libros_sin_detalles = coleccion.find(
    {"tema": "Python"},
    {"detalles_publicacion": 0, "estadisticas": 0}
)

# Excluir _id e incluir solo ciertos campos
libros_custom = coleccion.find(
    {"tema": "Python"},
    {"_id": 0, "titulo": 1, "autor": 1, "precio": 1}
)

Ordenación, límite y salto

Estos métodos nos permiten controlar cómo se presentan los resultados:

# Ordenar por precio (ascendente)
libros_ordenados = coleccion.find().sort("precio", 1)

# Ordenar por precio (descendente)
libros_caros_primero = coleccion.find().sort("precio", -1)

# Ordenar por múltiples campos
libros_ordenados = coleccion.find().sort([
    ("tema", 1),      # Primero por tema (ascendente)
    ("precio", -1)    # Luego por precio (descendente)
])

# Limitar resultados (los 5 primeros)
libros_top = coleccion.find().limit(5)

# Saltar resultados (paginación)
# Página 1 (primeros 10)
pagina_1 = coleccion.find().skip(0).limit(10)
# Página 2 (siguientes 10)
pagina_2 = coleccion.find().skip(10).limit(10)

Consultas con operadores de elementos

MongoDB proporciona operadores para verificar la existencia o el tipo de un campo:

# Libros que tienen el campo "isbn"
libros_con_isbn = coleccion.find({
    "isbn": {"$exists": True}
})

# Libros que NO tienen el campo "reseñas"
libros_sin_reseñas = coleccion.find({
    "reseñas": {"$exists": False}
})

# Libros donde "ventas" es un número
libros_ventas_numero = coleccion.find({
    "ventas": {"$type": "number"}
})

# Libros donde "etiquetas" es un array
libros_con_etiquetas = coleccion.find({
    "etiquetas": {"$type": "array"}
})

Consultas geoespaciales

MongoDB soporta consultas geoespaciales para datos con coordenadas:

# Crear un índice geoespacial (solo una vez)
db.librerias.create_index([("ubicacion", "2dsphere")])

# Encontrar librerías cercanas a un punto (en un radio de 5km)
librerias_cercanas = db.librerias.find({
    "ubicacion": {
        "$nearSphere": {
            "$geometry": {
                "type": "Point",
                "coordinates": [-3.7038, 40.4168]  # Madrid
            },
            "$maxDistance": 5000  # 5km en metros
        }
    }
})

Consultas con operadores de evaluación

Estos operadores permiten evaluaciones más complejas:

# Documentos donde el campo "precio" es divisible por 5
libros_precio_redondo = coleccion.find({
    "precio": {"$mod": [5, 0]}
})

# Documentos donde el campo cumple una condición
import json
libros_especiales = coleccion.find({
    "$expr": {"$gt": ["$precio", "$precio_original"]}
})

# Documentos que coinciden con un patrón JSON
patron = json.loads('{"autor": "John Doe", "tema": "Python"}')
libros_coincidentes = coleccion.find({
    "$jsonSchema": {
        "required": ["autor", "tema"],
        "properties": {
            "autor": {"type": "string"},
            "tema": {"enum": ["Python", "JavaScript"]}
        }
    }
})

Optimización de consultas con índices

Para mejorar el rendimiento de las consultas, es fundamental crear índices adecuados:

# Crear un índice simple
coleccion.create_index("titulo")

# Crear un índice compuesto
coleccion.create_index([("autor", 1), ("año_publicacion", -1)])

# Crear un índice único
coleccion.create_index("isbn", unique=True)

# Crear un índice en segundo plano (no bloquea operaciones)
coleccion.create_index("tema", background=True)

# Verificar índices existentes
indices = coleccion.index_information()
print(indices)

Explicación de consultas

Para entender cómo MongoDB ejecuta una consulta y si está utilizando índices:

# Explicar una consulta
explicacion = coleccion.find({"tema": "Python"}).explain()
print(explicacion)

# Explicación detallada
explicacion_detallada = coleccion.find({"tema": "Python"}).explain("executionStats")
print(f"Documentos examinados: {explicacion_detallada['executionStats']['totalDocsExamined']}")
print(f"Tiempo de ejecución: {explicacion_detallada['executionStats']['executionTimeMillis']}ms")

Consultas con hint (forzar uso de índice)

Podemos forzar a MongoDB a utilizar un índice específico:

# Forzar el uso de un índice específico
resultados = coleccion.find({"autor": "John Doe", "tema": "Python"}).hint("autor_1_tema_1")

# Alternativa especificando los campos del índice
resultados = coleccion.find({"autor": "John Doe", "tema": "Python"}).hint([("autor", 1), ("tema", 1)])

Consultas con count y distinct

Para contar documentos o encontrar valores únicos:

# Contar todos los documentos
total_libros = coleccion.count_documents({})

# Contar con filtro
libros_python = coleccion.count_documents({"tema": "Python"})

# Valores únicos de un campo
temas_unicos = coleccion.distinct("tema")

# Valores únicos con filtro
autores_python = coleccion.distinct("autor", {"tema": "Python"})

Ejemplo práctico: Sistema de búsqueda avanzada

Veamos un ejemplo completo que combina varias técnicas de consulta:

def buscar_libros(
    db, 
    texto=None, 
    tema=None, 
    precio_min=None, 
    precio_max=None, 
    año_desde=None,
    ordenar_por="relevancia",
    pagina=1,
    por_pagina=10
):
    """
    Función de búsqueda avanzada para libros.
    """
    # Construir el filtro base
    filtro = {}
    
    # Añadir condiciones según los parámetros
    if texto:
        filtro["$text"] = {"$search": texto}
    
    if tema:
        filtro["tema"] = tema
    
    # Condiciones de precio
    if precio_min is not None or precio_max is not None:
        filtro["precio"] = {}
        if precio_min is not None:
            filtro["precio"]["$gte"] = precio_min
        if precio_max is not None:
            filtro["precio"]["$lte"] = precio_max
    
    # Condición de año
    if año_desde:
        filtro["año_publicacion"] = {"$gte": año_desde}
    
    # Configurar proyección y ordenación
    proyeccion = {
        "titulo": 1,
        "autor": 1,
        "tema": 1,
        "precio": 1,
        "año_publicacion": 1
    }
    
    # Añadir puntuación si hay búsqueda de texto
    if texto:
        proyeccion["score"] = {"$meta": "textScore"}
    
    # Configurar ordenación
    if ordenar_por == "relevancia" and texto:
        orden = [("score", {"$meta": "textScore"})]
    elif ordenar_por == "precio_asc":
        orden = [("precio", 1)]
    elif ordenar_por == "precio_desc":
        orden = [("precio", -1)]
    elif ordenar_por == "año_desc":
        orden = [("año_publicacion", -1)]
    else:
        orden = [("titulo", 1)]
    
    # Calcular salto para paginación
    salto = (pagina - 1) * por_pagina
    
    # Ejecutar consulta
    cursor = db.libros.find(
        filtro,
        proyeccion
    ).sort(orden).skip(salto).limit(por_pagina)
    
    # Contar total de resultados (para paginación)
    total = db.libros.count_documents(filtro)
    
    # Convertir cursor a lista
    resultados = list(cursor)
    
    return {
        "resultados": resultados,
        "total": total,
        "pagina": pagina,
        "paginas_totales": (total + por_pagina - 1) // por_pagina
    }

# Ejemplo de uso
resultados = buscar_libros(
    db,
    texto="python avanzado",
    precio_min=20,
    precio_max=50,
    año_desde=2018,
    ordenar_por="relevancia",
    pagina=1
)

print(f"Encontrados {resultados['total']} libros")
for libro in resultados['resultados']:
    print(f"{libro['titulo']} - {libro['autor']} - ${libro['precio']}")

Este ejemplo muestra cómo construir un sistema de búsqueda flexible que combina múltiples criterios, ordenación y paginación, demostrando el poder de las consultas y filtros en MongoDB con PyMongo.

Agregaciones básicas

Las agregaciones en MongoDB representan una de las características más potentes para el análisis y transformación de datos. A diferencia de las consultas simples, el framework de agregación permite procesar múltiples documentos y realizar operaciones complejas en ellos, similar a cómo funcionaría un pipeline de procesamiento de datos.

PyMongo proporciona acceso completo a esta funcionalidad a través del método aggregate(), que nos permite construir pipelines de agregación para realizar análisis avanzados sobre nuestros datos.

Concepto de pipeline de agregación

Un pipeline de agregación es una secuencia de etapas de procesamiento donde la salida de cada etapa se convierte en la entrada de la siguiente. Cada etapa transforma los documentos a medida que pasan por el pipeline.

from pymongo import MongoClient

# Conexión a MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['tienda']
coleccion = db['ventas']

# Pipeline de agregación básico
pipeline = [
    {"$match": {"estado": "completado"}},
    {"$group": {"_id": "$categoria", "total": {"$sum": "$importe"}}}
]

# Ejecutar la agregación
resultados = coleccion.aggregate(pipeline)

# Iterar sobre los resultados
for resultado in resultados:
    print(f"Categoría: {resultado['_id']}, Total: {resultado['total']}")

En este ejemplo, el pipeline consta de dos etapas:

1. $match: Filtra los documentos para incluir solo ventas completadas
2. $group: Agrupa los documentos por categoría y calcula la suma de los importes

Etapas comunes de agregación

Etapa $match

La etapa $match filtra los documentos, similar a cómo funciona el método find():

# Filtrar productos con stock bajo
pipeline = [
    {"$match": {"stock": {"$lt": 10}}}
]

productos_stock_bajo = db.productos.aggregate(pipeline)

Esta etapa es más eficiente cuando se coloca al inicio del pipeline, ya que reduce la cantidad de documentos que pasan a las siguientes etapas.

Etapa $group

La etapa $group agrupa documentos por una clave especificada y permite aplicar operadores de acumulación:

# Calcular ventas totales por país
pipeline = [
    {"$group": {
        "_id": "$pais",
        "ventas_totales": {"$sum": "$importe"},
        "num_ventas": {"$sum": 1},
        "importe_promedio": {"$avg": "$importe"},
        "importe_maximo": {"$max": "$importe"},
        "importe_minimo": {"$min": "$importe"}
    }}
]

estadisticas_por_pais = db.ventas.aggregate(pipeline)

Los operadores de acumulación más comunes son:

• $sum: Suma valores o cuenta documentos (usando 1)
• $avg: Calcula el promedio
• $min y $max: Encuentran valores mínimos y máximos
• $first y $last: Obtienen el primer o último valor del grupo

Etapa $project

La etapa $project permite remodelar documentos, incluyendo, excluyendo o transformando campos:

# Transformar documentos para mostrar solo información relevante
pipeline = [
    {"$match": {"categoria": "electrónica"}},
    {"$project": {
        "_id": 0,
        "nombre": 1,
        "precio_con_iva": {"$multiply": ["$precio", 1.21]},
        "descuento_aplicable": {"$cond": [{"$gt": ["$precio", 100]}, 10, 5]},
        "disponibilidad": {"$switch": {
            "branches": [
                {"case": {"$eq": ["$stock", 0]}, "then": "Agotado"},
                {"case": {"$lt": ["$stock", 5]}, "then": "Últimas unidades"}
            ],
            "default": "Disponible"
        }}
    }}
]

productos_transformados = db.productos.aggregate(pipeline)

En este ejemplo:

• Excluimos el campo _id y mantenemos nombre
• Calculamos un nuevo campo precio_con_iva multiplicando el precio por 1.21
• Creamos un campo descuento_aplicable usando una condición
• Generamos un campo disponibilidad usando una estructura switch-case

Etapa $sort

La etapa $sort ordena los documentos según los campos especificados:

# Ordenar productos por precio descendente
pipeline = [
    {"$sort": {"precio": -1}}  # 1 para ascendente, -1 para descendente
]

productos_ordenados = db.productos.aggregate(pipeline)

Etapa $limit y $skip

Estas etapas limitan el número de documentos o saltan un número específico:

# Implementar paginación
pagina = 2
elementos_por_pagina = 10

pipeline = [
    {"$sort": {"fecha_creacion": -1}},
    {"$skip": (pagina - 1) * elementos_por_pagina},
    {"$limit": elementos_por_pagina}
]

productos_paginados = db.productos.aggregate(pipeline)

Operaciones de agregación avanzadas

Etapa $unwind

La etapa $unwind descompone arrays en documentos individuales, lo que facilita el análisis de datos en arrays:

# Descomponer productos por etiquetas
pipeline = [
    {"$unwind": "$etiquetas"},
    {"$group": {
        "_id": "$etiquetas",
        "cantidad": {"$sum": 1},
        "productos": {"$push": "$nombre"}
    }},
    {"$sort": {"cantidad": -1}}
]

etiquetas_populares = db.productos.aggregate(pipeline)

Este pipeline:

1. Descompone cada documento por su array de etiquetas
2. Agrupa por etiqueta y cuenta productos
3. Crea un array con los nombres de productos para cada etiqueta
4. Ordena por cantidad descendente

Etapa $lookup

La etapa $lookup realiza una operación similar a un JOIN en SQL, permitiendo combinar documentos de diferentes colecciones:

# Combinar pedidos con información de clientes
pipeline = [
    {"$match": {"estado": "pendiente"}},
    {"$lookup": {
        "from": "clientes",
        "localField": "cliente_id",
        "foreignField": "_id",
        "as": "info_cliente"
    }},
    {"$project": {
        "numero_pedido": 1,
        "importe": 1,
        "fecha": 1,
        "cliente": {"$arrayElemAt": ["$info_cliente", 0]}
    }}
]

pedidos_con_clientes = db.pedidos.aggregate(pipeline)

En este ejemplo:

1. Filtramos pedidos pendientes
2. Realizamos un lookup para obtener información del cliente
3. Proyectamos los campos necesarios, extrayendo el primer elemento del array info_cliente

Etapa $facet

La etapa $facet permite realizar múltiples agregaciones en paralelo dentro del mismo pipeline:

# Obtener múltiples métricas en una sola consulta
pipeline = [
    {"$match": {"fecha": {"$gte": datetime(2023, 1, 1)}}},
    {"$facet": {
        "ventas_por_categoria": [
            {"$group": {"_id": "$categoria", "total": {"$sum": "$importe"}}}
        ],
        "ventas_por_mes": [
            {"$group": {
                "_id": {"$dateToString": {"format": "%Y-%m", "date": "$fecha"}},
                "total": {"$sum": "$importe"}
            }},
            {"$sort": {"_id": 1}}
        ],
        "top_productos": [
            {"$group": {"_id": "$producto_id", "unidades": {"$sum": "$cantidad"}}},
            {"$sort": {"unidades": -1}},
            {"$limit": 5}
        ]
    }}
]

dashboard = db.ventas.aggregate(pipeline)

Esta agregación genera un dashboard completo con tres secciones diferentes en una sola consulta.

Operadores de agregación útiles

Operadores aritméticos

pipeline = [
    {"$project": {
        "nombre": 1,
        "precio_original": "$precio",
        "precio_con_descuento": {"$subtract": ["$precio", {"$multiply": ["$precio", {"$divide": ["$descuento", 100]}]}]},
        "ahorro": {"$multiply": ["$precio", {"$divide": ["$descuento", 100]}]}
    }}
]

Operadores de fecha

pipeline = [
    {"$project": {
        "año": {"$year": "$fecha"},
        "mes": {"$month": "$fecha"},
        "dia_semana": {"$dayOfWeek": "$fecha"},
        "es_fin_semana": {"$in": [{"$dayOfWeek": "$fecha"}, [1, 7]]},
        "dias_desde_compra": {
            "$divide": [
                {"$subtract": [new Date(), "$fecha"]},
                1000 * 60 * 60 * 24  # Convertir milisegundos a días
            ]
        }
    }}
]

Operadores de array

pipeline = [
    {"$project": {
        "nombre": 1,
        "num_etiquetas": {"$size": "$etiquetas"},
        "primera_etiqueta": {"$arrayElemAt": ["$etiquetas", 0]},
        "ultima_etiqueta": {"$arrayElemAt": ["$etiquetas", -1]},
        "tiene_oferta": {"$in": ["oferta", "$etiquetas"]},
        "etiquetas_ordenadas": {"$sort": "$etiquetas"}
    }}
]

Ejemplo práctico: Análisis de ventas

Veamos un ejemplo completo que combina varias etapas para realizar un análisis de ventas:

def analisis_ventas(db, año, mes=None):
    """
    Realiza un análisis completo de ventas para un período específico.
    """
    # Construir filtro de fecha
    match_fecha = {"$match": {"fecha": {"$gte": datetime(año, 1, 1)}}}
    if mes:
        # Si se especifica mes, ajustar el filtro
        primer_dia = datetime(año, mes, 1)
        if mes == 12:
            ultimo_dia = datetime(año + 1, 1, 1)
        else:
            ultimo_dia = datetime(año, mes + 1, 1)
        match_fecha = {"$match": {"fecha": {"$gte": primer_dia, "$lt": ultimo_dia}}}
    
    # Pipeline de agregación
    pipeline = [
        match_fecha,
        {"$facet": {
            "resumen": [
                {"$group": {
                    "_id": None,
                    "total_ventas": {"$sum": "$importe"},
                    "num_transacciones": {"$sum": 1},
                    "ticket_promedio": {"$avg": "$importe"},
                    "ticket_maximo": {"$max": "$importe"}
                }}
            ],
            "ventas_por_categoria": [
                {"$group": {
                    "_id": "$categoria",
                    "total": {"$sum": "$importe"},
                    "porcentaje": {"$sum": "$importe"}  # Calcularemos porcentaje después
                }},
                {"$sort": {"total": -1}}
            ],
            "ventas_diarias": [
                {"$group": {
                    "_id": {"$dateToString": {"format": "%Y-%m-%d", "date": "$fecha"}},
                    "total": {"$sum": "$importe"},
                    "transacciones": {"$sum": 1}
                }},
                {"$sort": {"_id": 1}}
            ],
            "top_productos": [
                {"$group": {
                    "_id": "$producto_id",
                    "nombre": {"$first": "$producto_nombre"},
                    "unidades": {"$sum": "$cantidad"},
                    "ingresos": {"$sum": "$importe"}
                }},
                {"$sort": {"ingresos": -1}},
                {"$limit": 10}
            ]
        }}
    ]
    
    # Ejecutar agregación
    resultado = list(db.ventas.aggregate(pipeline))[0]
    
    # Calcular porcentajes para ventas por categoría
    if resultado["ventas_por_categoria"] and resultado["resumen"]:
        total_general = resultado["resumen"][0]["total_ventas"]
        for categoria in resultado["ventas_por_categoria"]:
            categoria["porcentaje"] = round((categoria["total"] / total_general) * 100, 2)
    
    return resultado

# Ejemplo de uso
analisis = analisis_ventas(db, 2023, 3)  # Análisis de marzo 2023

print(f"Total ventas: ${analisis['resumen'][0]['total_ventas']:,.2f}")
print(f"Ticket promedio: ${analisis['resumen'][0]['ticket_promedio']:,.2f}")

print("\nVentas por categoría:")
for cat in analisis['ventas_por_categoria']:
    print(f"{cat['_id']}: ${cat['total']:,.2f} ({cat['porcentaje']}%)")

print("\nTop 3 productos:")
for i, prod in enumerate(analisis['top_productos'][:3], 1):
    print(f"{i}. {prod['nombre']}: {prod['unidades']} unidades (${prod['ingresos']:,.2f})")

Exportar resultados de agregación

A menudo necesitamos exportar los resultados de una agregación para su posterior análisis:

import csv
import json
from bson import json_util

def exportar_resultados(resultados, formato, archivo):
    """
    Exporta resultados de agregación a CSV o JSON.
    """
    if formato.lower() == 'csv':
        # Asumimos que todos los documentos tienen la misma estructura
        with open(archivo, 'w', newline='', encoding='utf-8') as f:
            if not resultados:
                return False
            
            # Obtener encabezados del primer documento
            encabezados = resultados[0].keys()
            writer = csv.DictWriter(f, fieldnames=encabezados)
            writer.writeheader()
            writer.writerows(resultados)
    
    elif formato.lower() == 'json':
        # Convertir a JSON manejando tipos especiales de MongoDB
        with open(archivo, 'w', encoding='utf-8') as f:
            json.dump(resultados, f, default=json_util.default, indent=2)
    
    else:
        raise ValueError("Formato no soportado. Use 'csv' o 'json'")
    
    return True

# Ejemplo: Exportar ventas mensuales
pipeline = [
    {"$group": {
        "_id": {
            "año": {"$year": "$fecha"},
            "mes": {"$month": "$fecha"}
        },
        "total": {"$sum": "$importe"},
        "transacciones": {"$sum": 1}
    }},
    {"$sort": {"_id.año": 1, "_id.mes": 1}},
    {"$project": {
        "_id": 0,
        "año": "$_id.año",
        "mes": "$_id.mes",
        "total": 1,
        "transacciones": 1
    }}
]

resultados = list(db.ventas.aggregate(pipeline))
exportar_resultados(resultados, 'csv', 'ventas_mensuales.csv')

Optimización de agregaciones

Las agregaciones pueden ser operaciones intensivas. Algunas técnicas para optimizarlas:

# Usar allowDiskUse para operaciones que exceden el límite de memoria
pipeline = [/* pipeline complejo */]
resultados = db.coleccion.aggregate(pipeline, allowDiskUse=True)

# Añadir índices para campos usados en $match y $sort
db.ventas.create_index([("fecha", 1)])
db.ventas.create_index([("categoria", 1), ("importe", -1)])

# Usar hint para forzar el uso de un índice específico
pipeline = [
    {"$match": {"categoria": "electrónica", "fecha": {"$gte": datetime(2023, 1, 1)}}},
    {"$sort": {"importe": -1}}
]
resultados = db.ventas.aggregate(pipeline, hint=[("categoria", 1), ("fecha", 1)])

# Explicar el plan de ejecución
pipeline = [/* pipeline a analizar */]
explicacion = db.ventas.aggregate(pipeline, explain=True)
print(json.dumps(explicacion, indent=2, default=json_util.default))

Agregaciones con expresiones de etapa

MongoDB 4.4+ introdujo las expresiones de etapa, que permiten usar operadores de acumulación fuera de la etapa $group:

# Calcular estadísticas acumulativas
pipeline = [
    {"$match": {"fecha": {"$gte": datetime(2023, 1, 1)}}},
    {"$sort": {"fecha": 1}},
    {"$set": {
        "ventas_acumuladas": {
            "$sum": {
                "$cond": [
                    {"$lte": ["$fecha", "$$NOW"]},
                    "$importe",
                    0
                ]
            }
        }
    }}
]

Las agregaciones en MongoDB con PyMongo proporcionan un sistema flexible y potente para analizar datos, generar informes y transformar información. Dominando estas técnicas, podrás extraer conocimientos valiosos de tus datos y construir funcionalidades avanzadas en tus aplicaciones.