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

碧海醫心

发布时间：2024-11-16 12:03:02

751人浏览过

来源于dev.to

转载

项目概况

这个项目是一个使用 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/', 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 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"]

这个 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 数据库 通信所需的工具。

flask
mysql-connector-python

创建所有文件后，下一步是构建并运行服务，使用以下命令来构建和运行服务。

极限网络办公Office Automation

专为中小型企业定制的网络办公软件,富有竞争力的十大特性： 1、独创 web服务器、数据库和应用程序全部自动傻瓜安装，建立企业信息中枢只需3分钟。 2、客户机无需安装专用软件，使用浏览器即可实现全球办公。 3、集成Internet邮件管理组件，提供web方式的远程邮件服务。 4、集成语音会议组件，节省长途话费开支。 5、集成手机短信组件，重要信息可直接发送到员工手机。 6、集成网络硬

下载

docker-compose build
docker-compose up

要以分离模式运行服务，我使用以下命令而不是 docker-compose up

docker-compose up -d

当我想停止服务时，我使用命令

docker-compose down

现在，一旦服务处于运行状态，请运行命令

docker ps

确保容器正在运行

现在是时候检查服务 api 以确保它们按预期工作了。

测试项目

通过 http://localhost:5000/ 访问应用程序。
运行上述命令后，我能够在浏览器上访问该应用程序，如下图所示。

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

您可以使用 postman 或 curl 来测试 /tasks 端点的 post、get 和 delete 操作。在这种情况下我会使用curl。

卷曲命令：

获取任务：

get 方法获取所有任务。

curl http://localhost:5000/tasks

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

请注意，每当您在浏览器上运行 http://localhost:5000/tasks 时，它都会显示您已添加的所有任务，如添加任务中所述。

添加任务：

post 方法在数据库中创建任务。

curl -x post http://localhost:5000/tasks -h "content-type: application/json" -d '{"description": "sample task"}'

这将向您的 flask 应用发送带有任务描述的 post 请求。如果任务添加成功，您应该收到如下响应：

{
    "message": "task added successfully",
    "task_id": 1
}

检查浏览器的网络选项卡或日志以验证 post 请求是否正确发出。

我运行了该命令几次，并自定义了“简单任务”部分以生成不同的输出，这是我运行的命令，输出可以在下图中看到。

 curl -x post http://localhost:5000/tasks -h "content-type: application/json" -d '{"description": "my name is matthew tarfa am the cloud chief"}'

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

 curl -x post http://localhost:5000/tasks -h "content-type: application/json" -d '{"description": "welcome to my world!"}'

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

 curl -x post http://localhost:5000/tasks -h "content-type: application/json" -d '{"description": "i love devops!"}'

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

删除任务：

delete 方法通过 id 删除任务。

curl -x delete http://localhost:5000/tasks/1

我运行了以下命令来删除 id 为 4 的任务，如下图所示，任务 4 已被删除。

curl -X DELETE http://localhost:5000/tasks/4

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

结论

使用 flask 和 mysql 创建任务管理器应用程序是了解 web 服务开发、数据库集成和 docker 容器化基础知识的绝佳方法。

该项目封装了 web 服务器和数据库如何协同工作以提供无缝功能。

拥抱这种学习体验，并将其用作更深层次的网络和基于云的开发项目的垫脚石。

sqlalchemy 如何写“插入或更新”并返回受影响行数

sqlalchemy 如何写“存在则更新，不存在则插入”（upsert）

sqlalchemy 如何写批量插入或忽略（on conflict do nothing）

如何将本地文本文件内容插入 MySQL 数据表

如何将本地文件内容插入 MySQL 数据表