Integrando MongoDB con Flask usando Flask-PyMongo

En este tutorial, repasaremos todo lo que necesita saber para integrar MongoDB con Flask y Python mediante la creación de una aplicación CRUD funcional.

Introducción

Crear una aplicación web casi siempre significa tratar con datos de una base de datos. Hay varias bases de datos para elegir, dependiendo de su preferencia.

En este artículo, veremos cómo integrar una de las bases de datos NoSQL más populares, MongoDB, con el micro-marco Flask.

Hay varias extensiones Flask para integrar MongoDB, aquí usaremos la extensión Flask-PyMongo.

También trabajaremos en una API Todo-List simple para explorar las capacidades CRUD de MongoDB.

Instalación y configuración

Para seguir este tutorial, necesitará acceso a una instancia de MongoDB. Puede obtener una de Mongo DB Atlas o puede usar una instancia local. Usaremos una instancia local en nuestra propia máquina personal.

Para instalar una instancia local de MongoDB, diríjase a su sitio web de documentación oficial para obtener instrucciones sobre cómo descargarlo e instalarlo.

También necesitará tener Flask instalado, y si no lo tiene, puede hacerlo con el siguiente comando:

1
$ pip install flask

A continuación, debemos configurar Flask-PyMongo, que es un envoltorio alrededor del paquete python PyMongo.

PyMongo es un contenedor de bajo nivel alrededor de MongoDB, utiliza comandos similares a los comandos CLI de MongoDB para:

  1. Crear datos
  2. Acceder a los datos
  3. Modificación de datos

No utiliza ningún esquema predefinido, por lo que puede aprovechar al máximo la naturaleza sin esquema de MongoDB.

Para comenzar a usar Flask-PyMongo, debemos instalarlo con el siguiente comando.

1
$ pip install Flask-PyMongo

Ahora que estamos listos, comencemos a integrar MongoDB en nuestra aplicación Flask.

