Crear una API REST con Flask

En el mundo de las API Rest con Python se hace un uso exhaustivo de la librería SQLAlchemy, para realizar operaciones sobre varias bases de datos, incluyendo MySQL. En este artículo vamos a crear una API Rest con Flask, Python y MySQL, manejandolo directamente con el módulo pymysql, sin necesidad de SQLAlchemy.Vamos a dividir este ejercicio didáctico en dos partes: El lado del servidor y el lado del cliente. Comenzamos con el lado del servidor definiendo la base de datos con MySQL y los endpoints con el microframework Flask.

API Rest – Lado del Servidor

Comenzamos nombrando la base de datos como api_test. Esta base de datos tendrá una sola tabla llamada user, y sus campos son los siguientes: id, username, email.

Vamos a crear un archivo llamado bd.py para crear nuestra conexión a la base de datos:

import pymysql

def conexion():

    returnpymysql.connect(host='localhost', user='erwinux', password='password', db='test_api')

Creamos ahora un archivo llamado modelo.py, que contendrá los querys y sentencias para las cuatro operaciones básicas tipo CRUD en nuestra BD:

from bd import conexion

def obtener_registros():
    con = conexion()
    registros = []
    withcon.cursor() ascursor:
        cursor.execute("SELECT * FROM user")
    registros = cursor.fetchall()
    con.close()

    returnregistros


def obtener_x_id(id):
    con = conexion()
    registros = []
    withcon.cursor() ascursor:
        cursor.execute("SELECT * FROM user WHERE id=%s", id)
    registros = cursor.fetchone()
    con.close()
    returnregistros


def creacion_usuario(nombre, correo):
    con = conexion()
    withcon.cursor() ascursor:
        cursor.execute("INSERT INTO user(username, email) VALUES(%s, %s)", (nombre, correo))
    con.commit()
    con.close()

def modifica_usuario(id, nombre, correo):
    con = conexion()
    withcon.cursor() ascursor:
        cursor.execute("UPDATE user SET username=%s, email=%s WHERE id=%s", (nombre, correo, id))
    con.commit()
    con.close()

def elimina_usuario(id):
    con = conexion()
    with con.cursor() ascursor:
        cursor.execute("DELETE FROM user WHERE id=%s", id)
    con.commit()
    con.close() 

Y ahora en nuestro archivo main.py definimos los endpoints para cada una de las operaciones CRUD. Comenzamos con la operación Read, la de consultar todos los registros de la tabla user:

from flask import Flask, request, jsonify
import modelo
app = Flask(__name__)

# Enpoint Muestra todos los Usuarios
@app.route("/user", methods=["GET"])
def get_user():
    resultados = []
    result = modelo.obtener_registros()
    claves = ['id', 'nombres', 'correo']
    for objetos in result:
        list2dic = dict(zip(claves, objetos)) #convertimos la tupla result y la lista claves en un diccionario
        resultados.append(list2dic)

    return jsonify({"data":resultados})

if __name__ == '__main__':
    withapp.app_context():
    app.run(host="localhost", port="5000", debug=True)

En la variable list2dic almacenamos el diccionario recién convertido entre las tuplas provenientes del query, y la lista clave. Esto lo hacemos para proporcionar una respuesta en formato JSON (clave – valor) al cliente.

Pero como aún no tenemos registros en nuestra base de datos, vamos a definir el endpoint con el que cliente registrará datos en la tabla user:

@app.route("/user", methods=["POST"])
def add_user():
    username = request.json['username']
    email = request.json['email']
    modelo.creacion_usuario(username, email)

    User_dict = {
        "username": username,
        "email": email
    }

    return jsonify(User_dict)

Para realizar las pruebas de ambos endpoints, basta con usar los siguientes comandos de curl:

curl -X POST -H "Content-Type: application/json" -d '{"username":"Colacho", "email":"colacho@yahoo.es"}' http://localhost:5000/user
curl -X GET http://localhost:5000/user

API Rest – Lado del Cliente

En un archivo llamado apirest.py, vamos a definir las solicitudes y lo métodos que enviaremos al lado del servidor, donde está la base de datos api_test:

import requests
import json

url = 'http://localhost:5000/user'

#OBTENER TODOS LOS DATOS
def obtener_todos():
    r = requests.get(url)
    r = r.json()

    return r


#CREAR UN USUARIO
def crear_usuario():
    user = {
        'username' : 'Pedro',
        'email' : 'pedro@gmail.com'
    }

    headers = {
        'Content-Type': 'application/json'
    }

    response = requests.post(url, data=json.dumps(user), headers=headers)
    print_response(response)


def print_response(response):
print(response.text)
print(response.json())
print(response.status_code)
print(response.headers)


#print(crear_usuario())
#print(obtener_todos())

Dentro de la función crear_usuario() hemos ‘quemado’ el objeto user con el nombre Pedro y su correo electrónico pedro@gmail.com, luego definimos el encabezado en el diccionario headers y finalmente enviamos el objeto al servidor con la línea:

requests.post(url, data=json.dumps(user), headers=headers)

Claro, debemos descomentar la línea print(crear_usuario()) para que funcione.

Esta función viene siendo el reemplazo del comando

curl -X POST -H "Content-Type: application/json" -d '{"username":"Colacho", "email":"colacho@yahoo.es"}' http://localhost:5000/user

Podemos cambiar las propiedades del objeto user y enviarlas al servidor para registrarlas en la BD.

Para consultar los registros de la base de datos, usamos la función obtener_todos() y descomentamos la línea print(obtener_todos()). y obtendremos algo como esto:

Esta función viene siendo el reemplazo del comando:

curl -X GET http://localhost:5000/user

Los archivos

Te comparto los archivos completos main.py y apirest.py con el resto de funciones.

https://espaciotecnologico.co/main.py
https://espaciotecnologico.co/apirest.py

Valora este artículo