Création d'une API REST Flask avec SQLite

Introduction

Dans ce projet, nous allons apprendre à construire une API REST avec Flask en utilisant SQLite comme base de données. Nous commencerons par créer une base de données SQLite d'exemple et la remplir avec des données. Ensuite, nous allons construire une application Flask avec des points d'accès (endpoints) pour effectuer des opérations CRUD (Créer, Lire, Mettre à jour, Supprimer) sur les données de la base de données SQLite.

👀 Aperçu

🎯 Tâches

Dans ce projet, vous allez apprendre :

• Comment créer une application Flask avec SQLite comme backend de base de données
• Comment créer une base de données et une table SQLite d'exemple
• Comment remplir la table de la base de données avec des données d'exemple
• Comment implémenter des points d'accès pour récupérer tous les utilisateurs, récupérer un utilisateur spécifique par son ID, créer un nouvel utilisateur, mettre à jour un utilisateur existant et supprimer un utilisateur
• Comment tester les points d'accès de l'API REST en utilisant cURL ou un navigateur

🏆 Réalisations

Après avoir terminé ce projet, vous serez en mesure de :

• Construire une application Flask avec SQLite comme base de données
• Effectuer des opérations CRUD sur les données d'une base de données SQLite en utilisant Flask
• Implémenter et tester les points d'accès d'une API REST pour une application Flask

Skills Graph

Créer les fichiers du projet

Commençons par créer les fichiers de projet nécessaires pour notre application Flask.

1. Utilisez la commande cd pour accéder au répertoire ~/project.
2. À l'intérieur du répertoire du projet, créez un nouveau fichier Python nommé app.py.
3. Créez un nouveau fichier nommé create_database.py dans le même répertoire.

cd ~/project
touch app.py create_database.py

La structure de votre projet devrait ressembler à ceci :

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

Créer une base de données SQLite d'exemple

Ensuite, créons une base de données SQLite et remplissons-la avec des données d'exemple en utilisant Python et la bibliothèque SQLite.

Dans le fichier create_database.py, ajoutez le code suivant pour importer les modules nécessaires :

import sqlite3

Ajoutez le code suivant pour créer la base de données et la table :

## 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
                )''')

Ensuite, insérons quelques données d'exemple dans la table :

## 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

Après avoir exécuté ce code, vous aurez un fichier de base de données SQLite nommé example.db avec une table "users" contenant les données d'exemple. Cette base de données sera utilisée dans notre application Flask pour récupérer et manipuler des données.

Construire une API REST avec Flask

Maintenant, construisons une API REST avec Flask en utilisant la base de données SQLite example.db que nous avons créée précédemment.

Dans le fichier app.py, ajoutez le code suivant pour importer les modules nécessaires :

from flask import Flask, jsonify, request
import sqlite3

Ajoutez le code suivant pour créer l'application Flask :

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

Maintenant, définissons les points d'accès (endpoints) de notre API REST :

Point d'accès pour récupérer tous les utilisateurs

## 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)

Dans ce code, nous nous connectons d'abord à la base de données SQLite en utilisant la méthode connect(). Ensuite, nous créons un objet curseur pour exécuter des instructions SQL en utilisant la méthode cursor(). Ensuite, nous exécutons la requête SQL pour récupérer tous les utilisateurs de la table "users" en utilisant la méthode execute(). Enfin, nous fermons le curseur et la connexion à la base de données en utilisant la méthode close() et nous renvoyons les utilisateurs au format JSON en utilisant la méthode jsonify().

Point d'accès pour récupérer un utilisateur spécifique par son 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

Ce point d'accès est similaire au précédent, sauf qu'il récupère un utilisateur spécifique par son ID au lieu de tous les utilisateurs. Nous utilisons la méthode fetchone() pour récupérer un seul utilisateur de la table "users".

Point d'accès pour créer un nouvel utilisateur

## 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'})

Ce point d'accès est utilisé pour créer un nouvel utilisateur dans la table "users". Nous utilisons la méthode get_json() pour obtenir les données de l'utilisateur à partir du corps de la requête. Ensuite, nous utilisons la méthode execute() pour insérer le nouvel utilisateur dans la table "users".

Point d'accès pour mettre à jour un utilisateur existant

## 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'})

Ce point d'accès est utilisé pour mettre à jour un utilisateur existant dans la table "users". Nous utilisons la méthode get_json() pour obtenir les données mises à jour de l'utilisateur à partir du corps de la requête. Ensuite, nous utilisons la méthode execute() pour mettre à jour l'utilisateur dans la table "users".

Point d'accès pour supprimer un utilisateur

## 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'})

Ce point d'accès est utilisé pour supprimer un utilisateur de la table "users". Nous utilisons la méthode execute() pour supprimer l'utilisateur de la table "users".

Enfin, ajoutons le code suivant pour exécuter l'application Flask :

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

Exécuter l'application Flask

Pour exécuter le projet et démarrer l'application Flask, ouvrez un terminal ou une invite de commande, accédez au répertoire du projet et exécutez la commande suivante :

cd ~/project
python app.py

Vous devriez voir une sortie similaire à ce qui suit :

* 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)

Cela signifie que l'application Flask est en cours d'exécution localement sur votre machine. Vous pouvez maintenant tester les points d'accès de l'API REST en utilisant un outil comme cURL ou un outil de test d'API comme Postman.

Tester les points d'accès de l'API REST

1. Ouvrez un nouveau terminal ou une nouvelle invite de commande.
2. Utilisez les commandes cURL suivantes pour tester chaque point d'accès :

• Récupérer tous les utilisateurs :

curl http://localhost:8080/users

• Récupérer un utilisateur spécifique par son ID (remplacez <user_id> par un véritable ID d'utilisateur) :

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

• Créer un nouvel utilisateur :

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

• Mettre à jour un utilisateur existant (remplacez <user_id> par un véritable ID d'utilisateur) :

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

• Supprimer un utilisateur (remplacez <user_id> par un véritable ID d'utilisateur) :

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

Assurez-vous de remplacer <user_id> par l'ID réel d'un utilisateur dans la base de données lors du test des points d'accès correspondants.

En exécutant ces commandes cURL, vous pouvez tester la fonctionnalité des points d'accès de l'API REST et vérifier si ils fonctionnent correctement.

Alternativement, vous pouvez passer à l'onglet Web 8080 pour tester les points d'accès de l'API REST à l'aide du navigateur.

Grâce à ces étapes de test, vous pouvez vous assurer que l'application Flask et ses points d'accès de l'API REST fonctionnent comme prévu.

Résumé

Dans ce projet, nous avons appris à construire une API REST avec Flask en utilisant SQLite comme base de données. Nous avons créé une base de données SQLite d'exemple et l'avons remplie de données. Ensuite, nous avons construit une application Flask avec des points d'accès pour récupérer, créer, mettre à jour et supprimer des utilisateurs de la base de données. En suivant les instructions étape par étape, vous devriez maintenant avoir une application Flask fonctionnelle qui fournit une API REST pour interagir avec la base de données SQLite.