使用 Flask、SQLAlchemy、Postgres、Docker、Docker Compose 的 Python CRUD Rest API

让我们在 Python 中创建一个 CRUD Rest API，使用以下命令：

• Flask（Python Web 框架）
• SQLAlchemy（ORM）
• Postgres（数据库）
• Docker（容器化）
• Docker Compose（在容器中运行应用程序和数据库）

如果您更喜欢视频版本：

所有代码均可在 GitHub 存储库中找到（视频描述中的链接）：https://youtube.com/live/fHQWTsWqBdE

🏁 简介

以下是我们要创建的应用程序架构图：

我们将为基本的 CRUD 操作创建 5 个端点：

• 创造
• 阅读全部
• 阅读一篇
• 更新
• 删除

以下是我们正在采取的步骤：

1. 使用 SQLALchemy 作为 ORM 创建 Flask 应用程序。
2. 将 Flask 应用程序 Docker 化，编写 Dockerfile 和 docker-compose.yml 文件来运行应用程序和数据库
3. 使用 Docker Compose 在容器中运行 Postgres 数据库，并使用 TablePlus 进行测试
4. 使用 Docker Compose 在容器中运行 Flask 应用程序，并使用 Postman 进行测试

我们将提供一步一步的指南，以便您可以跟随。

🏁 使用 SQLALchemy 作为 ORM 创建 Flask 应用程序

创建新文件夹：

mkdir flask-crud-api

进入文件夹：

cd flask-crud-api

使用您喜欢的 IDE 打开文件夹。我使用的是 VSCode，因此我将使用以下命令：

code .

对于 Flask 应用程序，我们只需要 4 个文件，包括容器化。

您可以通过不同的方式创建这些文件。其中一种是手动创建，另一种是使用命令行创建：

touch requirements.txt app.py Dockerfile docker-compose.yml

您的文件夹结构应如下所示：

🗒️requirements.txt 文件

requirements.txt 文件包含项目的所有依赖项。在本例中，我们只需要 3 个。

让我们将它们添加到requirements.txt文件中：

flask
psycopg2-binary
Flask-SQLAlchemy

依赖关系的简要说明：

flask是我们将要使用的 Python Web 框架。

psycopg2-binary是与 Postgres 数据库建立连接的驱动程序。

Flask-SQLAlchemy是用于对数据库进行查询的 ORM。

🐍 app.py 文件

app.py 文件是应用程序的主文件：它包含应用程序的所有端点和逻辑。

按如下方式填充 app.py 文件：

from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy
from os import environ

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = environ.get('DB_URL')
db = SQLAlchemy(app)

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def json(self):
        return {'id': self.id,'username': self.username, 'email': self.email}

db.create_all()

#create a test route
@app.route('/test', methods=['GET'])
def test():
  return make_response(jsonify({'message': 'test route'}), 200)


# create a user
@app.route('/users', methods=['POST'])
def create_user():
  try:
    data = request.get_json()
    new_user = User(username=data['username'], email=data['email'])
    db.session.add(new_user)
    db.session.commit()
    return make_response(jsonify({'message': 'user created'}), 201)
  except e:
    return make_response(jsonify({'message': 'error creating user'}), 500)

# get all users
@app.route('/users', methods=['GET'])
def get_users():
  try:
    users = User.query.all()
    return make_response(jsonify([user.json() for user in users]), 200)
  except e:
    return make_response(jsonify({'message': 'error getting users'}), 500)

# get a user by id
@app.route('/users/<int:id>', methods=['GET'])
def get_user(id):
  try:
    user = User.query.filter_by(id=id).first()
    if user:
      return make_response(jsonify({'user': user.json()}), 200)
    return make_response(jsonify({'message': 'user not found'}), 404)
  except e:
    return make_response(jsonify({'message': 'error getting user'}), 500)

# update a user
@app.route('/users/<int:id>', methods=['PUT'])
def update_user(id):
  try:
    user = User.query.filter_by(id=id).first()
    if user:
      data = request.get_json()
      user.username = data['username']
      user.email = data['email']
      db.session.commit()
      return make_response(jsonify({'message': 'user updated'}), 200)
    return make_response(jsonify({'message': 'user not found'}), 404)
  except e:
    return make_response(jsonify({'message': 'error updating user'}), 500)

# delete a user
@app.route('/users/<int:id>', methods=['DELETE'])
def delete_user(id):
  try:
    user = User.query.filter_by(id=id).first()
    if user:
      db.session.delete(user)
      db.session.commit()
      return make_response(jsonify({'message': 'user deleted'}), 200)
    return make_response(jsonify({'message': 'user not found'}), 404)
  except e:
    return make_response(jsonify({'message': 'error deleting user'}), 500)

