使用 Flask 和 SQLAlchemy 构建 REST API

2025-06-07

使用 Flask 和 SQLAlchemy 构建 REST API

点击此处阅读原文

Flask 是一个很棒的框架，它能让你用 Python 快速构建 Web 应用程序。它速度快、体积小，而且使用起来很有趣。在本教程中，我们将使用 Flask 框架和其他一些支持工具构建一个 RESTful API。

本教程的目标是从头开始了解构建 Flask 服务器的概念，学习如何通过对象关系映射器与 SQL 数据库通信，以及如何使用面向对象设计模式设计 RESTful API。

在本教程结束时，您将能够构建一个 REST API，该 API 可以使用创建、读取、更新和删除功能来管理博客文章数据。

入门

我已经在我的 GitHub 上发布了本教程的完成项目，您可以在这里查看它，或者通过运行以下命令将其克隆到您的机器中。

$ git clone https://github.com/rahmanfadhil/flask-rest-api.git

然后，创建一个新的 Python 虚拟环境并使用 Pip 安装依赖项。

$ python -m venv env
$ source env/bin/activate
(env) $ pip install -r requirements.txt

依赖项

根据其文档，Flask 是一个微框架。这意味着，与其他 Web 框架不同，Flask 没有提供太多花哨的开箱即用功能。

这使得 Flask 比其他语言更简单易学。但与此同时，我们经常被迫自己寻找解决方案来解决常见问题。例如，连接数据库、实现身份验证系统等等。

尽管如此，Flask 社区提供了一些有用的开源扩展，我们只需安装和配置它们就可以快速解决这些问题。

以下是我们将用于构建 API 的一些第三方包：

  • Flask SQLAlchemy：Flask 扩展增加了对SQLAlchemy的支持，SQLAlchemy 是一个对象关系映射器，它使我们更容易与 SQL 数据库交互。
  • Flask RESTful：另一个 Flask 扩展，用于使用面向对象的设计模式快速构建 REST API。
  • Flask Marshmallow：另一个与Marshmallow集成的 Flask 扩展，Marshmallow 是一个用于对象序列化的 Python 库。

从头开始的 Flask 项目

我们可以通过使用Python venv模块初始化一个新的虚拟环境来启动我们的项目。

$ python3 -m venv env
$ source env/bin/activate

其次，安装我们可以通过运行此命令通过 Pip 安装我们的依赖项。

$ pip install Flask \
    Flask-SQLAlchemy \
    Flask-RESTful \
    flask-marshmallow

最后，创建一个名为main.py（或任何你想命名的）的新 Python 文件。在这里，我们设置了全新的 Flask 服务器。

from flask import Flask

app = Flask(__name__)

if __name__ == '__main__':
    app.run(debug=True)

为了测试它，我们可以使用以下命令执行我们的 Python 脚本。

(env) $ python main.py

打开http://localhost:5000，你会看到一个 404 页面。

数据库设置

在这里，我们将使用 SQL 数据库来存储我们的博客文章数据。

出于学习目的，我将使用SQLite，这是一个小型 SQL 数据库实现，非常易于启动和运行。请记住，在生产环境中，您可能需要考虑使用更可靠的数据库，例如PostgreSQL或MySQL 。

要在 Flask 项目中设置 SQLAlchemy，我们可以导入flask_sqlalchemy包（之前已经安装），并将 Flaskapp变量包装到新SQLAlchemy对象中。我们还需要SQLALCHEMY_DATABASE_URI在 Flask 应用配置中指定要使用的数据库以及如何访问它。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy # new

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db' # new
db = SQLAlchemy(app) # new

if __name__ == '__main__':
    app.run(debug=True)

数据库模型

模型只不过是数据库的一种表示，你可以在其中存储、获取和操作数据。你可以把它想象成应用程序和数据库之间的中间人。它通常代表一个表/集合。

它允许你直接通过 Python 代码对特定表执行所有操作。这样你就不用再费力地处理 SQL 查询了。

让我们创建一个代表我们的博客文章数据的模型。

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(50))
    content = db.Column(db.String(255))

    def __repr__(self):
        return '<Post %s>' % self.title

这里，我们有一个id属性，它是一个自动生成的主键字段。然后，我们还有一个titleandcontent字段，它只是一个定义了最大长度的普通字符串字段。

这部分是可选的，但为了确保一切都显而易见，我们可以设置一种__repr__方法，使每个帖子对象都可以打印到控制台。

然后，我们需要设置模型的模式。这是必要的，因为我们需要将帖子对象解析为 JSON 响应。在这种情况下，我们可以使用flask_marshmallowpackage.json 。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow # new

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)
ma = Marshmallow(app) # new

# ...

if __name__ == '__main__':
    app.run(debug=True)

根据我们的Post模型创建一个新的棉花糖模式。

class PostSchema(ma.Schema):
    class Meta:
        fields = ("id", "title", "content")

post_schema = PostSchema()
posts_schema = PostSchema(many=True)

在此模式中，我们可以选择向用户公开哪些字段。如果您的模型包含一些敏感数据，您可能需要在此处将其排除。然后，我们还会在posts_schema和 中实例化它post_schema。用于posts_schema序列化帖子数组。否则，请使用post_schema。

