搜尋
首頁後端開發Python教學使用 Flask 和 MySQL 的任務管理器應用程式
使用 Flask 和 MySQL 的任務管理器應用程式
Mary-Kate Olsen
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查找文本文件的ZIPF分佈如何使用Python查找文本文件的ZIPF分佈Mar 05, 2025 am 09:58 AM

本教程演示如何使用Python處理Zipf定律這一統計概念，並展示Python在處理該定律時讀取和排序大型文本文件的效率。 您可能想知道Zipf分佈這個術語是什麼意思。要理解這個術語，我們首先需要定義Zipf定律。別擔心，我會盡量簡化說明。 Zipf定律 Zipf定律簡單來說就是：在一個大型自然語言語料庫中，最頻繁出現的詞的出現頻率大約是第二頻繁詞的兩倍，是第三頻繁詞的三倍，是第四頻繁詞的四倍，以此類推。 讓我們來看一個例子。如果您查看美國英語的Brown語料庫，您會注意到最頻繁出現的詞是“th

python中的圖像過濾python中的圖像過濾Mar 03, 2025 am 09:44 AM

處理嘈雜的圖像是一個常見的問題，尤其是手機或低分辨率攝像頭照片。 本教程使用OpenCV探索Python中的圖像過濾技術來解決此問題。 圖像過濾：功能強大的工具圖像過濾器

我如何使用美麗的湯來解析HTML？我如何使用美麗的湯來解析HTML？Mar 10, 2025 pm 06:54 PM

本文解釋瞭如何使用美麗的湯庫來解析html。 它詳細介紹了常見方法，例如find（），find_all（），select（）和get_text（），以用於數據提取，處理不同的HTML結構和錯誤以及替代方案（SEL）

Python中的平行和並發編程簡介Python中的平行和並發編程簡介Mar 03, 2025 am 10:32 AM

Python是數據科學和處理的最愛，為高性能計算提供了豐富的生態系統。但是，Python中的並行編程提出了獨特的挑戰。本教程探討了這些挑戰，重點是全球解釋

如何使用TensorFlow或Pytorch進行深度學習？如何使用TensorFlow或Pytorch進行深度學習？Mar 10, 2025 pm 06:52 PM

本文比較了Tensorflow和Pytorch的深度學習。 它詳細介紹了所涉及的步驟：數據準備，模型構建，培訓，評估和部署。 框架之間的關鍵差異，特別是關於計算刻度的

如何在Python中實現自己的數據結構如何在Python中實現自己的數據結構Mar 03, 2025 am 09:28 AM

本教程演示了在Python 3中創建自定義管道數據結構，利用類和操作員超載以增強功能。 管道的靈活性在於它能夠將一系列函數應用於數據集的能力，GE

python對象的序列化和避難所化：第1部分python對象的序列化和避難所化：第1部分Mar 08, 2025 am 09:39 AM

Python 對象的序列化和反序列化是任何非平凡程序的關鍵方面。如果您將某些內容保存到 Python 文件中，如果您讀取配置文件，或者如果您響應 HTTP 請求，您都會進行對象序列化和反序列化。 從某種意義上說，序列化和反序列化是世界上最無聊的事情。誰會在乎所有這些格式和協議？您想持久化或流式傳輸一些 Python 對象，並在以後完整地取回它們。 這是一種在概念層面上看待世界的好方法。但是，在實際層面上，您選擇的序列化方案、格式或協議可能會決定程序運行的速度、安全性、維護狀態的自由度以及與其他系

Python中的數學模塊：統計Python中的數學模塊：統計Mar 09, 2025 am 11:40 AM

Python的statistics模塊提供強大的數據統計分析功能，幫助我們快速理解數據整體特徵，例如生物統計學和商業分析等領域。無需逐個查看數據點，只需查看均值或方差等統計量，即可發現原始數據中可能被忽略的趨勢和特徵，並更輕鬆、有效地比較大型數據集。 本教程將介紹如何計算平均值和衡量數據集的離散程度。除非另有說明，本模塊中的所有函數都支持使用mean()函數計算平均值，而非簡單的求和平均。 也可使用浮點數。 import random import statistics from fracti

See all articles

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅動的應用程序，用於創建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

AI Hentai Generator

AI Hentai Generator

免費產生 AI 無盡。

顯示更多

熱門文章

R.E.P.O.能量晶體解釋及其做什麼（黃色晶體）
2 週前By尊渡假赌尊渡假赌尊渡假赌
倉庫：如何復興隊友
1 個月前By尊渡假赌尊渡假赌尊渡假赌
Hello Kitty Island冒險：如何獲得巨型種子
4 週前By尊渡假赌尊渡假赌尊渡假赌
擊敗分裂小說需要多長時間？
3 週前ByDDD
R.E.P.O.保存文件位置：在哪里以及如何保護它？
3 週前ByDDD
顯示更多

熱工具

Dreamweaver CS6

Dreamweaver CS6

視覺化網頁開發工具

禪工作室 13.0.1

禪工作室 13.0.1

強大的PHP整合開發環境

EditPlus 中文破解版

EditPlus 中文破解版

體積小，語法高亮，不支援程式碼提示功能

SublimeText3 英文版

SublimeText3 英文版

推薦：為Win版本，支援程式碼提示！

ZendStudio 13.5.1 Mac

ZendStudio 13.5.1 Mac

強大的PHP整合開發環境

顯示更多

熱門話題

gmail信箱登陸入口在哪裡
7341
9
Java教學
1627
14
CakePHP 教程
1352
46
Laravel 教程
1265
25
PHP教程
1210
29
顯示更多