Construir una API REST de Flask con SQLite

Introducción

En este proyecto, aprenderemos cómo construir una API REST en Flask utilizando SQLite como base de datos. Comenzaremos creando una base de datos SQLite de muestra y llenándola con algunos datos. Luego, construiremos una aplicación Flask con endpoints para realizar operaciones CRUD (Crear, Leer, Actualizar, Eliminar) en los datos de la base de datos SQLite.

👀 Vista previa

🎯 Tareas

En este proyecto, aprenderás:

• Cómo crear una aplicación Flask con SQLite como backend de la base de datos
• Cómo crear una base de datos y una tabla SQLite de muestra
• Cómo llenar la tabla de la base de datos con datos de muestra
• Cómo implementar endpoints para recuperar todos los usuarios, recuperar un usuario específico por ID, crear un nuevo usuario, actualizar un usuario existente y eliminar un usuario
• Cómo probar los endpoints de la API REST utilizando cURL o un navegador

🏆 Logros

Después de completar este proyecto, podrás:

• Construir una aplicación Flask con SQLite como base de datos
• Realizar operaciones CRUD en los datos de una base de datos SQLite utilizando Flask
• Implementar y probar endpoints de API REST para una aplicación Flask

Skills Graph

Crear los archivos del proyecto

Comencemos creando los archivos necesarios del proyecto para nuestra aplicación Flask.

1. Utiliza el comando cd para entrar al directorio ~/project.
2. Dentro del directorio del proyecto, crea un nuevo archivo Python llamado app.py.
3. Crea un nuevo archivo llamado create_database.py en el mismo directorio.

cd ~/project
touch app.py create_database.py

La estructura de tu proyecto debería verse así:

project/
├── app.py
└── create_database.py

Crear una base de datos SQLite de muestra

A continuación, creemos una base de datos SQLite y llenémosla con datos de muestra utilizando Python y la biblioteca SQLite.

En el archivo create_database.py, agrega el siguiente código para importar los módulos necesarios:

import sqlite3

Agrega el siguiente código para crear la base de datos y la tabla:

## Create a connection to the database
conn = sqlite3.connect('example.db')

## Create a cursor object to execute SQL statements
cursor = conn.cursor()

## Create a table called "users"
cursor.execute('''CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY,
                    name TEXT,
                    email TEXT
                )''')

A continuación, insertemos algunos datos de muestra en la tabla:

## Insert sample data into the "users" table
users = [
    ('John Doe', 'johndoe@example.com'),
    ('Jane Smith', 'janesmith@example.com'),
    ('Bob Johnson', 'bobjohnson@example.com')
]
cursor.executemany('INSERT INTO users (name, email) VALUES (?,?)', users)

## Commit the changes to the database
conn.commit()

## Close the cursor and the database connection
cursor.close()
conn.close()

cd ~/project
python create_database.py

Después de ejecutar este código, tendrás un archivo de base de datos SQLite llamado example.db con la tabla "users" que contiene los datos de muestra. Esta base de datos se utilizará en nuestra aplicación Flask para recuperar y manipular datos.

Construir una API REST en Flask

Ahora, construyamos una API REST en Flask utilizando la base de datos SQLite example.db que creamos anteriormente.

En el archivo app.py, agrega el siguiente código para importar los módulos necesarios:

from flask import Flask, jsonify, request
import sqlite3

Agrega el siguiente código para crear la aplicación Flask:

## Create the Flask application
app = Flask(__name__)

Ahora, definamos los endpoints para nuestra API REST:

Endpoint para recuperar todos los usuarios

## Endpoint to retrieve all users
@app.route('/users', methods=['GET'])
def get_users():
    ## Connect to the database
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## Execute the SQL query to retrieve all users
    cursor.execute('SELECT * FROM users')
    users = cursor.fetchall()

    ## Close the cursor and the database connection
    cursor.close()
    conn.close()

    ## Return the users as JSON
    return jsonify(users)

En este código, primero nos conectamos a la base de datos SQLite utilizando el método connect(). Luego, creamos un objeto cursor para ejecutar declaraciones SQL utilizando el método cursor(). A continuación, ejecutamos la consulta SQL para recuperar todos los usuarios de la tabla "users" utilizando el método execute(). Finalmente, cerramos el cursor y la conexión a la base de datos utilizando el método close() y devolvemos los usuarios como JSON utilizando el método jsonify().

Endpoint para recuperar un usuario específico por ID

## Endpoint to retrieve a specific user by ID
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    ## Connect to the database
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## Execute the SQL query to retrieve the user by ID
    cursor.execute('SELECT * FROM users WHERE id =?', (user_id,))
    user = cursor.fetchone()

    ## Close the cursor and the database connection
    cursor.close()
    conn.close()

    ## Check if the user exists
    if user:
        return jsonify(user)
    else:
        return jsonify({'message': 'User not found'}), 404

Este endpoint es similar al anterior, excepto que recupera un usuario específico por ID en lugar de todos los usuarios. Utilizamos el método fetchone() para recuperar un solo usuario de la tabla "users".