下一个重要的步骤，大多数人（包括我自己）通常会忘记，那就是通过db.create_all在 Python 交互式环境 (REPL) 中调用 SQLAchemy 实例中的函数来初始化数据库模式。我们之所以在 REPL 中执行此操作，是因为我们只需要初始化一次数据库模式。

另外，请确保您处于正确的 Python 环境中。

(env) $ python
>>> from main import db
>>> db.create_all()
>>> exit()

RESTful 路由

最后，我们可以开始定义 RESTful 处理程序了。我们将使用 Flask-RESTful 包，这是一组帮助我们构建面向对象设计的 RESTful 路由的工具。

我们需要设置 Flask-RESTful 扩展才能在我们的 Flask 服务器中启动并运行。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api, Resource # new

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)
api = Api(app) # new

# ...

if __name__ == '__main__':
    app.run(debug=True)

创建一个新的 RESTful 资源，这样，我们就可以定义我们的业务逻辑，例如如何从数据库中获取数据、如何进行身份验证等等。

class PostListResource(Resource):
    def get(self):
        posts = Post.query.all()
        return posts_schema.dump(posts)

api.add_resource(PostListResource, '/posts')

在这里，我们接受一个GET请求，并根据模型进行查询以获取所有帖子Post。然后，我们利用posts_schema序列化数据库的数据并将其作为响应返回给客户端。

最后，我们使用api.add_resource方法注册我们的资源并定义路由端点。

启动服务器，向/posts端点发送请求，您将获得一个空数组。

$ curl http://localhost:5000/posts
[]

太棒了，现在是时候进行创建操作了。

# ...
from flask import Flask, request # change

# ...

class PostListResource(Resource):
    def get(self):
        posts = Post.query.all()
        return posts_schema.dump(posts)

    # new
    def post(self):
        new_post = Post(
            title=request.json['title'],
            content=request.json['content']
        )
        db.session.add(new_post)
        db.session.commit()
        return post_schema.dump(new_post)

# ...

我们在资源中创建一个新post方法，使用请求数据实例化一个新的 post 对象，并将记录保存到数据库。最后，我们将 post 数据作为响应返回给客户端。

/todos尝试向端点发送带有发布数据的POST 请求。

$ curl http://localhost:5000/posts \
    -X POST \
    -H "Content-Type: application/json" \
    -d '{"title":"Post 1", "content":"Lorem ipsum"}'
{
    "content": "Lorem ipsum",
    "id": 1,
    "title": "Post 1"
}

让我们创建一个新资源来获取单个帖子。

class PostResource(Resource):
    def get(self, post_id):
        post = Post.query.get_or_404(post_id)
        return post_schema.dump(post)

api.add_resource(PostResource, '/posts/<int:post_id>')

这里，我们也接受了一个GET请求，但不是查询所有帖子，而是只获取指定 id 的单个帖子。如果不存在，则会引发 404 错误。

尝试获取/todos/<int:id>，您将获得我们刚刚创建的帖子。

$ curl http://localhost:5000/posts/1
{
    "title": "Post 1",
    "content": "Lorem ipsum",
    "id": 1
}

并且，如果您请求一个不存在的 ID 的帖子，您将收到 404 错误。

$ curl http://localhost:5000/posts/12 -I
HTTP/1.0 404 NOT FOUND
...

现在，让我们添加剩余的 CRUD 操作、更新和删除。

class PostResource(Resource):
    # ...

    def patch(self, post_id):
        post = Post.query.get_or_404(post_id)

        if 'title' in request.json:
            post.title = request.json['title']
        if 'content' in request.json:
            post.content = request.json['content']

        db.session.commit()
        return post_schema.dump(post)

    def delete(self, post_id):
        post = Post.query.get_or_404(post_id)
        post.delete()
        db.session.commit()
        return '', 204

在该方法中patch，我们首先获取 post 对象（如果存在），然后更新请求体（request.json）中定义的属性。这就是为什么我们需要用表达式检查这两个属性in。使用 将更改保存到数据库db.session.commit()，并将更新数据发送到客户端。

在该方法中delete，我们也获取了 post 对象。但这次，我们只是从 post 对象中删除了带有方法的对象delete。保存更改，并且不向客户端返回任何内容（因为没有任何内容可显示）。

现在，我们已经启动并运行了所有功能，让我们测试一下！

更新帖子：

$ curl http://localhost:5000/posts/1 \
    -X PATCH \
    -H "Content-Type: application/json" \
    -d '{"title":"Updated Post", "content":"Updated post content"}'
{
    "content": "Updated post content",
    "id": 1,
    "title": "Updated Post"
}

删除帖子：

$ curl http://localhost:5000/posts/1 -X DELETE -I
HTTP/1.0 204 NO CONTENT
...
文章来源：https://dev.to/rahmanfadhil/build-rest-api-with-flask-sqlalchemy-4lbp
Devto 翻译
PREV
让我们建立一个 DevOps 项目：议程：DevOps 中的阶段：现在我们必须测试我们的应用程序：使用 ansible 进行配置管理：使用 Nagios 进行持续监控：感谢您阅读🤖🤖🤖
NEXT
创建 Node TypeScript 项目最简单的方法！