Conexión a una instancia de base de datos de MongoDB con Flask {#conexión a una instancia de base de datos de Godb con Flask}

Antes de realizar cualquier trabajo, queremos conectar nuestra instancia de MongoDB a la aplicación Flask. Comenzaremos importando Flask y Flask-PyMongo a nuestra aplicación:

1
2
from flask_pymongo import PyMongo
import flask

A continuación, crearemos un objeto de aplicación Flask:

1
app = flask.Flask(__name__)

Que luego usaremos para inicializar nuestro cliente MongoDB. PyMongo Constructor (importado de flask_pymongo) acepta nuestro objeto de aplicación Flsk y una cadena URI de base de datos.

Esto vincula nuestra aplicación a la instancia de MongoDB:

1
2
mongodb_client = PyMongo(app, uri="mongodb://localhost:27017/todo_db")
db = mongodb_client.db

La cadena URI también podría asignarse a la clave MONGO_URI en app.config

1
2
3
app.config["MONGO_URI"] = "mongodb://localhost:27017/todo_db"
mongodb_client = PyMongo(app)
db = mongodb_client.db

Una vez que la aplicación tiene una conexión con la instancia, podemos comenzar a implementar la funcionalidad CRUD de la aplicación.

Crear documentos: agregar nuevos elementos a la base de datos

MongoDB funciona con colecciones, que son análogas a la tabla SQL normal. Ya que estamos creando una aplicación de lista TODO, tendremos una colección todos. Para referenciarlo, usamos el objeto db. Cada entidad es un documento, y una colección es realmente una colección de documentos.

Para insertar una nueva entrada en nuestra colección todos, usamos el método db.colection.insert_one(). MongoDB funciona naturalmente con Python dada su sintaxis para inserción, consulta y eliminación.

Al insertar un documento en una colección de MongoDB, debe especificar un diccionario con <field>s y <value>s. Para insertar un documento en una colección de MongoDB usando Python como intermediario, pasará los diccionarios integrados en Python.

Por lo tanto, para insertar una nueva entidad, haremos algo como:

1
2
3
4
@app.route("/add_one")
def add_one():
    db.todos.insert_one({'title': "todo title", 'body': "todo body"})
    return flask.jsonify(message="success")

También podríamos agregar múltiples entradas a la vez usando el método db.colection.insert_many(). El método insert_many() toma una lista de diccionarios y los agrega a la colección:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
@app.route("/add_many")
def add_many():
    db.todos.insert_many([
        {'_id': 1, 'title': "todo title one ", 'body': "todo body one "},
        {'_id': 2, 'title': "todo title two", 'body': "todo body two"},
        {'_id': 3, 'title': "todo title three", 'body': "todo body three"},
        {'_id': 4, 'title': "todo title four", 'body': "todo body four"},
        {'_id': 5, 'title': "todo title five", 'body': "todo body five"},
        {'_id': 1, 'title': "todo title six", 'body': "todo body six"},
        ])
    return flask.jsonify(message="success")

Si intentamos agregar un registro duplicado, se lanzará un BulkWriteError, lo que significa que solo se insertarán los registros hasta dicho duplicado, y todo lo que esté después del duplicado se perderá, así que tenga esto en cuenta cuando intente insertar muchos documentos.

Si queremos insertar solo registros válidos y únicos en nuestra lista, tendremos que establecer el parámetro ordered del método insert_many() en false y luego detectar la excepción BulkWriteError:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from pymongo.errors import BulkWriteError

@app.route("/add_many")
def add_many():
    try:
        todo_many = db.todos.insert_many([
            {'_id': 1, 'title': "todo title one ", 'body': "todo body one "},
            {'_id': 8, 'title': "todo title two", 'body': "todo body two"},
            {'_id': 2, 'title': "todo title three", 'body': "todo body three"},
            {'_id': 9, 'title': "todo title four", 'body': "todo body four"},
            {'_id': 10, 'title': "todo title five", 'body': "todo body five"},
            {'_id': 5, 'title': "todo title six", 'body': "todo body six"},
        ], ordered=False)
    except BulkWriteError as e:
        return flask.jsonify(message="duplicates encountered and ignored",
                             details=e.details,
                             inserted=e.details['nInserted'],
                             duplicates=[x['op'] for x in e.details['writeErrors']])

    return flask.jsonify(message="success", insertedIds=todo_many.inserted_ids)

Este enfoque insertará todos los documentos válidos en la colección MongoDB. Además, registrará los detalles de las adiciones fallidas y se las devolverá al usuario, como un mensaje JSON.

Lo hemos hecho a través del método jsonify() de Flasks, que acepta un mensaje que deseamos devolver, así como parámetros adicionales que nos permiten personalizarlo para fines de registro.

Finalmente, devolvemos las inserciones exitosas, de la misma manera.

Lectura de documentos: recuperación de datos de la base de datos

Flask-PyMongo proporciona varios métodos (extendidos de PyMongo) y algunos métodos auxiliares para recuperar datos de la base de datos.

Para recuperar todos los documentos de la colección todos, usaremos el método db.collection.find().

Este método devolverá una lista de todos los todos en nuestra base de datos. Similar a find(), el método find_one() devuelve un documento, dado su ID.

Empecemos con find():

1
2
3
4
@app.route("/")
def home():
    todos = db.todos.find()
    return flask.jsonify([todo for todo in todos])

El método find() también puede tomar un parámetro de filtro opcional. Este parámetro de filtro se representa con un diccionario que especifica las propiedades que estamos buscando. Si ha trabajado con MongoDB anteriormente, probablemente esté familiarizado con el aspecto de sus consultas y comparadores.

Si no, así es como podemos usar el diccionario de Python para acomodar el formato de consulta de MongoDB:

1
2
3
4
5
6
7
8
9
# Query document where the `id` field is `3`
{"id":3}

# Query document where both `id` is `3` and `title` is `Special todo`
{"id":3, "title":"Special todo"}

# Query using special operator - Greater than Or Equal To, denoted with
# the dollar sign and name ($gte)
{"id" : {$gte : 5}}

Algunos otros operadores especiales incluyen los operadores $eq, $ne, $gt, $lt, $lte y $nin.

Si no está familiarizado con estos, un excelente lugar para obtener más información sobre ellos es la [documentación oficial] (https://docs.mongodb.com/manual/reference/operator/query/).

Ahora que hemos cubierto la especificación de consultas MongoDB para filtrar el método find(), echemos un vistazo a cómo recuperar un documento, dado su _id:

1
2
3
4
@app.route("/get_todo/<int:todoId>")
def insert_one(todoId):
    todo = db.todos.find_one({"_id": todoId})
    return todo

Entonces, si enviáramos una solicitud GET a http://localhost:5000/get_todo/5, obtendríamos el siguiente resultado:

1
2
3
4
5
{
    "_id": 5,
    "body": "todo body six",
    "title": "todo title six"
}

Tenga en cuenta que 5000 es el puerto de servidor Flask predeterminado, pero se puede cambiar fácilmente al crear un objeto de aplicación Flask

La mayoría de las veces querríamos obtener un artículo o devolver un error 404 si no se encuentra el artículo.

Flask-PyMongo proporciona una función auxiliar para esto, el método find_one_or_404() que generará un error 404 si no se encuentra el recurso solicitado.

Actualizar y reemplazar documentos

Para actualizar entradas en nuestra base de datos, podemos usar el método update_one() o replace_one() para cambiar el valor de una entidad existente.

replace_one() tiene los siguientes argumentos:

  1. filtro: una consulta que define qué entradas se reemplazarán.
  2. reemplazo - Entradas que se colocarán en su lugar cuando se reemplacen.
  3. {}: un objeto de configuración que tiene algunas opciones, de las cuales nos centraremos en: upsert.

upsert, cuando se establece en true insertará replacement como un nuevo documento si no hay coincidencias de filtro en la base de datos. Y si hay coincidencias, entonces coloca reemplazo en su lugar. Si upsert es falso e intenta actualizar un documento que no existe, no pasará nada.

Echemos un vistazo a cómo podemos actualizar los documentos:

1
2
3
4
5
6
7
8
9
@app.route("/replace_todo/<int:todoId>")
def replace_one(todoId):
    result = db.todos.replace_one({'_id': todoId}, {'title': "modified title"})
    return {'id': result.raw_result}

@app.route("/update_todo/<int:todoId>")
def update_one(todoId):
    result = db.todos.update_one({'_id': todoId}, {"$set": {'title': "updated title"}})
    return result.raw_result

Entonces, si enviáramos una solicitud a http://localhost:5000/update_todo/5, obtendríamos el siguiente resultado:

1
2
3
4
5
6
7
8
{
    "id": {
        "n": 1,
        "nModified": 1,
        "ok": 1.0,
        "updatedExisting": true
    }
}

De manera similar, si también enviáramos una solicitud a http://localhost:5000/replace_todo/5, obtendríamos el siguiente resultado:

1
2
3
4
5
6
7
8
{
    "id": {
        "n": 1,
        "nModified": 1,
        "ok": 1.0,
        "updatedExisting": true
    }
}

El bloque de código devolverá un objeto UpdatedResult, que puede ser un poco tedioso para trabajar. Es por eso que Flask-PyMongo proporciona métodos más convenientes como find_one_and_update() y find_one_and_replace(), que actualizarán una entrada y devolverán esa entrada:

1
2
3
4
5
6
7
8
9
@app.route("/replace_todo/<int:todoId>")
def replace_one(todoId):
    todo = db.todos.find_one_and_replace({'_id': todoId}, {'title': "modified title"})
    return todo

@app.route("/update_todo/<int:todoId>")
def update_one(todoId):
    result = db.todos.find_one_and_update({'_id': todoId}, {"$set": {'title': "updated title"}})
    return result

Ahora, si enviáramos una solicitud a http://localhost:5000/update_todo/5, obtendríamos el siguiente resultado:

1
2
3
4
{
    "_id": 5,
    "title": "updated title"
}

De manera similar, si también enviáramos una solicitud a http://localhost:5000/replace_todo/5, obtendríamos el siguiente resultado:

1
2
3
4
{
    "_id": 5,
    "title": "modified title"
}

Flask-PyMongo también permite actualizaciones masivas con el método update_many():

1
2
3
4
@app.route('/update_many')
def update_many():
    todo = db.todos.update_many({'title' : 'todo title two'}, {"$set": {'body' : 'updated body'}})
    return todo.raw_result

El bloque de código anterior encontrará y actualizará todas las entradas con el título "todo el título dos" y dará como resultado:

Enviar una solicitud a nuestros puntos recién creados devuelve el siguiente resultado:

1
2
3
4
5
6
{
    "n": 1,
    "nModified": 1,
    "ok": 1.0,
    "updatedExisting": true
}

Eliminación de documentos

Al igual que con los demás, Flask-PyMongo proporciona métodos para eliminar una sola o una colección de entradas usando los métodos delete_one() y delete_many() respectivamente.

Los argumentos de este método son los mismos que con los otros métodos. Veamos un ejemplo:

1
2
3
4
@app.route("/delete_todo/<int:todoId>", methods=['DELETE'])
def delete_todo(todoId):
    todo = db.todos.delete_one({'_id': todoId})
    return todo.raw_result

Esto buscará y eliminará la entrada con la ID proporcionada. Si enviamos una solicitud DELETE como http://localhost:5000/delete_todo/5 a este punto final, obtendríamos el siguiente resultado:

1
2
3
4
{
    "n": 1,
    "ok": 1.0
}

Alternativamente, puede usar el método find_one_and_delete() que elimina y devuelve el elemento eliminado, para evitar usar el objeto de resultado poco práctico:

1
2
3
4
5
6
@app.route("/delete_todo/<int:todoId>", methods=['DELETE'])
def delete_todo(todoId):
    todo = db.todos.find_one_and_delete({'_id': todoId})
    if todo is not None:
        return todo.raw_result
    return "ID does not exist"

Enviar http://localhost:5000/delete_todo/8 a nuestro servidor ahora da como resultado:

1
2
3
4
5
{
    "_id": 8,
    "body": "todo body two",
    "title": "todo title two"
}

Finalmente, puedes eliminar en masa, usando el método delete_many():

1
2
3
4
@app.route('/delete_many', methods=['DELETE'])
def delete_many():
    todo = db.todos.delete_many({'title': 'todo title two'})
    return todo.raw_result

Enviar http://localhost:5000/delete_many a nuestro servidor resultará en algo similar a:

1
2
3
4
{
    "n": 1,
    "ok": 1.0
}

Guardar y recuperar archivos

MongoDB nos permite guardar datos binarios en su base de datos usando la especificación GridFS.

Flask-PyMongo proporciona el método save_file() para guardar un archivo en GridFS y el método send_file() para recuperar archivos de GridFS.

Comencemos con una ruta para subir un archivo a GridFS:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
@app.route("/save_file", methods=['POST', 'GET'])
def save_file():
    upload_form = """<h1>Save file</h1>
                     <form method="POST" enctype="multipart/form-data">
                     <input type="file" name="file" id="file">
                     <br><br>
                     <input type="submit">
                     </form>"""
                     
    if request.method=='POST':
        if 'file' in request.files:
            file = request.files['file']
            mongodb_client.save_file(file.filename, file)
            return {"file name": file.filename}
    return upload_form

En el bloque de código anterior, creamos un formulario para manejar las cargas y devolver el nombre de archivo del documento cargado.

A continuación, veamos cómo recuperar el archivo que acabamos de cargar:

1
2
3
@app.route("/get_file/<filename>")
def get_file(filename):
    return mongodb_client.send_file(filename)

Este bloque de código devolverá el archivo con el nombre de archivo dado o generará un error 404 si no se encontró el archivo.

Conclusión

La extensión Flask-PyMongo proporciona una API de bajo nivel (muy similar al lenguaje oficial de MongoDB) para comunicarse con nuestra instancia de MongoDB.

La extensión también proporciona varios métodos de ayuda para que podamos evitar tener que escribir demasiado código repetitivo.

En este artículo, hemos visto cómo integrar MongoDB con nuestra aplicación Flask, también hemos realizado algunas operaciones CRUD y hemos visto cómo trabajar con archivos con MongoDB usando GridFS.

He tratado de cubrir todo lo que he podido, pero si tiene alguna pregunta y/o contribución, deje un comentario a continuación.