notifications Notificaciones

Marcar todas como leídas

Ver más

lightbulb_outline

227 veces ha sido leído este artículo

¿Cómo crear tu propia API con flask?

Lo lees en 3 Min.

En esta ocasión aprenderemos a crear nuestro propio servicio RESTFul con Flask, unos de los micro frameworks más populares de Python.

Para el tutorial estaré trabajando con el gestor de base de datos PostreSQL, sin embargo tú puedes utilizar el gestor que desees.

Bien, sin más dilación comencemos. 😎

Dependencias

Lo primero que haremos será instalar las bibliotecas necesarias. Para esto recomiendo te encuentres en un entorno virtual. 🤭

python3 -m venv env

Las dependecias necesarías serán: Flask, psycopg2 y SQLAlchemy.

pip install flask
pip install psycopg2-binary
pip install Flask-SQLAlchemy

SQLAlchemy es un ORM el cual nos permite trabajar con una base de datos sin tener que conocer y/o dominar SQL.

API

El primer paso será definir los endpoints de nuestra API, en mi caso cinco endpoints. 🤠

main.py

from flask import Flask
from flask import jsonify

def create_app(enviroment):
    app = Flask(__name__)
    return app

app = create_app()

app.route('/api/v1/users', methods=['GET'])
def get_users():
    response = {'message': 'success'}
    return jsonify(response)

app.route('/api/v1/users/<id>', methods=['GET'])
def get_user(id):
    response = {'message': 'success'}
    return jsonify(response)

@app.route('/api/v1/users/', methods=['POST'])
def create_user():
    response = {'message': 'success'}
    return jsonify(response)

@app.route('/api/v1/users/<id>', methods=['PUT'])
def update_user(id):
    response = {'message': 'success'}
    return jsonify(response)

@app.route('/api/v1/users/<id>', methods=['DELETE'])
def delete_user(id):
    response = {'message': 'success'}
    return jsonify(response)

if __name__ == '__main__':
    app.run(debug=True)

Para probar nuestro API podemos utilizar cURL, una herramienta nativa para sistemas operativos unix o podemos utilizar POSTMAN.

El siguiente paso será definir nuestro archivo config.py, archivo donde se encontrarán las configuraciones del proyecto.

class Config:
    pass

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/users'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

config = {
    'development': DevelopmentConfig,
}

En este caso coloco la dirección de la base de datos directamente en el archivo, en texto plano. Esto lo hago solo por fines prácticos, yo te recomiendo utilices una variable de entorno.

Importamos nuestra configuración.

main.py

from config import config

def create_app(enviroment):
    app = Flask(__name__)

    app.config.from_object(enviroment)

    return app

enviroment = config['development']
app = create_app(enviroment)

Ahora procedemos a crear nuestro modelo. En este caso un modelo User. Para ello trabajaremos en el archivo models.py.

from datetime import datetime
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), nullable=False) 
    created_at = db.Column(db.DateTime(), nullable=False, default=db.func.current_timestamp())


Un modelo bastante sencillo, sin embargo para fines practicos funcionará.

El siguiente paso será crear nuestra tabla y asociar la base de datos con la aplicación de Flask.

from models import db

def create_app(enviroment):
    app = Flask(__name__)

    app.config.from_object(enviroment)

    with app.app_context():
        db.init_app(app)
        db.create_all()

    return app

Antes de ejecutar el proyecto es necesario que crees la base de datos en tu servidor. 🤖

CREATE DATABASE users;

Listo, ya hicimos lo más complicado, ahora toca dar funcionalidad a cada uno de los endpoints. Comencemos con crear.

from models import User

@app.route('/api/v1/users/', methods=['POST'])
def create_user():
    json = request.get_json(force=True)

    if json.get('username') is None:
        return jsonify({'message': 'Bad request'}), 400

    user = User.create(json['username'])

    return jsonify({'user': user.json() })

En este caso creamos un método de clase (create) en nuestro modelo User para instanciar y almace nuestro usuario. El método encargado de persistir es save.

@classmethod
def create(cls, username):
        user = User(username=username)
        return user.save()

def save(self):
        try:
                db.session.add(self)
                db.session.commit()

                return self
        except:
                return False

De igual forma creamos un método de instancia json para serializar nuestro objeto.

def json(self):
        return {
            'id': self.id,
            'username': self.username,
            'created_at': self.created_at
        }

Procedemos a obtener a los usuarios, algo muy sencillo con SQLAlchemy. ☺


@app.route('/api/v1/users', methods=['GET'])
def get_users():
    users = [ user.json() for user in User.query.all() ] 
    return jsonify({'users': users })

@app.route('/api/v1/users/<id>', methods=['GET'])
def get_user(id):
    user = User.query.filter_by(id=id).first()
    if user is None:
        return jsonify({'message': 'User does not exists'}), 404

    return jsonify({'user': user.json() })

El siguiente paso será actualizar un usuario.

user = User.query.filter_by(id=id).first()
if user is None:
        return jsonify({'message': 'User does not exists'}), 404

json = request.get_json(force=True)
if json.get('username') is None:
        return jsonify({'message': 'Bad request'}), 400

user.username = json['username']

user.update()

return jsonify({'user': user.json() })

Tendro del modelo User generamos un nuevo método.

def update(self):
        self.save()

En este caso el endpoint para editar es, quizás, el más complejo; como podemos observar tenemos código duplicado, sinónimos que podemos mejorar aún más el proyecto. 🤪

y finalmente pasamos a eliminar a un usuario.

@app.route('/api/v1/users/<id>', methods=['DELETE'])
def delete_user(id):
    user = User.query.filter_by(id=id).first()
    if user is None:
        return jsonify({'message': 'User does not exists'}), 404

    user.delete()

    return jsonify({'user': user.json() })

Creamos un método de instancia en nuestro modelo User.

def delete(self):
        try:
                db.session.delete(self)
                db.session.commit()

                return True
        except:
                return False

De esta forma es como quedaría, por el momento, mi API. Aunque tenemos código duplicado podemos llegar a solucionarlo mediante decoradores.

En Flask nuestros decoradores tendrán la siguiente estructura.

def decorator_name(function):
    def wrap(*args, **kwargs):
        return function(*args, **kwargs)

    wrap.__name__ = function.__name__
    return wrap

Su implementació depende de ti. 😎


Bien, de esta forma estaríamos terminando el tutorial en esta ocación. Si te interesa desplegar tu aplicación FLASK en un servidor de producción te recomiendo le heche un vistazo al siguiente post.

Recomendaciones

Comunidad