使用 Flask 和 MySQL 的任务管理器应用程序
项目概况
这个项目是一个使用 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/<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 数据库设置脚本
创建名为 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/<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 应用程序设置了一个轻量级的 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/<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
);
要以分离模式运行服务，我使用以下命令而不是 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/ 访问应用程序。
运行上述命令后，我能够在浏览器上访问该应用程序，如下图所示。
您可以使用 Postman 或 curl 来测试 /tasks 端点的 POST、GET 和 DELETE 操作。在这种情况下，我将使用curl。
卷曲命令：
- 获取任务：
GET 方法获取所有任务。
docker-compose build docker-compose up
请注意，每当您在浏览器上运行 http://localhost:5000/tasks 时，它都会显示您已添加的所有任务，如添加任务中所述。
- 添加任务：
POST 方法在数据库中创建任务。
docker-compose up -d
这将向您的 Flask 应用发送带有任务描述的 POST 请求。如果任务添加成功，您应该收到如下响应：
docker-compose down
检查浏览器的网络选项卡或日志以验证 POST 请求是否正确发出。
我运行了该命令几次，并自定义了“简单任务”部分以生成不同的输出，这是我运行的命令，输出可以在下图中看到。
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
);
- 删除任务：
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"]
结论
使用 Flask 和 MySQL 创建任务管理器应用程序是了解 Web 服务开发、数据库集成和 Docker 容器化基础知识的绝佳方法。
该项目封装了 Web 服务器和数据库如何协同工作以提供无缝功能。
拥抱这种学习体验，并将其用作更深层次的网络和基于云的开发项目的垫脚石。
以上是使用 Flask 和 MySQL 的任务管理器应用程序的详细内容。更多信息请关注PHP中文网其他相关文章！
热AI工具
Undresser.AI Undress
人工智能驱动的应用程序，用于创建逼真的裸体照片
AI Clothes Remover
用于从照片中去除衣服的在线人工智能工具。
Undress AI Tool
免费脱衣服图片
Clothoff.io
AI脱衣机
Video Face Swap
使用我们完全免费的人工智能换脸工具轻松在任何视频中换脸！
热门文章
热工具
记事本++7.3.1
好用且免费的代码编辑器
SublimeText3汉化版
中文版，非常好用
禅工作室 13.0.1
功能强大的PHP集成开发环境
Dreamweaver CS6
视觉化网页开发工具
SublimeText3 Mac版
神级代码编辑软件(SublimeText3)
Python与C：学习曲线和易用性
Apr 19, 2025 am 12:20 AM
Python更易学且易用，C 则更强大但复杂。1.Python语法简洁，适合初学者，动态类型和自动内存管理使其易用，但可能导致运行时错误。2.C 提供低级控制和高级特性，适合高性能应用，但学习门槛高，需手动管理内存和类型安全。
学习Python：2小时的每日学习是否足够？
Apr 18, 2025 am 12:22 AM
每天学习Python两个小时是否足够？这取决于你的目标和学习方法。1)制定清晰的学习计划，2)选择合适的学习资源和方法，3)动手实践和复习巩固，可以在这段时间内逐步掌握Python的基本知识和高级功能。
Python vs.C：探索性能和效率
Apr 18, 2025 am 12:20 AM
Python在开发效率上优于C ，但C 在执行性能上更高。1.Python的简洁语法和丰富库提高开发效率。2.C 的编译型特性和硬件控制提升执行性能。选择时需根据项目需求权衡开发速度与执行效率。
Python vs. C：了解关键差异
Apr 21, 2025 am 12:18 AM
Python和C 各有优势，选择应基于项目需求。1)Python适合快速开发和数据处理，因其简洁语法和动态类型。2)C 适用于高性能和系统编程，因其静态类型和手动内存管理。
Python标准库的哪一部分是：列表或数组？
Apr 27, 2025 am 12:03 AM
pythonlistsarepartofthestAndArdLibrary，herilearRaysarenot.listsarebuilt-In，多功能，和Rused ForStoringCollections，而EasaraySaraySaraySaraysaraySaraySaraysaraySaraysarrayModuleandleandleandlesscommonlyusedDduetolimitedFunctionalityFunctionalityFunctionality。
Python：自动化，脚本和任务管理
Apr 16, 2025 am 12:14 AM
Python在自动化、脚本编写和任务管理中表现出色。1)自动化：通过标准库如os、shutil实现文件备份。2)脚本编写：使用psutil库监控系统资源。3)任务管理：利用schedule库调度任务。Python的易用性和丰富库支持使其在这些领域中成为首选工具。
科学计算的Python：详细的外观
Apr 19, 2025 am 12:15 AM
Python在科学计算中的应用包括数据分析、机器学习、数值模拟和可视化。1.Numpy提供高效的多维数组和数学函数。2.SciPy扩展Numpy功能，提供优化和线性代数工具。3.Pandas用于数据处理和分析。4.Matplotlib用于生成各种图表和可视化结果。
Web开发的Python：关键应用程序
Apr 18, 2025 am 12:20 AM
Python在Web开发中的关键应用包括使用Django和Flask框架、API开发、数据分析与可视化、机器学习与AI、以及性能优化。1.Django和Flask框架：Django适合快速开发复杂应用，Flask适用于小型或高度自定义项目。2.API开发：使用Flask或DjangoRESTFramework构建RESTfulAPI。3.数据分析与可视化：利用Python处理数据并通过Web界面展示。4.机器学习与AI：Python用于构建智能Web应用。5.性能优化：通过异步编程、缓存和代码优
