Application Task Manager avec Flask et MySQL-Tutoriel Python-php.cn

Aperçu du projet

Ce projet est une Application Task Manager construite avec Flask et MySQL. Il fournit une API RESTful simple pour gérer les tâches, démontrant les opérations de base CRUD (Créer, Lire, Supprimer).

Cette application est parfaite pour comprendre comment les applications Flask peuvent être conteneurisées à l'aide de Docker et connectées à une base de données MySQL.

Caractéristiques

Ajouter de nouvelles tâches
Afficher toutes les tâches
Supprimer une tâche par ID

Code du flacon : app.py

from flask import Flask, request, jsonify
import mysql.connector
from mysql.connector import Error

app = Flask(__name__)

# Database connection function
def get_db_connection():
    try:
        connection = mysql.connector.connect(
            host="db",
            user="root",
            password="example",
            database="task_db"
        )
        return connection
    except Error as e:
        return str(e)

# Route for the home page
@app.route('/')
def home():
    return "Welcome to the Task Management API! Use /tasks to interact with tasks."

# Route to create a new task
@app.route('/tasks', methods=['POST'])
def add_task():
    task_description = request.json.get('description')
    if not task_description:
        return jsonify({"error": "Task description is required"}), 400

    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,))
    connection.commit()
    task_id = cursor.lastrowid
    cursor.close()
    connection.close()

    return jsonify({"message": "Task added successfully", "task_id": task_id}), 201

# Route to get all tasks
@app.route('/tasks', methods=['GET'])
def get_tasks():
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("SELECT id, description FROM tasks")
    tasks = cursor.fetchall()
    cursor.close()
    connection.close()

    task_list = [{"id": task[0], "description": task[1]} for task in tasks]
    return jsonify(task_list), 200

# Route to delete a task by ID
@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,))
    connection.commit()
    cursor.close()
    connection.close()

    return jsonify({"message": "Task deleted successfully"}), 200

if __name__ == "__main__":
    app.run(host='0.0.0.0')

Copier après la connexion

Script de configuration de la base de données MySQL

Créez un script MySQL nommé init-db.sql pour configurer la base de données et la table des tâches :

Pour créer le script init-db.sql, suivez ces étapes :

Créez un nouveau fichier dans le répertoire de votre projet :

Accédez au dossier du projet et créez un nouveau fichier nommé init-db.sql
Ajoutez des commandes SQL pour configurer la base de données et la table des tâches :

Ouvrez init-db.sql dans un éditeur de texte et ajoutez les commandes SQL suivantes :

CREATE DATABASE IF NOT EXISTS task_db;
USE task_db;

CREATE TABLE IF NOT EXISTS tasks (
    id INT AUTO_INCREMENT PRIMARY KEY,
    description VARCHAR(255) NOT NULL
);

Copier après la connexion

Enregistrez le fichier :

J'ai enregistré le fichier sous init-db.sql dans le dossier du projet où se trouve mon docker-compose.yml .

Dans le docker-compose.yml :

Dans mon fichier docker-compose.yml, j'ai la configuration des volumes qui pointe vers ce script.

Ci-dessous le docker-compose.ymlfichier

Configuration du menu fixe

docker-compose.yml :

version: '3'
services:
  db:
    image: mysql:5.7
    environment:
      MYSQL_ROOT_PASSWORD: example
      MYSQL_DATABASE: task_db
    ports:
      - "3306:3306"
    volumes:
      - db_data:/var/lib/mysql
      - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql

  web:
    build: .
    ports:
      - "5000:5000"
    depends_on:
      - db
    environment:
      FLASK_ENV: development
    volumes:
      - .:/app

volumes:
  db_data:

Copier après la connexion

Cette configuration garantit que lorsque le conteneur MySQL démarre, il exécutera le script init-db.sql pour configurer le task_db base de données et créez la table tâches.

Remarque : Le répertoire docker-entrypoint-initdb.d/ est utilisé par les conteneurs MySQL pour exécuter .sql scripts lors du démarrage initial du conteneur.

Explication:

1. version : '3' : Spécifie la version de Docker Compose utilisée.

