Servicio REST con Flask, SQLAlchemy y SQLite

by ·

Holas, el día de hoy les vengo a comentar como hacer un servicio REST con Flask, SQLALchemy como ORM y SQLite como base de datos.

SQLAlchemy es un ORM para el lenguaje python, el cual nos permite mapear clases hacia tablas de una base de datos y poder ejecutar operaciones en dicha base de datos através de las clases creadas para este trabajo.

El servicio de este post es una actualización hacia SQLAlchemy del servicio expuesto en el post de Servicio REST simple en Flask y utilizando como base de datos SQLite, sin mas preámbulo vamos a explicar como hacer nuestro servicio.

La estructura del proyecto es la misma que en el anterior servicio, así como el archivo run.py, pero nuestro archivo de requerimientos ha cambiado para incluir SQLAclhemy, admeas ahora también tenemos un archivo de configuración, el cual contiene valga la redundancia la configuración de donde se encuntra el archivo SQLite de nuestra base de datos.

requirements.txt

Flask
Flask-Cors
Flask-SQLAlchemy
Flask-Migrate

config.py

import os

basedir = os.path.abspath(os.path.dirname(__file__))

class Config(object):
    # Esta configuración nos permitirá crear un archivo llamado "to-do.db" en el mismo directorio de nuestro proyecto
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///' + os.path.join(basedir, 'to-do.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False

El archivo init.pydel paquete app también ha sido actualizado

from flask import Flask
from flask_cors import CORS
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy

# Importamos la clase Config del archivo config
from config import Config

app = Flask(__name__)
# Le decimos a nuestra aplicación flask que la
# configuración a usar es la que creamos
app.config.from_object(Config)
CORS(app)
# Añadimos SQLAlchemy a nuestra aplicación
db = SQLAlchemy(app)
migrate = Migrate(app, db)

from app.views import todo

Ahora viene la parte interesante, el esquema de nuestra base de datos hecho através de clases de python, para ello creamos dentro del paquete app un archivo llamado models.py con el siguiente código:

from app import db

# Esta clase contiene el modelo que va a ser mapeado en nuestra base de datos hacia una tabla llamada "to_do"
# Las clases de los modelos heredan de "db.Model"
class ToDo(db.Model):
    # Id es un entero, es primary key y por tanto se auto generará
    id = db.Column(db.Integer, primary_key=True)
    # Title es un string que permite máximo 100 caracteres, estará indexado y no se permite nulo
    title = db.Column(db.String(100), index=True, nullable=False)
    # Description es un string que permite máximo 400 caracteres, estará indexado y permite ser nulo
    description = db.Column(db.String(400), index=True, nullable=True)

    # Esta función permite retornar un diccionario con los datos
    # extraídos de la base de datos para luego poder convertirlos
    # en Json al momento de mostrar en el endpoint
    def json_dump(self):
        return dict(
            id=self.id,
            title=self.title,
            description=self.description
        )

    # Esta funcion tiene como propósito ayudarnos al momento de depurar
    def __repr__(self):
        return '<ToDo Title %r>' % self.title

Pues bien, ahora que tenemos la configuración de nuestro servicio, así como el modelo de la base de datos vamos a proceder con los endpoints. Los endpoints y su forma de usarlos son iguales que en nuestra anterior aplicación, pero internamnete sus operaciones son diferentes, es por eso que vamos a ver a continuación cómo funciona el código de los mismos para que puedan trabajar con SQLAclhmey y así tener persistencia de los datos.

# Ahora tenemos otros imports para poder usar SQLAlchemy
import json

from flask import abort, jsonify
from flask import request
from sqlalchemy.exc import IntegrityError

from app import app, db
from app.models import ToDo
from ..utils import json_utils

@app.route("/", methods=['GET'])
def get_all_todos():
    # Buscamos todos los to-do's en la base de datos
    all_todos = ToDo.query.all()
    # Retornamos dichos datos como un Json List
    # Como podemos ver ahora usamos jsonify para convertir los datos
    # en json, pero estos datos primero son convertidos al diccionario
    # que retorna la funcion json_dump() de la clase de nuestro modelo
    return jsonify([each_todo.json_dump() for each_todo in all_todos])

@app.route("/" + '<string:todo_id>', methods=['GET'])
def get_todo(todo_id):
    # Buscamos un to-do por su id
    selected_todo = ToDo.query.get(todo_id)
    # Si el valor buscado es None retornamos un error 404
    if selected_todo is None:
        abort(404)
    # Retornamos el dato buscado convertido en Json
    return jsonify(selected_todo.json_dump())

