Cómo acceder a un servidor MongoDB mediante Python y cómo insertar, listar, eliminar y actualizar documentos.
- Requisitos para acceder a MongoDB con Python.
- Script Python con acceso a MongoDB para seleccionar, insertar, actualizar y eliminar documentos.
- Resultado de la ejecución del programa Python con acceso a MongoDB.
Requisitos para acceder a MongoDB con Python
Necesitaremos disponer de un servidor con MongoDB. En el siguiente tutorial explicamos cómo desplegar un servidor MongoDB con Docker en segundos:
Para el acceso a MongoDB desde Python, usaremos la librería pymongo, por lo que debemos instarla, con el comando:
|
1 |
pip install pymongo |
Script Python con acceso a MongoDB para seleccionar, insertar, actualizar y eliminar documentos
Mostramos a continuación el script Python completo que realiza las siguientes tareas (las líneas más importantes están comentadas con su correspondiente explicación):
- Crea tres tareas (objetos de la clase Tareas) con datos para almacenar en MongoDB.
- Solicita los datos de conexión al servidor MongoDB (servidor, puerto, base de datos, colección, usuario y contraseña).
- Inserta los documentos «Tareas» en la colección especificada de MongoDB.
- Realizan un «select» para mostrar los documentos que cumplan un determinado filtro.
- Actualiza los documentos (tareas) que cumplan un determinado filtro.
- Elimina los documentos que cumplan un determinado filtro.
- Cierra la conexión con el servidor de base de datos MongoDB.
El programa Python tendrá dos ficheros. El primer fichero, con nombre Tarea.py, tendrá el siguiente contenido:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
# Definimos la clase tarea class Tarea: # Constructor de la clase def __init__(self, nombre, accion, activa, alcance, periodicidad): self.nombre = nombre self.accion = accion self.activa = activa self.alcance = alcance self.periodicidad = periodicidad # Definiremos un método para convertir la clase Tarea en diccionario, entendible por MongoDB def coleccionTareaMongoDB (self): return { "nombre":self.nombre, "accion":self.accion, "activa": self.activa, "alcance":self.alcance, "periodicidad":self.periodicidad } def __str__(self): return f"Tarea: {self.nombre} | Acción: {self.accion} | Activa: {self.activa} | Alcance: {self.alcance} | Periodicidad: {self.periodicidad}" |
Este fichero contendrá la clase «Tarea» con los campos que se guardarán en la BD MongoBD y el método coleccionTareaMongoDB para convertir el objeto Tarea en un diccionario, para que sea entendible por MongoDB.
Por otro lado, tendremos el fichero principal de la aplicación Python, que tendrá el siguiente contenido y que guardaremos con el nombre mongodb.py:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
from pymongo import MongoClient from urllib.parse import quote_plus import getpass from Tarea import Tarea # Para el ejemplo, crearemos una lista de tareas (será la que insertemos en MongoDB) # En producción, estos datos o bien los obtenemos de un fichero JSON o bien de otro servicio que los proporcione tareas = [ Tarea("Instalar_Java", "C:\\Software\\install_java.exe", True, ["eq23", "eq12", "server11", "server2"], "Semanal"), Tarea("Instalar_Adobe_Reader", "C:\\Software\\install_reader.ps", False, ["eq50", "serverrdp"], "Mensual"), Tarea("Instalar_Python", "C:\\Software\\install_python.py", False, ["dis_e", "equipo_ora"], "Diaria"), Tarea("Instalar_Zip", "C:\\Software\\zip.bat", True, ["*"], "Diaria") ] # Solicitamos los datos de conexión al servidor MongoDB servidorMongoDB = input("Introduzca la IP o nombre DNS del servidor MongoDB (por defecto localhost): ") if servidorMongoDB == "": servidorMongoDB ="localhost" puerto = input("Introduzca el puerto de conexión al servidor MongoDB (Por defecto 27017): ") if puerto == "": puerto = "27017" bd = input("Introduzca el nombre de la base de datos (debe existir, por defecto bdproyectoa): ") if bd == "": bd = "bdproyectoa" coleccion = input("Introduzca el nombre de la colección (debe existir en la BD, por defecto bdproyectoa): ") if coleccion == "": coleccion = "bdproyectoa" usuario = input("Introduzca el usuario de conexión: ") contrasena = getpass.getpass("Introduzca la contraseña: ") # Establecemos la URI de conexión con el servidor MongoDB # Usamos el método quote_plus para hacer más seguro el acceso y limpiar de posibles caracteres no deseados uri = "mongodb://%s:%s@%s:%s" % ( quote_plus(usuario), quote_plus(contrasena), servidorMongoDB, puerto) try: # Conexión al servidor MongoDB pasándole los datos de conexión conMongo = MongoClient(uri) # Abrimos la base de datos "bdproyectoa" (debe existir en el servidor MongoDB) bdMongo = conMongo[bd] # Abrimos la colección "bdproyectoa" de la base de datos "bdproyectoa" (debe existir en MongoDB) colMongo = bdMongo[coleccion] try: # INSERTAR DOCUMENTOS (Tareas) # Hacer el equivalente a un insert en bases de datos relacionales SQL # Insertamos cada documento (tarea) y mostramos el ID asociado en MongoDB print("Insertando documentos en MongoDB...") for tarea in tareas: resultadoInsertar = colMongo.insert_one(tarea.coleccionTareaMongoDB()) print(f"ID de documento insertado: {resultadoInsertar.inserted_id}") # MOSTRAR DOCUMENTOS (Tareas) # Hacer el equivalente a un Select en bases de datos relacionales SQL # Filtramos para mostrar sólo las tareas activas print("Mostrando documentos de colección con filtro en MongoDB...") curMongo = colMongo.find({"activa":True}) for tarea in curMongo: print (f"{tarea["nombre"]:20} {tarea["periodicidad"]:20} {tarea["accion"]}") # ACTUALIZAR DOCUMENTOS (Tareas) # Hacer el equivalente a un Update en bases de datos relacionales SQL # Filtramos para actualizar el valor de la periodicidad para las tareas activas=true print("Actualizando documentos de colección con filtro en MongoDB...") resultadoActualizar = colMongo.update_many( {"activa": True}, {"$set": {"periodicidad": "Anual"} }, upsert=True ) print(f"Documentos modificados: {resultadoActualizar.modified_count}") # ELIMINAR DOCUMENTOS (Tareas) # Hacer el equivalente a un Delete en bases de datos relacionales SQL # Filtramos para eliminar los documentos con periodicidad "Diaria" y activos = True print("Eliminando documentos de colección con filtro en MongoDB...") resultadoEliminar = colMongo.delete_many( {"activa": False, "periodicidad": "Diaria"} ) print(f"Documentos eliminados: {resultadoEliminar.deleted_count}") # Cerramos la conexión con el servidor MongoDB conMongo.close() except Exception as e: print(f"Error al realizar acción en MongoDB: {e}") except Exception as e: print(f"Error al conectar al sevidor MongoDB [{servidorMongoDB}]: {e}") |
Resultado de la ejecución del programa Python con acceso a MongoDB
A continuación mostramos el resultado de la ejecución del programa Python que accederá al servidor MongoDB y realizará las acciones indicadas anteriormente:
Por supuesto, si disponemos del cliente MongoDB Compass, también podremos ver los documentos insertados en la base de datos bdproyectoa y en la colección bdproyectoa:
