Inicio de un proyecto de desarrollo de API (con Flask)

🎥 Here's a video tutorial about creating Flask API's using this boilerplate.

How to Start coding?

Empezando por el boilerplate flast-rest-hello, puedes encontrar un ejemplo de API trabajando con una base de datos. Todo el código de tu aplicación debe estar escrito dentro de la carpeta ./src/.

• src/app.py: Es donde la aplicación se inicializa, en APIs pequeñas también puedes codificar tus diferentes endpoints aquí, por favor ten en cuenta que codificar los endpoints aquí sólo se recomienda si no hay un archivo routes.py en el proyecto.
• src/routes.py (opcional): Si tu proyecto tiene un fichero src/routes.py, aquí es donde debes codificar para añadir tus endpoints.
• src/models.py: Tus tablas de base de datos y lógica de serialización.
• src/utils.py: Algunas clases y funciones reutilizables.
• src/admin.py: Añade tus modelos al admin y gestiona tus datos fácilmente.

Para una explicación más detallada, busca el tutorial dentro de la carpeta docs.

Instalación en Ubuntu y Mac

⚠️ Asegúrese de que tiene python 3.6+ y MySQL instalado en su ordenador y MySQL se está ejecutando, a continuación, ejecute los siguientes comandos:

1pipenv install #(to install pip packages)
2pipenv run migrate #(to create the database)
3pipenv run start #(to start the flask webserver)

Añadir un endpoint

Para cada endpoint necesitará tener:

1. Un decorador @app que especifique la ruta para el endpoint.
   • Puedes tener parámetros en la URL como este <int:person_id>.
   • Puedes especificar qué métodos pueden ser invocados en ese endpoint como methods=['PUT', 'GET'].
2. El método que se ejecutará cuando se llame a ese endpoint (en este caso get_single_person).
3. Dentro del método puedes especificar qué lógica ejecutar de cada tipo de petición usando if request.method == 'PUT'.
4. Tienes que devolver siempre un json y un código de estado (200, 400, 404, etc.)

1from sqlalchemy import select
2from sqlalchemy.exc import NoResultFound
3
4
5@app.route('/person/<int:person_id>', methods=['PUT', 'GET'])
6def get_single_person(person_id):
7    """A person (Updated: Using session.execute)"""
8    body = request.get_json() #{ 'username': 'new_username'}
9    try:
10        stmt = select(Person).where(Person.id == person_id)
11        result = db.session.execute(stmt)
12        user1 = result.scalars().first()
13        
14        if user1 is None:
15            return jsonify({'error': 'Person not found'}), 404
16
17        if request.method == 'PUT':
18            user1.username = body['username']
19            db.session.commit()
20            return jsonify(user1.serialize()), 200
21        elif request.method == 'GET':
22            return jsonify(user1.serialize()), 200
23
24    except NoResultFound:
25        return jsonify({'error': 'Person not found'}), 404
26    except Exception as e:
27        return jsonify({'error': str(e)}), 500
28
29    return jsonify({'error': 'Invalid method'}), 405

Cómo validar la carga útil de la solicitud o la cadena de consulta

---

Digamos que una petición viene del cliente y necesitamos asegurarnos de que contiene la información correcta.

Tenemos que usar condicionales para hacer las validaciones, si queremos validar el cuerpo de la petición tenemos que recuperarlo primero y luego añadir la condición, así:

1body = request.get_json()
2if 'username' not in body:
3    raise APIException('You need to specify the username', status_code=400)

• Es una buena práctica lanzar excepciones porque detendrá la ejecución del código.
• Es una buena práctica devolver 400 porque así el cliente sabe que fue su error y no el nuestro (el servidor).

He aquí un ejemplo completo de una solicitud POST para añadir una nueva persona a una base de datos