2. services :

base de données :
- image : mysql:5.7 : Utilise l'image MySQL 5.7.
- environnement : Définit les variables d'environnement pour le conteneur MySQL :
  - MYSQL_ROOT_PASSWORD : Le mot de passe root pour MySQL.
  - MYSQL_DATABASE : La base de données à créer au démarrage.
- ports : Mappe le port 3306 du conteneur MySQL au port 3306 de votre hôte.
- volumes :
  - db_data:/var/lib/mysql: Conserve les données de la base de données dans un volume Docker appelé db_data.
  - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql: Monte le init-db.sql script dans le répertoire d'initialisation du conteneur MYSQL afin qu'il s'exécute au démarrage du conteneur.
web :
- build : . : crée l'image Docker pour votre application Flask à l'aide du Dockerfile dans le répertoire actuel.
- ports : Mappe le port 5000 de l'application Flask au port 5000 de votre hôte.
- depends_on : Garantit que le service db démarre avant le service Web.
- environnement : Définit la variable d'environnement pour Flask.
- volumes : Monte le répertoire du projet actuel dans le répertoire /app à l'intérieur du conteneur. Section des volumes ### : db_data : Définit un volume nommé db_data pour conserver les données MySQL entre les redémarrages du conteneur.

Fichier Docker :

Définissez les instructions de construction pour l'application Flask :

from flask import Flask, request, jsonify
import mysql.connector
from mysql.connector import Error

app = Flask(__name__)

# Database connection function
def get_db_connection():
    try:
        connection = mysql.connector.connect(
            host="db",
            user="root",
            password="example",
            database="task_db"
        )
        return connection
    except Error as e:
        return str(e)

# Route for the home page
@app.route('/')
def home():
    return "Welcome to the Task Management API! Use /tasks to interact with tasks."

# Route to create a new task
@app.route('/tasks', methods=['POST'])
def add_task():
    task_description = request.json.get('description')
    if not task_description:
        return jsonify({"error": "Task description is required"}), 400

    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,))
    connection.commit()
    task_id = cursor.lastrowid
    cursor.close()
    connection.close()

    return jsonify({"message": "Task added successfully", "task_id": task_id}), 201

# Route to get all tasks
@app.route('/tasks', methods=['GET'])
def get_tasks():
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("SELECT id, description FROM tasks")
    tasks = cursor.fetchall()
    cursor.close()
    connection.close()

    task_list = [{"id": task[0], "description": task[1]} for task in tasks]
    return jsonify(task_list), 200

# Route to delete a task by ID
@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,))
    connection.commit()
    cursor.close()
    connection.close()

    return jsonify({"message": "Task deleted successfully"}), 200

if __name__ == "__main__":
    app.run(host='0.0.0.0')

Copier après la connexion

Ce Dockerfile configure un environnement Python léger pour une application Flask :

1. Image de base : Utilise python:3.9-slim pour un temps d'exécution Python minimal.
Répertoire de travail : définit /app comme répertoire de travail.

2. Dépendances : Copie les exigences.txt et installe les dépendances via pip.

3. Installation de l'outil : Installe l'attente pour vérifier l'état de préparation du service.

4. Code d'application : Copie tout le code de l'application dans le conteneur.

5. Commande de démarrage : Exécute wait-for-it pour s'assurer que la base de données MySQL (db:3306) est prête avant de démarrer app.py.

Fichier Exigences.txt

Ce requirements.txt précise que le projet Python nécessite le Flask framework pour créer des applications Web et mysql-connector-python pour se connecter et interagir avec une base de données MySQL. Ces packages seront installés dans le conteneur Docker lorsque pip install -r Requirements.txt sera exécuté pendant le processus de création d'image. Cela garantit que l'application dispose des outils nécessaires pour exécuter le serveur Flask et communiquer avec la base de données MySQL.

from flask import Flask, request, jsonify
import mysql.connector
from mysql.connector import Error

app = Flask(__name__)

# Database connection function
def get_db_connection():
    try:
        connection = mysql.connector.connect(
            host="db",
            user="root",
            password="example",
            database="task_db"
        )
        return connection
    except Error as e:
        return str(e)

