使用 Flask 和 MySQL 的任務管理器應用程式-Python教學-PHP中文網

首頁

後端開發

Python教學

使用 Flask 和 MySQL 的任務管理器應用程式

Mary-Kate Olsen

Nov 17, 2024 am 08:19 AM

項目概況

這個專案是一個使用 Flask 和 MySQL 建構的任務管理器應用程式。它提供了一個簡單的 RESTful API 來管理任務，演示了基本的 CRUD（建立、讀取、刪除）操作。

此應用程式非常適合了解如何使用 Docker 將 Flask 應用程式容器化並與 MySQL 資料庫連接。

特徵

新增任務
查看所有任務
透過ID刪除任務

燒瓶代碼：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/<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_id>

MySQL 資料庫設定腳本

建立名為 init-db.sql 的 MySQL 腳本來設定資料庫和任務表：

要建立 init-db.sql 腳本，請依照下列步驟操作：

在專案目錄中建立一個新的檔案：

導覽至專案資料夾並建立一個名為 init-db.sql
的新文件新增 SQL 指令來設定資料庫和任務表：

在文字編輯器中開啟 init-db.sql 並新增下列 SQL 指令：

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

儲存文件：

我將檔案儲存為init-db.sql 位於我的docker-compose.yml 所在的專案資料夾中.

在 docker-compose.yml 中：

在我的 docker-compose.yml 檔案中，我有指向此腳本的捲配置。

下面是docker-compose.yml檔案

Docker 配置

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:

此配置確保當MySQL 容器啟動時，它將執行init-db.sql 腳本來設定task_db 資料庫並建立tasks 表。

注意：docker-entrypoint-initdb.d/目錄被MySQL容器用來執行.sql 容器初始啟動期間的腳本。解釋：

1。 version: '3':

指定正在使用的 Docker Compose 版本。

2。服務：

資料庫：

image: mysql:5.7: 使用 MySQL 5.7 鏡像。
環境： 設定 MySQL 容器的環境變數：
- MYSQL_ROOT_PASSWORD: MySQL 的 root 密碼。
- MYSQL_DATABASE:啟動時所建立的資料庫。
ports: 將 MySQL 容器的連接埠 3306 對應到主機的連接埠 3306。
卷：
- db_data:/var/lib/mysql: 將資料庫資料儲存在名為 db_data. 的 Docker 磁碟區中
- ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql: 掛載init-db.sql 腳本寫入MYSQL容器的初始化目錄，以便在容器啟動時運作。

網頁：
- build: .: 使用目前目錄中的 Dockerfile 為您的 Flask 應用建立 Docker 映像。
- 連接埠： 將 Flask 應用程式的連接埠 5000 對應到主機的連接埠 5000。
- depends_on: 確保 db 服務在 Web 服務之前啟動。
- 環境： 設定 Flask 的環境變數。
- volumes: 將目前專案目錄掛載到容器內的 /app 目錄中。 ### 卷部分： db_data: 定義一個命名卷 db_data 以在容器重新啟動之間保留 MySQL 資料。

Dockerfile：

定義 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/<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_id>

這個 Dockerfile 為 Flask 應用程式設定了一個輕量級的 Python 環境：

1。基礎鏡像： 使用 python:3.9-slim 來實現最短的 Python 運行時間。
工作目錄：將 /app 設定為工作目錄。

2。依賴項：複製requirements.txt並透過pip安裝依賴項。

3。工具安裝： 安裝 wait-for-it 以檢查服務準備。

4。應用程式程式碼： 將所有應用程式程式碼複製到容器中。

5。啟動指令： 執行 wait-for-it 以確保 MySQL DB (db:3306) 在啟動 app.py 之前準備就緒。

需求.txt 檔案

此requirements.txt 指定Python 專案需要Flask 框架 來建立Web 應用程式和mysql-connector-python 用於與MySQL 資料庫 連線和互動。當在映像建置過程中執行 pip install -rrequirements.txt 時，這些套件將安裝在 Docker 容器中。這確保應用程式擁有運行 Flask 伺服器並與 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/<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_id>

建立所有檔案後，下一步是建置並執行服務，使用以下命令來建置和執行服務。

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

要以分離模式運行服務，我使用以下命令而不是

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:

當我想停止服務時，我使用指令

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

現在，一旦服務處於運作狀態，請執行指令

Flask
mysql-connector-python

確保容器正在運作