Endpoint para crear un nuevo usuario

## Endpoint to create a new user
@app.route('/users', methods=['POST'])
def create_user():
    ## Get the user data from the request body
    data = request.get_json()
    name = data['name']
    email = data['email']

    ## Connect to the database
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## Execute the SQL query to insert a new user
    cursor.execute('INSERT INTO users (name, email) VALUES (?,?)', (name, email))
    conn.commit()

    ## Close the cursor and the database connection
    cursor.close()
    conn.close()

    ## Return a success message
    return jsonify({'message': 'User created successfully'})

Este endpoint se utiliza para crear un nuevo usuario en la tabla "users". Utilizamos el método get_json() para obtener los datos del usuario del cuerpo de la solicitud. Luego, utilizamos el método execute() para insertar el nuevo usuario en la tabla "users".

Endpoint para actualizar un usuario existente

## Endpoint to update an existing user
@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    ## Get the updated user data from the request body
    data = request.get_json()
    name = data['name']
    email = data['email']

    ## Connect to the database
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## Execute the SQL query to update the user
    cursor.execute('UPDATE users SET name =?, email =? WHERE id =?', (name, email, user_id))
    conn.commit()

    ## Close the cursor and the database connection
    cursor.close()
    conn.close()

    ## Return a success message
    return jsonify({'message': 'User updated successfully'})

Este endpoint se utiliza para actualizar un usuario existente en la tabla "users". Utilizamos el método get_json() para obtener los datos actualizados del usuario del cuerpo de la solicitud. Luego, utilizamos el método execute() para actualizar el usuario en la tabla "users".

Endpoint para eliminar un usuario

## Endpoint to delete a user
@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    ## Connect to the database
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    ## Execute the SQL query to delete the user
    cursor.execute('DELETE FROM users WHERE id =?', (user_id,))
    conn.commit()

    ## Close the cursor and the database connection
    cursor.close()
    conn.close()

    ## Return a success message
    return jsonify({'message': 'User deleted successfully'})

Este endpoint se utiliza para eliminar un usuario de la tabla "users". Utilizamos el método execute() para eliminar el usuario de la tabla "users".

Finalmente, agreguemos el siguiente código para ejecutar la aplicación Flask:

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=8080)

Ejecutar la aplicación Flask

Para ejecutar el proyecto y iniciar la aplicación Flask, abre una terminal o el símbolo del sistema, navega hasta el directorio del proyecto y ejecuta el siguiente comando:

cd ~/project
python app.py

Deberías ver una salida similar a la siguiente:

* Serving Flask app 'app' (lazy loading)
* Environment: production
WARNING: This is a development server. Do not use it in a production deployment.
Use a production WSGI server instead.
* Debug mode: on
* Running on http://0.0.0.0:8080/ (Press CTRL+C to quit)

Esto significa que la aplicación Flask se está ejecutando localmente en tu máquina. Ahora puedes probar los endpoints de la API REST utilizando una herramienta como cURL o una herramienta de prueba de API como Postman.

Probar los endpoints de la API REST

1. Abre una nueva terminal o el símbolo del sistema.
2. Utiliza los siguientes comandos cURL para probar cada endpoint:

• Recuperar todos los usuarios:

curl http://localhost:8080/users

• Recuperar un usuario específico por ID (reemplaza <user_id> con un ID de usuario real):

curl http://localhost:8080/users/<user_id>

• Crear un nuevo usuario:

curl -X POST -H "Content-Type: application/json" -d '{"name":"John Doe","email":"johndoe@example.com"}' http://localhost:8080/users

• Actualizar un usuario existente (reemplaza <user_id> con un ID de usuario real):

curl -X PUT -H "Content-Type: application/json" -d '{"name":"Updated Name","email":"updated@example.com"}' http://localhost:8080/users/<user_id>

• Eliminar un usuario (reemplaza <user_id> con un ID de usuario real):

curl -X DELETE http://localhost:8080/users/<user_id>

Asegúrate de reemplazar <user_id> con el ID real de un usuario en la base de datos al probar los endpoints correspondientes.

Al ejecutar estos comandos cURL, puedes probar la funcionalidad de los endpoints de la API REST y verificar si funcionan correctamente.

Como alternativa, puedes cambiar a la pestaña Web 8080 para probar los endpoints de la API REST utilizando el navegador.

Con estos pasos de prueba, puedes asegurarte de que la aplicación Flask y sus endpoints de la API REST funcionen como se espera.

Resumen

En este proyecto, aprendimos cómo construir una API REST en Flask utilizando SQLite como base de datos. Creamos una base de datos SQLite de muestra y la llenamos con datos. Luego, construimos una aplicación Flask con endpoints para recuperar, crear, actualizar y eliminar usuarios de la base de datos. Siguiendo las instrucciones paso a paso, ahora deberías tener una aplicación Flask funcional que proporciona una API REST para interactuar con la base de datos SQLite.