解释：

我们正在进口：

• Flask 作为框架
• 处理 HTTP 请求
• jsonify 用于处理 json 格式，不是 Python 原生的
• make_response 处理 HTTP 响应
• flask_sqlalchemy 处理数据库查询
• environ 处理环境变量

我们正在创建 Flask 应用，并通过设置名为“DB_URL”的环境变量来配置数据库。稍后我们将在 docker-compose.yml 文件中设置它。

然后我们创建一个带有 id、用户名和电子邮件的用户类。当我们创建用户时，SQLAlchemy 会自动增加 id。该__tablename__ = 'users'行用于定义数据库中表的名称

重要的一行是db.create_all()。这将使数据库与定义的模型同步，例如创建一个“用户”表。

那么我们有 6 个端点

• 测试：只是一条测试路线
• 创建用户：创建具有用户名和电子邮件的用户
• 获取所有用户：获取数据库中的所有用户
• 获取一个用户：通过 id 获取一个用户
• 更新一个用户：通过 id 更新一个用户
• 删除一个用户：通过id删除一个用户

所有路由都有错误处理，例如，如果未找到用户，我们将返回 404 HTTP 响应。

您可以在此处查看视频说明

🐳 将 Flask 应用程序 Docker 化

让我们填充Dockerfile：

Dockerfile：

FROM python:3.6-slim-buster

WORKDIR /app

COPY requirements.txt ./

RUN pip install -r requirements.txt

COPY . .

EXPOSE 4000

CMD [ "flask", "run", "--host=0.0.0.0", "--port=4000"]

FROM设置要使用的基础镜像。在本例中，我们使用 Python 3.6 slim buster 镜像

WORKDIR设置图像内的工作目录

COPY requirements.txt ./将 requirements.txt 文件复制到工作目录

RUN pip install -r requirements.txt安装要求

COPY . .将当前目录中的所有文件复制到工作目录

EXPOSE 4000暴露端口 4000

CMD [ "flask", "run", "--host=0.0.0.0", "--port=4000"]设置容器启动时运行的命令

🐳🐳Docker 撰写

“Docker compose” 这个术语可能有点令人困惑，因为它既指文件，又指一组 CLI 命令。这里我们用这个术语来指文件。

填充docker-compose.yml文件：

version: "3.9"

services:
  flask_app:
    container_name: flask_app
    image: dockerhub-flask_live_app:1.0.0
    build: .
    ports:
      - "4000:4000"
    environment:
      - DB_URL=postgresql://postgres:postgres@flask_db:5432/postgres
    depends_on:
      - flask_db
  flask_db:
    container_name: flask_db
    image: postgres:12
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_USER=postgres
      - POSTGRES_DB=postgres
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata: {}

我们刚刚定义了两个服务flask_app：flask_db

flask_app是我们刚刚 docker 化的 Flask 应用程序

flask_db是一个 Postgres 容器，用于存储数据。我们将使用 Postgres 官方镜像

解释：

version是 docker-compose 文件的版本。我们使用的是 3.9 版本

services是我们要运行的服务（容器）列表。在本例中，我们有两个服务：flask_app 和 flask_db

container_name是容器的名称。这不是强制性的，但为容器指定名称是一个好习惯。容器之间通过名称相互识别，因此为想要通信的容器指定名称非常重要。

image是我们要使用的镜像名称。我建议将“dockerhub-”替换为您的 Dockerhub 帐户（免费）。

build是 Dockerfile 的路径。在本例中，它是当前目录，因此我们使用.

ports是我们要公开的端口列表。在本例中，我们公开了 flask_app 容器的 4000 端口和 flask_db 容器的 5432 端口。格式如下：host_port:container_port

depends_on是我们希望在此之前启动的服务列表。在本例中，我们希望在 flask_app 容器之前启动 flask_db 容器

environment定义环境变量。对于 flask_app，我们将使用数据库 URL 来配置配置。对于 flask_db 容器，我们将定义一些在使用 Postgres 容器时需要定义的环境变量（此处无法更改键，因为我们使用的是 Postgres 团队定义的 Postgres 镜像）。

volumes在 flask_db 中定义了一个用于持久化的命名卷。容器根据定义是短暂的，因此我们需要这个附加功能来确保在容器被移除时数据能够持久化（容器只是一个进程）。