# Route for the home page
@app.route('/')
def home():
    return "Welcome to the Task Management API! Use /tasks to interact with tasks."

# Route to create a new task
@app.route('/tasks', methods=['POST'])
def add_task():
    task_description = request.json.get('description')
    if not task_description:
        return jsonify({"error": "Task description is required"}), 400

    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,))
    connection.commit()
    task_id = cursor.lastrowid
    cursor.close()
    connection.close()

    return jsonify({"message": "Task added successfully", "task_id": task_id}), 201

# Route to get all tasks
@app.route('/tasks', methods=['GET'])
def get_tasks():
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("SELECT id, description FROM tasks")
    tasks = cursor.fetchall()
    cursor.close()
    connection.close()

    task_list = [{"id": task[0], "description": task[1]} for task in tasks]
    return jsonify(task_list), 200

# Route to delete a task by ID
@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,))
    connection.commit()
    cursor.close()
    connection.close()

    return jsonify({"message": "Task deleted successfully"}), 200

if __name__ == "__main__":
    app.run(host='0.0.0.0')

Copier après la connexion

Après avoir créé tous les fichiers, l'étape suivante consiste à créer et exécuter le service, la commande suivante est utilisée pour créer et exécuter le service.

CREATE DATABASE IF NOT EXISTS task_db;
USE task_db;

CREATE TABLE IF NOT EXISTS tasks (
    id INT AUTO_INCREMENT PRIMARY KEY,
    description VARCHAR(255) NOT NULL
);

Copier après la connexion

pour exécuter le service en mode détaché, j'ai utilisé la commande suivante au lieu de docker-compose up

version: '3'
services:
  db:
    image: mysql:5.7
    environment:
      MYSQL_ROOT_PASSWORD: example
      MYSQL_DATABASE: task_db
    ports:
      - "3306:3306"
    volumes:
      - db_data:/var/lib/mysql
      - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql

  web:
    build: .
    ports:
      - "5000:5000"
    depends_on:
      - db
    environment:
      FLASK_ENV: development
    volumes:
      - .:/app

volumes:
  db_data:

Copier après la connexion

quand je veux arrêter le service j'utilise la commande

FROM python:3.9-slim

WORKDIR /app

# Install dependencies

COPY requirements.txt .
RUN pip install -r requirements.txt

# Install wait-for-it tool#

RUN apt-get update && apt-get install -y wait-for-it

#Copy the application code>

COPY . .

# Use wait-for-it to wait for DB and start the Flask app

CMD ["wait-for-it", "db:3306", "--", "python", "app.py"]

Copier après la connexion

Maintenant, une fois que le service est en cours d'exécution, exécutez la commande

Flask
mysql-connector-python

Copier après la connexion

pour s'assurer que les conteneurs fonctionnent

Il est maintenant temps de vérifier l'API du service pour vous assurer qu'elle fonctionne comme prévu.

Tester le projet

Accédez à l'application sur http://localhost:5000/ .
J'ai pu accéder à l'application sur mon navigateur après avoir exécuté la commande ci-dessus, comme le montre l'image ci-dessous.

Task Manager App with Flask and MySQL

Vous pouvez utiliser Postman ou curl pour tester le point de terminaison /tasks pour les opérations POST, GET et DELETE. Dans ce cas, j'utiliserais curl.

Commandes curl :

Obtenir des tâches :

La méthode GET récupère toutes les tâches.

docker-compose build
docker-compose up

Copier après la connexion

Task Manager App with Flask and MySQL

Notez que chaque fois que vous exécutez http://localhost:5000/tasks sur votre navigateur, il vous montre toutes les tâches que vous avez ajoutées comme expliqué dans la tâche d'ajout.

Ajouter une tâche :

La méthode POST crée des tâches dans la base de données.

docker-compose up -d

Copier après la connexion

Cela enverra une requête POST à votre application Flask avec une description de la tâche. Si la tâche est ajoutée avec succès, vous devriez recevoir une réponse telle que :

docker-compose down

Copier après la connexion