現在是時候檢查服務 API 以確保它們按預期工作了。

測試項目

透過 http://localhost:5000/ 存取應用程式。

運行上述命令後，我能夠在瀏覽器上存取該應用程序，如下圖所示。

Task Manager App with Flask and MySQL

您可以使用 Postman 或 curl 來測試 /tasks 端點的 POST、GET 和 DELETE 操作。在這種情況下，我將使用curl。

捲曲命令：

取得任務：

GET 方法取得所有任務。

docker-compose build
docker-compose up

Task Manager App with Flask and MySQL

請注意，每當您在瀏覽器上執行 http://localhost:5000/tasks 時，它都會顯示您已新增的所有任務，如新增任務中所述。

新增任務：

POST 方法在資料庫中建立任務。

docker-compose up -d

這將向您的 Flask 應用程式發送帶有任務描述的 POST 請求。如果任務新增成功，您應該會收到以下回應：

docker-compose down

檢查瀏覽器的網路標籤或日誌以驗證 POST 請求是否正確發出。

我運行了該命令幾次，並自訂了「簡單任務」部分以產生不同的輸出，這是我運行的命令，輸出可以在下圖中看到。

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/<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_id>

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

刪除任務：

DELETE 方法透過 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:

我執行了以下命令來刪除 ID 為 4 的任務，如下圖所示，任務 4 已被刪除。

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

結論

使用 Flask 和 MySQL 建立任務管理器應用程式是了解 Web 服務開發、資料庫整合和 Docker 容器化基礎知識的絕佳方法。

該專案封裝了 Web 伺服器和資料庫如何協同工作以提供無縫功能。

擁抱這種學習體驗，並將其用作更深層的網路和基於雲端的開發專案的墊腳石。

以上是使用 Flask 和 MySQL 的任務管理器應用程式的詳細內容。更多資訊請關注PHP中文網其他相關文章！

陳述

本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn

Python中的合併列表：選擇正確的方法May 14, 2025 am 12:11 AM

Tomergelistsinpython，YouCanusethe操作員，estextMethod，ListComprehension，Oritertools

如何在Python 3中加入兩個列表？May 14, 2025 am 12:09 AM

在Python3中，可以通過多種方法連接兩個列表：1)使用運算符，適用於小列表，但對大列表效率低；2)使用extend方法，適用於大列表，內存效率高，但會修改原列表；3)使用*運算符，適用於合併多個列表，不修改原列表；4)使用itertools.chain，適用於大數據集，內存效率高。

Python串聯列表字符串May 14, 2025 am 12:08 AM

使用join()方法是Python中從列表連接字符串最有效的方法。 1)使用join()方法高效且易讀。 2)循環使用運算符對大列表效率低。 3)列表推導式與join()結合適用於需要轉換的場景。 4)reduce()方法適用於其他類型歸約，但對字符串連接效率低。完整句子結束。

Python執行，那是什麼？May 14, 2025 am 12:06 AM

pythonexecutionistheprocessoftransformingpypythoncodeintoExecutablestructions.1）InternterPreterReadSthecode，ConvertingTingitIntObyTecode，whepythonvirtualmachine（pvm）theglobalinterpreterpreterpreterpreterlock（gil）the thepythonvirtualmachine（pvm）

Python：關鍵功能是什麼May 14, 2025 am 12:02 AM

Python的關鍵特性包括：1.語法簡潔易懂，適合初學者；2.動態類型系統，提高開發速度；3.豐富的標準庫，支持多種任務；4.強大的社區和生態系統，提供廣泛支持；5.解釋性，適合腳本和快速原型開發；6.多範式支持，適用於各種編程風格。

Python：編譯器還是解釋器？May 13, 2025 am 12:10 AM

Python是解釋型語言，但也包含編譯過程。 1）Python代碼先編譯成字節碼。 2）字節碼由Python虛擬機解釋執行。 3）這種混合機制使Python既靈活又高效，但執行速度不如完全編譯型語言。

python用於循環與循環時：何時使用哪個？May 13, 2025 am 12:07 AM

UseeAforloopWheniteratingOveraseQuenceOrforAspecificnumberoftimes; useAwhiLeLoopWhenconTinuingUntilAcIntiment.forloopsareIdealForkNownsences，而WhileLeleLeleLeleLeleLoopSituationSituationsItuationsItuationSuationSituationswithUndEtermentersitations。