1@app.route('/person', methods=['POST'])
2def handle_person():
3
4    # Primero obtenemos el payload json
5    body = request.get_json()
6
7    if body is None:
8        raise APIException("You need to specify the request body as a json object", status_code=400)
9    if 'username' not in body:
10        raise APIException('You need to specify the username', status_code=400)
11    if 'email' not in body:
12        raise APIException('You need to specify the email', status_code=400)
13
14    # en este punto, todos los datos han sido validados, podemos proceder a insertar en la base de datos
15    user1 = Person(username=body['username'], email=body['email'])
16    db.session.add(user1)
17    db.session.commit()
18    return "ok", 200

La base de datos

El boilerplate de Flask viene con una base de datos Postgres instalada y funcionando, tómate 6 min para ver este video sobre Postgres.

También usamos SQLAlchemy para abstraer nuestra base de datos, eso significa que no tienes que escribir SQL para tratar con tu base de datos, todo se hará usando python.

Añadir, Actualizar y Borrar datos

Ya que no vamos a usar SQL directamente, en su lugar vamos a trabajar con SQLAlchemy usando Python.

Pulsa en este enlace para ver un tutorial más detallado sobre cómo trabajar con SQLAlchemy y Postgres para CRUD (Crear, Leer, Actualizar y Borrar datos).

Flask Admin

Cualquier API desarrollada usando este boilerplate vendrá con una rápida y sencilla UI llamada: Flask Admin.

El flask admin te permite rápidamente ver, añadir, borrar o actualizar información de las tablas de tu base de datos.

Puede acceder a su administrador de flask añadiendo /admin al final de su API Host, por ejemplo:

Si su API host es https://8000-blabla-us33.gitpod.io entonces usted puede acceder a su base de datos admin así: https://8000-blabla-us33.gitpod.io/admin

Aquí hay un video de 8 minutos explicando el Flask Admin.

Añadiendo tus modelos a tu admin de Flask

Con sólo un par de líneas de código puedes integrar tu modelo en el admin de Flask, por ejemplo, si tienes un modelo Car puedes añadir el modelo en el admin así:

1from models import Car
2...
3admin.add_view(ModelView(Car, db.session))

Pero tienes que añadir esas dos líneas dentro del archivo admin.py así:

1from flask_admin.contrib.sqla import ModelView
2from flask_admin import Admin
3from models import db, Car # < ------ Importar el modelo
4
5def setup_admin(app):
6    admin = Admin(app, name='your_admin_name', template_mode='bootstrap3')
7    admin.add_view(ModelView(Car, db.session)) # < ------ Add the model to the admin

Puedes añadir tantos modelos como quieras, así:

1from flask_admin.contrib.sqla import ModelView
2from flask_admin import Admin
3from models import db, Car, Person, Patient # < ------ Importar el modelo
4
5def setup_admin(app):
6    admin = Admin(app, name='your_admin_name', template_mode='bootstrap3')
7    admin.add_view(ModelView(Car, db.session)) # < ------ Añadir el modelo al admin
8    admin.add_view(ModelView(Person, db.session)) # < ------ Añadir el modelo al admin
9    admin.add_view(ModelView(Patient, db.session)) # < ------ Añadir el modelo al admin

Migraciones

Los cambios en la base de datos se rastrean mediante migraciones alembic. Debe migrar y actualizar las migraciones para cada actualización que realice en sus modelos y que deba reflejarse en la estructura de las tablas:

1pipenv run migrate #(para realizar las migraciones)
2pipenv run upgrade #(para actualizar su base de datos con las migraciones)

Reiniciar Migraciones

A veces la carpeta de migraciones puede estropearse, es realmente difícil arreglar algunos de los problemas y, dado que todavía estamos en desarrollo, tiene sentido resetear toda la base de datos y las migraciones.

Este boilerplate contiene un script que puede ayudarte a resetear desde cero toda tu base de datos en caso de que lo necesites. Para ejecutarlo ejecute pipenv run reset_db, esto borrará toda su base de datos y la reconstruirá desde cero, perdiendo todos los datos en el proceso. Estas son las acciones que realiza el script:

