Heim >Backend-Entwicklung >Python-Tutorial >Task-Manager-App mit Flask und MySQL

Task-Manager-App mit Flask und MySQL

Mary-Kate OlsenOriginal: 2024-11-17 08:19:03470Durchsuche

Projektübersicht

Dieses Projekt ist eine Task-Manager-App, die mit Flask und MySQL erstellt wurde. Es bietet eine einfache RESTful-API zum Verwalten von Aufgaben und demonstriert grundlegende CRUD-Vorgänge (Erstellen, Lesen, Löschen).

Diese Anwendung ist perfekt, um zu verstehen, wie Flask-Anwendungen mithilfe von Docker containerisiert und mit einer MySQL-Datenbank verbunden werden können.

Merkmale

Neue Aufgaben hinzufügen
Alle Aufgaben anzeigen
Eine Aufgabe nach ID löschen

Flaschencode: 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')

MySQL-Datenbank-Setup-Skript

Erstellen Sie ein MySQL-Skript mit dem Namen init-db.sql, um die Datenbank und die Aufgabentabelle einzurichten:

Um das Skript init-db.sql zu erstellen, befolgen Sie diese Schritte:

Erstellen Sie eine neue Datei in Ihrem Projektverzeichnis:

Navigieren Sie zum Projektordner und erstellen Sie eine neue Datei mit dem Namen init-db.sql
Fügen Sie SQL-Befehle hinzu, um die Datenbank und die Aufgabentabelle einzurichten:

Öffnen Sie init-db.sql in einem Texteditor und fügen Sie die folgenden SQL-Befehle hinzu:

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

Speichern Sie die Datei:

Ich habe die Datei als init-db.sql im Projektordner gespeichert, in dem sich meine docker-compose.yml befindet .

In der docker-compose.yml:

In meiner docker-compose.yml-Datei habe ich die Volume-Konfiguration, die auf dieses Skript verweist.

Unten finden Sie die Datei docker-compose.yml

Docker-Konfiguration

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:

Diese Konfiguration stellt sicher, dass der MySQL-Container beim Start das Skript init-db.sql ausführt, um die task_dbDatenbank und erstellen Sie die Tabelle Aufgaben.

Hinweis: Das Verzeichnis docker-entrypoint-initdb.d/ wird von MySQL-Containern zum Ausführen von .sql Skripte beim ersten Start des Behälter. Erläuterung:

1. version: '3':

Gibt die verwendete Version von Docker Compose an.

2. Dienstleistungen:

- image: mysql:5.7: Verwendet das MySQL 5.7-Image.
- Umgebung: Legt Umgebungsvariablen für den MySQL-Container fest:
  - MYSQL_ROOT_PASSWORD: Das Root-Passwort für MySQL.
  - MYSQL_DATABASE: Die Datenbank, die beim Start erstellt werden soll.
- Ports: Ordnet den Port 3306 des MySQL-Containers dem Port 3306 Ihres Hosts zu.
- Bände:
  - db_data:/var/lib/mysql: Behält die Datenbankdaten in einem Docker-Volume namens db_data. bei
  - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql: Mountet die init-db.sql-Skript in das Initialisierungsverzeichnis des MYSQL-Containers, damit es ausgeführt wird, wenn der Container startet.
Web:
- build: .: Erstellt das Docker-Image für Ihre Flask-App mithilfe der Docker-Datei im aktuellen Verzeichnis.
- Ports: Ordnet den Port 5000 der Flask-App dem Port 5000 Ihres Hosts zu.
- abhängig_von: Stellt sicher, dass der db-Dienst vor dem Webdienst startet.
- Umgebung: Legt die Umgebungsvariable für Flask fest.
- Volumes: Mountet das aktuelle Projektverzeichnis im Verzeichnis /app im Container. Abschnitt ### Bände: db_data: Definiert ein benanntes Volume db_data, um die MySQL-Daten zwischen Container-Neustarts beizubehalten.

Docker-Datei:

Definieren Sie die Build-Anweisungen für die Flask-App:

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

Diese Docker-Datei richtet eine leichte Python-Umgebung für eine Flask-App ein:

1. Basisbild: Verwendet python:3.9-slim für minimale Python-Laufzeit.
Arbeitsverzeichnis: Legt /app als Arbeitsverzeichnis fest.

2. Abhängigkeiten:Kopiert „requirements.txt“ und installiert Abhängigkeiten über pip.

3. Tool-Installation: Installiert „Wait-for-It“ zur Überprüfung der Servicebereitschaft.

4. Anwendungscode: Kopiert den gesamten App-Code in den Container.

5. Startbefehl: Führt „wait-for-it“ aus, um sicherzustellen, dass die MySQL-Datenbank (db:3306) bereit ist, bevor app.py gestartet wird.

Anforderungen.txt-Datei

