社区 学习 工具库 休闲
搜索
简体中文
首页 > 后端开发 > Python教程 > 正文

使用 Flask 和 MySQL 的任务管理器应用程序

Mary-Kate Olsen
发布： 2024-11-17 08:19:03
原创
412 人浏览过

项目概况

这个项目是一个使用 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/ 访问应用程序。
运行上述命令后，我能够在浏览器上访问该应用程序，如下图所示。

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

  • 删除任务：

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中文网其他相关文章！

来源：dev.to
上一篇：Python 3.x 中使用“str.replace()”的字符串替换有何变化？ 下一篇：如何在 Python 中读取用户输入和命令行参数？
本站声明
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系admin@php.cn
作者最新文章
最新问题
function_exists()无法判定自定义函数 function test()    {        return true;    }    if (function_exists('TEST')) {        ech...
来自于 2024-04-29 11:01:01
0
2
1870
google 浏览器 手机版显示的怎么实现 老师您好，google 浏览器怎么变成手机版样式的？
来自于 2024-04-23 00:22:19
0
10
2037
子窗口操作父窗口，输出没反应 前两句可执行，最后一句没法应
来自于 2024-04-19 15:37:47
0
1
1713
父窗口没有输出 document.onclick = function(){ window.opener.document.write('我是子窗口的输出');                  ...
来自于 2024-04-18 23:52:34
0
1
1614
关于CSS思维导图的课件在哪？ 课件
来自于 2024-04-16 10:10:18
0
0
1627
相关专题
更多>
热门推荐
热门教程
更多>
相关教程
热门推荐
最新课程
最新下载
更多>
网站特效
网站源码
网站素材
前端模板