@app.route("/", methods=['POST'])
def post_todo():
    # Verificamos que nuestro request sea un Json valido
    # caso contrario retornamos un error 400
    json_utils.is_not_json_request(request)
    # El siguiente código esta envuelto en un bloque "try-except"
    # que nos permitirá devolver un error si los datos eviados
    # son inválidos
    try:
        # Extraemos el valor del campo title de nuestro request
        title = request.json.get('title')
        # Extraemos el valor del campo description de nuestro request
        description = request.json.get('description')
        # Creamos una nueva instancia del tipo "ToDo" con las variables
        # obtenidas arriba
        new_todo = ToDo(title=title, description=description)
        # Añadimos el nuevo to-do a una sesión de la base de datos
        db.session.add(new_todo)
        # Hacemos "commit" en la sesión de la base de datos
        # esto ejecutará la operación "INSERT" en la misma
        db.session.commit()
    except IntegrityError:
        # Si algo sale mal al momento de grabar vamos a retornar un error 400
        abort(400)
    # Retornamos el mismo objeto que enviamos, pero con el código 201 de creado
    return jsonify(request.json), 201

@app.route("/" + '<string:todo_id>', methods=['PUT'])
def put_todo(todo_id):
    # Verificamos que nuestro request sea un Json valido
    # caso contrario retornamos un error 400
   json_utils.is_not_json_request(request)
    # Buscamos un to-do por su id
    selected_todo = ToDo.query.get(todo_id)
    # Si el valor buscado es None retornamos un error 404
    if selected_todo is None:
        abort(404)
    # Extraemos los datos del request y los asignamos al to-do
    # buscado anteriormente
    selected_todo.title = request.json.get('title')
    selected_todo.description = request.json.get('description')
    try:
        # Hacemos un commit a la base de datos con la actualización
        # del to-do que buscamos arriba
        db.session.commit()
    except IntegrityError:
       # Si algo sale mal al momento de grabar vamos a retornar un error 400
        abort(400)
    return json.dumps(request.json)

@app.route("/" + '<string:todo_id>', methods=['DELETE'])
def delete_todo(todo_id):
    # Buscamos un to-do por su id
    selected_todo = ToDo.query.get(todo_id)
    # Si el valor buscado es None retornamos un error 404
    if selected_todo is None:
        abort(404)
    # Le decimos a la sesión de la base de datos que vamos a borrar un registro
    db.session.delete(selected_todo)
    try:
        # Hacemos commit a la session, esto ejecutará
        # la operación "DELETE" en la misma
        db.session.commit()
    except IntegrityError:
        # Si algo sale mal al momento de grabar vamos a retornar un error 400
        abort(400)
    # Retornamos vacío por que el dato ya fue borrado
    return ""

Como podemos ver los cambios que hemos realizado en nuestro código no son muy grandes, pero esto nos ayudará a tener persistencia de datos en una base SQLite, para así poder trabajar con ellos nuevamente en cualquier momento.

Ahora se pregutarán ¿cómo el archivo de la base to-do.db se va a crear? y ¿cómo la migración a la misma se va a realizar?. Pues bien, para eso es necesario seguir unos pasos previos en nuestro "deploy", los cuales voy a explicar a continuación:

  • Actualizar nuestro entorno virtual
  • Iniciar la base
  • Crear la migración
  • Migrar la base a nuestro archivo SQLite

Todas las operaciones a continuación serán ejecutadas dentro de la carpeta de nuestro proyecto

Actualizar nuestro entorno virtual

Como vimos nuestro archivo de requerimientos ahora contiene las dependencias para utilizar SQLAlchemy, para actualizarlo primero activamos nuestro entorno virtual como lo explicamos en Entornos virtuales en Python y luego ejecutamos el siguiente comando:

pip install -r requirements.txt

Iniciar la base

Para iniciar la base de datos ejecutamos el siguiente comando:

flask db init

Esto creará una carpeta llamada migrations con algunos archivos en su interior, dicha carpeta se creará dentro de la carpeta de nuestro proyecto

Crear la migración

Para crear la migración ejecutamos el siguiente comando:

flask db migrate

Esto creará un archivo python dentro de la carpeta versions que se encuentra dentro de migrations, además de crear el archivo to-do.db si es que no existe

El hecho de tener esta carpeta versions nos permite poder hacer upgrades o downgrades al esquema de nuestra base de datos cuando actualicemos el modelo en la clase ToDo o cuando agreguemos mas clases al modelo.

Migrar la base a nuestro archivo SQLite

Para migrar la base a nuestro archivo SQLite ejecutamos el siguiente comando:

flask db upgrade

Esto creará o actualizará las tablas dentro nuestro archivo to-do.db

Ahora para ejecutar nuestro proyecto lo hacemos de la misma manera que lo hemos hecho con el proyecto anterior ejecutando python run.py con nuesto entorno virtual activado.

Como nota extra, al usar los mismos endpoints que utilizamos en el proyecto anterior, el frontend explicado en el post de Consumo de una API REST con Vue.js seguirá funcionando de la misma manera.

La URL donde se encuentra este ejemplo es: https://github.com/nano-bytes/flask/tree/master/simple-todo-with-db

Eso es todo por hoy.

Happy hacking!!

Tags:

You may also like...