Diese requirements.txt gibt an, dass das Python-Projekt das Flask-Framework zum Erstellen von Webanwendungen und mysql-connector-python erfordert für die Verbindung und Interaktion mit einem MySQL Datenbank. Diese Pakete werden im Docker-Container installiert, wenn pip install -r require.txt während des Image-Erstellungsprozesses ausgeführt wird. Dadurch wird sichergestellt, dass die App über die notwendigen Tools verfügt, um den Flask-Server auszuführen und mit der MySQL-Datenbank zu kommunizieren.

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

Nachdem alle Dateien erstellt wurden, besteht der nächste Schritt darin, den Dienst zu erstellen und auszuführen. Der folgende Befehl wird zum Erstellen und Ausführen des Dienstes verwendet.

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

Um den Dienst im getrennten Modus auszuführen, habe ich den folgenden Befehl anstelle von docker-compose up
verwendet

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:

Wenn ich den Dienst beenden möchte, verwende ich den Befehl

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"]

Sobald der Dienst nun ausgeführt wird, führen Sie den Befehl aus

Flask
mysql-connector-python

um sicherzustellen, dass die Container ausgeführt werden

Jetzt ist es an der Zeit, die Service-API zu überprüfen, um sicherzustellen, dass sie wie erwartet funktioniert.

Testen des Projekts

Zugriff auf die App unter http://localhost:5000/ .
Ich konnte über meinen Browser auf die App zugreifen, nachdem ich den obigen Befehl ausgeführt hatte, wie im Bild unten gezeigt.

Task Manager App with Flask and MySQL

Sie können Postman oder Curl verwenden, um den /tasks-Endpunkt für POST-, GET- und DELETE-Vorgänge zu testen. In diesem Fall würde ich Curl verwenden.

Curl-Befehle:

Aufgaben abrufen:

Die GET-Methode ruft alle Aufgaben ab.

docker-compose build
docker-compose up

Task Manager App with Flask and MySQL

Beachten Sie, dass Ihnen jedes Mal, wenn Sie http://localhost:5000/tasks in Ihrem Browser ausführen, alle Aufgaben angezeigt werden, die Sie hinzugefügt haben, wie in der Aufgabe hinzufügen erläutert.

Eine Aufgabe hinzufügen:

Die POST-Methode erstellt Aufgaben in der Datenbank.

docker-compose up -d

Dadurch wird eine POST-Anfrage mit einer Aufgabenbeschreibung an Ihre Flask-App gesendet. Wenn die Aufgabe erfolgreich hinzugefügt wurde, sollten Sie eine Antwort wie:
erhalten

docker-compose down

Überprüfen Sie die Registerkarte „Netzwerk“ oder die Protokolle Ihres Browsers, um sicherzustellen, dass die POST-Anfrage korrekt erfolgt.

Ich habe den Befehl ein paar Mal ausgeführt und den Teil angepasst, in dem „Einfache Aufgabe“ steht, um verschiedene Ausgaben zu generieren. Hier sind die Befehle, die ich ausgeführt habe, und die Ausgaben sind in den Bildern unten zu sehen.

docker ps

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

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

Task Manager App with Flask and MySQL

Eine Aufgabe löschen:

Die DELETE-Methode entfernt Aufgaben nach 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:

Ich habe den folgenden Befehl ausgeführt, um die Aufgabe mit der ID:4 zu entfernen, wie im Bild unten zu sehen ist. Aufgabe 4 wurde entfernt.

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"]

Task Manager App with Flask and MySQL

Abschluss

Das Erstellen einer Task-Manager-App mit Flask und MySQL ist eine hervorragende Möglichkeit, die Grundlagen der Webdienstentwicklung, Datenbankintegration und Containerisierung mit Docker zu verstehen.

Dieses Projekt fasst zusammen, wie Webserver und Datenbanken zusammenarbeiten, um nahtlose Funktionalität bereitzustellen.

Nehmen Sie diese Lernerfahrung an und nutzen Sie sie als Sprungbrett für tiefergehende web- und cloudbasierte Entwicklungsprojekte.

Das obige ist der detaillierte Inhalt vonTask-Manager-App mit Flask und MySQL. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!

Python sql mysql flask restful postman pip define if for cURL Directory using var delete this table docker database http Access

Stellungnahme：

Der Inhalt dieses Artikels wird freiwillig von Internetnutzern beigesteuert und das Urheberrecht liegt beim ursprünglichen Autor. Diese Website übernimmt keine entsprechende rechtliche Verantwortung. Wenn Sie Inhalte finden, bei denen der Verdacht eines Plagiats oder einer Rechtsverletzung besteht, wenden Sie sich bitte an admin@php.cn

Vorheriger Artikel：Wie hat sich die String-Ersetzung in Python 3.x mit „str.replace()“ geändert?Nächster Artikel：Wie hat sich die String-Ersetzung in Python 3.x mit „str.replace()“ geändert?

In Verbindung stehende Artikel

Mehr sehen