⚠️ Advertencia: sus datos se perderán

1. Borre toda la carpeta de migraciones rm -R -f ./migrations.
2. Crea una nueva carpeta de migraciones para flask flask db init.
3. Eliminar la base de datos dropdb -h localhost -U gitpod ejemplo.
4. Crear de nuevo la base de datos createdb -h localhost -U gitpod ejemplo";.
5. Crear la extensión 'inaccent' psql -h localhost ejemplo -U gitpod -c 'CREATE EXTENSION unaccent;'
6. Crea de nuevo los archivos de migración: pipenv run migrate.
7. 1. Aplique los archivos de migración a su base de datos: pipenv run upgrade >

⚠️ Nota: Por favor, recuerde que todos sus datos se perderán.

Codificación de una operación CRUD típica

Como ejemplo, vamos a crear una pequeña API para gestionar una Persona.

Añadir Modelos

Para cada model tendrás que declarar una clase con las propiedades del modelo y un método serialize que devuelva una representación de diccionario de la clase:

1from sqlalchemy import String
2from sqlalchemy.orm import Mapped, mapped_column
3
4
5class Person(db.Model):
6    id: Mapped[int] = mapped_column(primary_key=True)
7    username: Mapped[str] = mapped_column(unique=True)
8    email: Mapped[str] = mapped_column(String(120), unique=True, nullable=False)
9
10    # indica a python cómo convertir el objeto de clase en un diccionario listo para jsonificar
11    def serialize(self):
12        return {
13            "username": self.username,
14            "email": self.email
15        }

📝 Encontrará más información en creación de modelos aquí.

Operaciones CRUD

Hay muchas maneras de manipular bases de datos, pero hemos decidido utilizar Python y SQLAlchemy para hacerlo. Esto significa que no necesitas conocimientos de SQL, pero te recomendamos encarecidamente que aún así practiques y domines SQL con fines de depuración (la mayoría de los errores se muestran en lenguaje SQL)

Consulta (SELECT) de datos

Asumiendo que tienes un objeto Person en tu fichero models.py.

1# coger a toda la gente
2stmt = select(Person)
3people_query = db.session.execute(stmt).scalars().all()
4
5# obtener sólo las que se llamen "Joe
6stmt = select(Person).where(Person.name == 'Joe')
7people_query = db.session.execute(stmt).scalars().all()
8
9# asigna los resultados y tu lista de personas dentro de la variable all_people
10all_people = list(map(lambda x: x.serialize(), people_query))
11
12# obtener sólo una persona
13stmt = select(Person).where(Person.id == person_id)
14user1 = db.session.execute(stmt).scalars().first()

Insertar datos

Asumiendo que tienes un objeto Persona en tu fichero models.py.

1user1 = Person(username="my_super_username", email="my_super@email.com")
2db.session.add(user1)
3db.session.commit()

Actualización de datos

1stmt = select(Person).where(Person.id == person_id)
2user1 = db.session.execute(stmt).scalars().first()
3
4if user1 is None:
5    raise APIException('User not found', status_code=404)
6
7if "username" in body:
8    user1.username = body["username"]
9if "email" in body:
10    user1.email = body["email"]
11db.session.commit()

Borrar datos

1stmt = select(Person).where(Person.id == person_id)
2user1 = db.session.execute(stmt).scalars().first()
3
4if user1 is None:
5   raise APIException('User not found', status_code=404)
6db.session.delete(user1)
7db.session.commit()

¿Cómo implementar un esquema JWT en mi API con Flask?

En éste artículo encontrarás todos los detalles de cómo implementar este esquema en tu API Flask

Despliegue

Esta plantilla es 100% compatible con Heroku y Render.com, sólo asegúrate de leer las guías de despliegue rápido.

1. Cómo desplegar en Render.com (gratis)
2. Cómo desplegar en Heroku.com (desde $0.01 mensual)