volumes文件末尾是我们要创建的卷列表。在本例中，我们将创建一个名为 的卷pgdata。格式如下volume_name: {}

👟 运行 Postgres 容器并使用 TablePlus 进行测试

要运行 Postgres 容器，请输入：

docker compose up -d flask_db

该-d标志以分离模式运行容器，因此它将在后台运行。

您应该看到类似这样的内容：

Docker 正在从我们的本地机器上拉取（下载）Postgres 镜像，并且基于该 Postgres 镜像运行一个容器。

要检查容器是否正在运行，请输入：

docker compose logs

如果一切正常，你应该会看到类似这样的内容：

如果最后一行是LOG: database system is ready to accept connections，则表示容器正在运行并且 Postgres 服务器已准备好接受连接。

但为了确保万无一失，我们再做一次测试。

要显示所有容器（正在运行和已停止的容器），请输入：

docker ps -a

输出应该类似于此：

现在，我们可以使用任何工具来测试数据库连接。我个人使用 TablePlus。

使用以下配置：

主持人：localhost

港口：5432

用户：postgres

密码：postgres

数据库：postgres

然后点击“测试”（右下角）。

如果您收到“连接正常”消息，则可以继续。

您也可以点击“连接”，然后会看到一个空的数据库。这是正确的。

🔨构建并运行 Flask 应用程序

现在，让我们构建并运行 Flask 应用程序。

让我们回到所在的文件夹docker-compose.yml并输入：

docker compose build

这应该会构建 flask_app 镜像，其名称在“image”值中定义。在我的例子中，它是 francescoxx/flask_live_app:1.0.0，因为这是我的 Dockerhub 用户名。你应该将“francescoxx”替换为你的 Dockerhub 用户名。

您还可以逐层查看 Docker 构建镜像的所有步骤。您可能认识其中一些步骤，因为我们在 Dockerfile 中定义了它们。

现在，要检查图像是否已成功构建，请输入：

docker images

使用我们刚刚构建的图像，我们应该看到类似的结果：

⚗️ 运行 flask_app 服务

我们快完成了，但最后一步是根据我们刚刚构建的图像运行一个容器。

为此，我们只需输入：

docker compose up flask_app

在这种情况下，我们不使用 -d 标志，因为我们想在终端中查看日志。

我们应该看到类似这样的内容：

🔍 测试应用程序

让我们测试一下我们的应用程序。首先，让我们打开任意浏览器并访问localhost:4000/test

您应该会看到这个结果：

（请注意，如果您访问，localhost:4000您会收到一个错误，因为没有与此端点关联的路由，但收到错误是一件好事，因为这意味着服务器正在运行！）

现在是时候使用 Postman 测试所有端点了。您可以随意使用任何您想要的工具。

如果我们向 GET 请求，localhost:4000/users我们将得到一个空数组。这是正确的

📝 创建用户

现在让我们创建一个用户，并发出一个 POST 请求，localhost:4000/users请求主体如下：

让我们创建另一个：

再说一句：

📝 获取所有用户

现在，让我们发出 GET 请求来localhost:4000/users获取所有用户：

我们刚刚创建了 3 个用户。

📝 获取特定用户

如果您想获取特定用户，您可以向 发出 GET 请求localhost:4000/users/<user_id>。

例如，要获取 ID 为 2 的用户，你可以向localhost:4000/users/2

📝 更新用户

如果您想更新用户，您可以向发出 PUT 请求localhost:4000/users/<user_id>。

例如，要更新 ID 为 2 的用户，您可以发出 PUT 请求，并localhost:4000/users/2使用以下主体作为请求主体：

要检查用户是否已更新，您可以发出 GET 请求localhost:4000/users/2

📝 删除用户

要删除用户，您可以向 发出 DELETE 请求localhost:4000/users/<user_id>。

例如，要删除 ID 为 2 的用户，您可以向localhost:4000/users/2

要检查用户是否已被删除，您可以向localhost:4000/users

正如您所见，id 为 2 的用户不再存在。

🏁 结论

我们成功了！我们用 Python 构建了一个 CRUD rest API，使用了 Flask、SQLAlchemy、Postgres、Docker 和 Docker Compose。

这只是一个例子，但您可以用它作为构建自己的应用程序的起点。

如果您更喜欢视频版本：

所有代码均可在 GitHub 存储库中找到（视频描述中的链接）：https://youtube.com/live/fHQWTsWqBdE

就这样。

如果您有任何疑问，请在下面发表评论。

弗朗西斯科