vérifiez l'onglet réseau ou les journaux de votre navigateur pour vérifier que la requête POST est effectuée correctement.

J'ai exécuté la commande plusieurs fois et personnalisé la partie où il est dit Tâche simple pour générer différentes sorties. Voici les commandes que j'ai exécutées et les sorties peuvent être vues dans les images ci-dessous.

docker ps

Copier après la connexion

Task Manager App with Flask and MySQL

from flask import Flask, request, jsonify
import mysql.connector
from mysql.connector import Error

app = Flask(__name__)

# Database connection function
def get_db_connection():
    try:
        connection = mysql.connector.connect(
            host="db",
            user="root",
            password="example",
            database="task_db"
        )
        return connection
    except Error as e:
        return str(e)

# Route for the home page
@app.route('/')
def home():
    return "Welcome to the Task Management API! Use /tasks to interact with tasks."

# Route to create a new task
@app.route('/tasks', methods=['POST'])
def add_task():
    task_description = request.json.get('description')
    if not task_description:
        return jsonify({"error": "Task description is required"}), 400

    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("INSERT INTO tasks (description) VALUES (%s)", (task_description,))
    connection.commit()
    task_id = cursor.lastrowid
    cursor.close()
    connection.close()

    return jsonify({"message": "Task added successfully", "task_id": task_id}), 201

# Route to get all tasks
@app.route('/tasks', methods=['GET'])
def get_tasks():
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("SELECT id, description FROM tasks")
    tasks = cursor.fetchall()
    cursor.close()
    connection.close()

    task_list = [{"id": task[0], "description": task[1]} for task in tasks]
    return jsonify(task_list), 200

# Route to delete a task by ID
@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    connection = get_db_connection()
    if isinstance(connection, str):  # If connection fails
        return jsonify({"error": connection}), 500

    cursor = connection.cursor()
    cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,))
    connection.commit()
    cursor.close()
    connection.close()

    return jsonify({"message": "Task deleted successfully"}), 200

if __name__ == "__main__":
    app.run(host='0.0.0.0')

Copier après la connexion

Task Manager App with Flask and MySQL

CREATE DATABASE IF NOT EXISTS task_db;
USE task_db;

CREATE TABLE IF NOT EXISTS tasks (
    id INT AUTO_INCREMENT PRIMARY KEY,
    description VARCHAR(255) NOT NULL
);

Copier après la connexion

Task Manager App with Flask and MySQL

Supprimer une tâche :

La méthode DELETE supprime les tâches par ID.

version: '3'
services:
  db:
    image: mysql:5.7
    environment:
      MYSQL_ROOT_PASSWORD: example
      MYSQL_DATABASE: task_db
    ports:
      - "3306:3306"
    volumes:
      - db_data:/var/lib/mysql
      - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql

  web:
    build: .
    ports:
      - "5000:5000"
    depends_on:
      - db
    environment:
      FLASK_ENV: development
    volumes:
      - .:/app

volumes:
  db_data:

Copier après la connexion

J'ai exécuté la commande ci-dessous pour supprimer la tâche avec l'ID : 4, comme le montre l'image ci-dessous, la tâche 4 a été supprimée.

FROM python:3.9-slim

WORKDIR /app

# Install dependencies

COPY requirements.txt .
RUN pip install -r requirements.txt

# Install wait-for-it tool#

RUN apt-get update && apt-get install -y wait-for-it

#Copy the application code>

COPY . .

# Use wait-for-it to wait for DB and start the Flask app

CMD ["wait-for-it", "db:3306", "--", "python", "app.py"]

Copier après la connexion

Task Manager App with Flask and MySQL

Conclusion

Créer une application Task Manager à l'aide de Flask et MySQL est un excellent moyen de comprendre les principes fondamentaux du développement de services Web, de l'intégration de bases de données et de la conteneurisation avec Docker.

Ce projet résume la façon dont les serveurs Web et les bases de données fonctionnent à l'unisson pour fournir des fonctionnalités transparentes.

Profitez de cette expérience d'apprentissage et utilisez-la comme tremplin vers des projets de développement Web et cloud plus approfondis.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!