項目概況
這個項目是一個使用 Flask 和 MySQL 構(gòu)建的任務(wù)管理器應(yīng)用程序。它提供了一個簡單的 RESTful API 來管理任務(wù)，演示了基本的 CRUD（創(chuàng)建、讀取、刪除）操作。
此應(yīng)用程序非常適合了解如何使用 Docker 將 Flask 應(yīng)用程序容器化并與 MySQL 數(shù)據(jù)庫連接。
特征
- 添加新任務(wù)
- 查看所有任務(wù)
- 通過ID刪除任務(wù)
燒瓶代碼：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 數(shù)據(jù)庫設(shè)置腳本
創(chuàng)建名為 init-db.sql 的 MySQL 腳本來設(shè)置數(shù)據(jù)庫和任務(wù)表：
要創(chuàng)建 init-db.sql 腳本，請按照以下步驟操作：
在項目目錄中創(chuàng)建一個新的文件：
導航到項目文件夾并創(chuàng)建一個名為 init-db.sql
的新文件
添加 SQL 命令來設(shè)置數(shù)據(jù)庫和任務(wù)表：
在文本編輯器中打開 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 容器啟動時，它將執(zhí)行 init-db.sql 腳本來設(shè)置 task_db 數(shù)據(jù)庫并創(chuàng)建 tasks 表。
注意：docker-entrypoint-initdb.d/目錄被MySQL容器用來執(zhí)行.sql 容器初始啟動期間的腳本。
解釋：
1。 version: '3': 指定正在使用的 Docker Compose 版本。
2。服務(wù)：
-
數(shù)據(jù)庫：
- image: mysql:5.7: 使用 MySQL 5.7 鏡像。
-
環(huán)境： 設(shè)置 MySQL 容器的環(huán)境變量：
- MYSQL_ROOT_PASSWORD: MySQL 的 root 密碼。
- MYSQL_DATABASE:啟動時創(chuàng)建的數(shù)據(jù)庫。
- ports: 將 MySQL 容器的端口 3306 映射到主機的端口 3306。
-
卷：
- db_data:/var/lib/mysql: 將數(shù)據(jù)庫數(shù)據(jù)保存在名為 db_data. 的 Docker 卷中
- ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql: 掛載 init-db.sql 腳本寫入 MYSQL 容器的初始化目錄，以便在容器啟動時運行。
-
網(wǎng)頁：
- build: .: 使用當前目錄中的 Dockerfile 為您的 Flask 應(yīng)用構(gòu)建 Docker 鏡像。
- 端口： 將 Flask 應(yīng)用程序的端口 5000 映射到主機的端口 5000。
- depends_on: 確保 db 服務(wù)在 Web 服務(wù)之前啟動。
- 環(huán)境： 設(shè)置 Flask 的環(huán)境變量。
- volumes: 將當前項目目錄掛載到容器內(nèi)的 /app 目錄中。 ### 卷部分： db_data: 定義一個命名卷 db_data 以在容器重新啟動之間保留 MySQL 數(shù)據(jù)。
Dockerfile：
定義 Flask 應(yīng)用程序的構(gòu)建指令：
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')
這個 Dockerfile 為 Flask 應(yīng)用程序設(shè)置了一個輕量級的 Python 環(huán)境：
1?；A(chǔ)鏡像： 使用 python:3.9-slim 來實現(xiàn)最短的 Python 運行時間。
工作目錄：將 /app 設(shè)置為工作目錄。
2。依賴項：復(fù)制requirements.txt并通過pip安裝依賴項。
3。工具安裝： 安裝 wait-for-it 以檢查服務(wù)準備情況。
4。應(yīng)用程序代碼： 將所有應(yīng)用程序代碼復(fù)制到容器中。
5。啟動命令： 運行 wait-for-it 以確保 MySQL DB (db:3306) 在啟動 app.py 之前準備就緒。
需求.txt 文件
此 requirements.txt 指定 Python 項目需要 Flask 框架 來構(gòu)建 Web 應(yīng)用程序和 mysql-connector-python 用于與 MySQL 數(shù)據(jù)庫 連接和交互。當在鏡像構(gòu)建過程中運行 pip install -rrequirements.txt 時，這些包將安裝在 Docker 容器中。這確保應(yīng)用程序擁有運行 Flask 服務(wù)器 并與 MySQL 數(shù)據(jù)庫 通信所需的工具。
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')
創(chuàng)建所有文件后，下一步是構(gòu)建并運行服務(wù)，使用以下命令來構(gòu)建和運行服務(wù)。
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
);
要以分離模式運行服務(wù)，我使用以下命令而不是 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:
當我想停止服務(wù)時，我使用命令
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"]
現(xiàn)在，一旦服務(wù)處于運行狀態(tài)，請運行命令
Flask mysql-connector-python
確保容器正在運行
現(xiàn)在是時候檢查服務(wù) API 以確保它們按預(yù)期工作了。
測試項目
通過 http://localhost:5000/ 訪問應(yīng)用程序。
運行上述命令后，我能夠在瀏覽器上訪問該應(yīng)用程序，如下圖所示。
您可以使用 Postman 或 curl 來測試 /tasks 端點的 POST、GET 和 DELETE 操作。在這種情況下，我將使用curl。
卷曲命令：
- 獲取任務(wù)：
GET 方法獲取所有任務(wù)。
docker-compose build docker-compose up
請注意，每當您在瀏覽器上運行 http://localhost:5000/tasks 時，它都會顯示您已添加的所有任務(wù)，如添加任務(wù)中所述。
- 添加任務(wù)：
POST 方法在數(shù)據(jù)庫中創(chuàng)建任務(wù)。
docker-compose up -d
這將向您的 Flask 應(yīng)用發(fā)送帶有任務(wù)描述的 POST 請求。如果任務(wù)添加成功，您應(yīng)該收到如下響應(yīng)：
docker-compose down
檢查瀏覽器的網(wǎng)絡(luò)選項卡或日志以驗證 POST 請求是否正確發(fā)出。
我運行了該命令幾次，并自定義了“簡單任務(wù)”部分以生成不同的輸出，這是我運行的命令，輸出可以在下圖中看到。
docker ps
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')
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
);
- 刪除任務(wù)：
DELETE 方法通過 ID 刪除任務(wù)。
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 的任務(wù)，如下圖所示，任務(wù) 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"]
結(jié)論
使用 Flask 和 MySQL 創(chuàng)建任務(wù)管理器應(yīng)用程序是了解 Web 服務(wù)開發(fā)、數(shù)據(jù)庫集成和 Docker 容器化基礎(chǔ)知識的絕佳方法。
該項目封裝了 Web 服務(wù)器和數(shù)據(jù)庫如何協(xié)同工作以提供無縫功能。
擁抱這種學習體驗，并將其用作更深層次的網(wǎng)絡(luò)和基于云的開發(fā)項目的墊腳石。
以上是使用 Flask 和 MySQL 的任務(wù)管理器應(yīng)用程序的詳細內(nèi)容。更多信息請關(guān)注PHP中文網(wǎng)其他相關(guān)文章！
熱AI工具
Undress AI Tool
免費脫衣服圖片
Undresser.AI Undress
人工智能驅(qū)動的應(yīng)用程序，用于創(chuàng)建逼真的裸體照片
AI Clothes Remover
用于從照片中去除衣服的在線人工智能工具。
Clothoff.io
AI脫衣機
Video Face Swap
使用我們完全免費的人工智能換臉工具輕松在任何視頻中換臉！
熱門文章
熱工具
記事本++7.3.1
好用且免費的代碼編輯器
SublimeText3漢化版
中文版，非常好用
禪工作室 13.0.1
功能強大的PHP集成開發(fā)環(huán)境
Dreamweaver CS6
視覺化網(wǎng)頁開發(fā)工具
SublimeText3 Mac版
神級代碼編輯軟件(SublimeText3)
Python的UNITDEST或PYTEST框架如何促進自動測試？
Jun 19, 2025 am 01:10 AM
Python的unittest和pytest是兩種廣泛使用的測試框架，它們都簡化了自動化測試的編寫、組織和運行。1.二者均支持自動發(fā)現(xiàn)測試用例并提供清晰的測試結(jié)構(gòu)：unittest通過繼承TestCase類并以test\_開頭的方法定義測試；pytest則更為簡潔，只需以test\_開頭的函數(shù)即可。2.它們都內(nèi)置斷言支持：unittest提供assertEqual、assertTrue等方法，而pytest使用增強版的assert語句，能自動顯示失敗詳情。3.均具備處理測試準備與清理的機制：un
如何將Python用于數(shù)據(jù)分析和與Numpy和Pandas等文庫進行操作？
Jun 19, 2025 am 01:04 AM
pythonisidealfordataanalysisionduetonumpyandpandas.1）numpyExccelSatnumericalComputationswithFast，多dimensionalArraysAndRaysAndOrsAndOrsAndOffectorizedOperationsLikenp.sqrt（）
什么是動態(tài)編程技術(shù)，如何在Python中使用它們？
Jun 20, 2025 am 12:57 AM
動態(tài)規(guī)劃（DP）通過將復(fù)雜問題分解為更簡單的子問題并存儲其結(jié)果以避免重復(fù)計算，來優(yōu)化求解過程。主要方法有兩種：1.自頂向下（記憶化）：遞歸分解問題，使用緩存存儲中間結(jié)果；2.自底向上（表格化）：從基礎(chǔ)情況開始迭代構(gòu)建解決方案。適用于需要最大/最小值、最優(yōu)解或存在重疊子問題的場景，如斐波那契數(shù)列、背包問題等。在Python中，可通過裝飾器或數(shù)組實現(xiàn)，并應(yīng)注意識別遞推關(guān)系、定義基準情況及優(yōu)化空間復(fù)雜度。
如何使用__ITER__和__NEXT __在Python中實現(xiàn)自定義迭代器？
Jun 19, 2025 am 01:12 AM
要實現(xiàn)自定義迭代器，需在類中定義__iter__和__next__方法。①__iter__方法返回迭代器對象自身，通常為self，以兼容for循環(huán)等迭代環(huán)境；②__next__方法控制每次迭代的值，返回序列中的下一個元素，當無更多項時應(yīng)拋出StopIteration異常；③需正確跟蹤狀態(tài)并設(shè)置終止條件，避免無限循環(huán)；④可封裝復(fù)雜邏輯如文件行過濾，同時注意資源清理與內(nèi)存管理；⑤對簡單邏輯可考慮使用生成器函數(shù)yield替代，但需結(jié)合具體場景選擇合適方式。
Python編程語言及其生態(tài)系統(tǒng)的新興趨勢或未來方向是什么？
Jun 19, 2025 am 01:09 AM
Python的未來趨勢包括性能優(yōu)化、更強的類型提示、替代運行時的興起及AI/ML領(lǐng)域的持續(xù)增長。首先，CPython持續(xù)優(yōu)化，通過更快的啟動時間、函數(shù)調(diào)用優(yōu)化及擬議中的整數(shù)操作改進提升性能；其次，類型提示深度集成至語言與工具鏈，增強代碼安全性與開發(fā)體驗；第三，PyScript、Nuitka等替代運行時提供新功能與性能優(yōu)勢；最后，AI與數(shù)據(jù)科學領(lǐng)域持續(xù)擴張，新興庫推動更高效的開發(fā)與集成。這些趨勢表明Python正不斷適應(yīng)技術(shù)變化，保持其領(lǐng)先地位。
如何使用插座在Python中執(zhí)行網(wǎng)絡(luò)編程？
Jun 20, 2025 am 12:56 AM
Python的socket模塊是網(wǎng)絡(luò)編程的基礎(chǔ)，提供低級網(wǎng)絡(luò)通信功能，適用于構(gòu)建客戶端和服務(wù)器應(yīng)用。要設(shè)置基本TCP服務(wù)器，需使用socket.socket()創(chuàng)建對象，綁定地址和端口，調(diào)用.listen()監(jiān)聽連接，并通過.accept()接受客戶端連接。構(gòu)建TCP客戶端需創(chuàng)建socket對象后調(diào)用.connect()連接服務(wù)器，再使用.sendall()發(fā)送數(shù)據(jù)和.recv()接收響應(yīng)。處理多個客戶端可通過1.線程：每次連接啟動新線程；2.異步I/O：如asyncio庫實現(xiàn)無阻塞通信。注意事
Python類中的多態(tài)性
Jul 05, 2025 am 02:58 AM
多態(tài)是Python面向?qū)ο缶幊讨械暮诵母拍睿浮耙环N接口，多種實現(xiàn)”，允許統(tǒng)一處理不同類型的對象。1.多態(tài)通過方法重寫實現(xiàn)，子類可重新定義父類方法，如Animal類的speak()方法在Dog和Cat子類中有不同實現(xiàn)。2.多態(tài)的實際用途包括簡化代碼結(jié)構(gòu)、增強可擴展性，例如圖形繪制程序中統(tǒng)一調(diào)用draw()方法，或游戲開發(fā)中處理不同角色的共同行為。3.Python實現(xiàn)多態(tài)需滿足：父類定義方法，子類重寫該方法，但不要求繼承同一父類，只要對象實現(xiàn)相同方法即可，這稱為“鴨子類型”。4.注意事項包括保持方
如何在Python中切片列表？
Jun 20, 2025 am 12:51 AM
Python列表切片的核心答案是掌握[start:end:step]語法并理解其行為。1.列表切片的基本格式為list[start:end:step]，其中start是起始索引（包含）、end是結(jié)束索引（不包含）、step是步長；2.省略start默認從0開始，省略end默認到末尾，省略step默認為1；3.獲取前n項用my_list[:n]，獲取后n項用my_list[-n:]；4.使用step可跳過元素，如my_list[::2]取偶數(shù)位，負step值可反轉(zhuǎn)列表；5.常見誤區(qū)包括end索引不
