如何使用Flask-RESTX构建和文档化RESTful API

RESTful API 简介

REST（Representational State Transfer，表述性状态转移）是一种基于 RESTful API。它们为客户端与服务器之间的交互提供了统一且可预测的接口，从而使 Web 服务更加易于访问和维护。

设置开发环境

在开始构建 API 之前，我们需要先设置开发环境。

先决条件

Python 3.7 或更高版本

确保系统上已安装 Python，可以通过以下命令验证版本：

   python --version

虚拟环境

为项目创建虚拟环境以管理依赖项是一种推荐的实践：

   python -m venv venv
   source venv/bin/activate

# Windows 系统使用以下命令激活虚拟环境：# venvScriptsactivate

安装 Flask 和 Flask-RESTX

激活虚拟环境后，使用 pip 安装 Flask 和 Flask-RESTX：

pip install flask-restx

此命令会同时安装 Flask 和 Flask-RESTX，从而为构建和管理 API 提供必要支持。

安装和配置 Flask-RESTX

Flask-RESTX 是 Flask 的一个扩展，旨在简化 RESTful API 的构建，同时鼓励开发者以最少的配置实现最佳实践。

安装

如果尚未安装 Flask-RESTX，可以通过以下命令安装：

pip install flask-restx

配置

创建一个新的 Python 文件（例如 app.py），并添加以下代码进行基本配置：

from flask import Flask
from flask_restx import Api

app = Flask(__name__)
api = Api(app, version='1.0', title='示例 API', description='使用 Flask-RESTX 的 API 示例')if __name__ == '__main__':
    app.run(debug=True)

此代码初始化了一个 Flask 应用，并将其包装为 Flask-RESTX API 实例，为构建端点提供基础。

创建第一个 API 端点

接下来，我们通过创建一个简单的端点来了解 Flask-RESTX 的工作原理。

定义命名空间

命名空间用于组织 API 并避免端点名称冲突：

from flask_restx import Namespace, Resource

ns = Namespace('hello', description='Hello World 操作')

创建资源

资源对应于端点，并定义如何处理 HTTP 方法：

@ns.route('/')
class HelloWorld(Resource):
    def get(self):
        return {"message": "Hello, World!"}

注册命名空间

将命名空间注册到 API：

api.add_namespace(ns)

运行应用后，访问 http://localhost:5000/hello/，即可看到以下响应：

{"message": "Hello, World!"}

构建 Flask-RESTX 应用程序

随着应用程序的扩展，保持代码结构清晰有序至关重要。

初始化应用程序

在 app/__init__.py 中：

from flask import Flask
from flask_restx import Api

def create_app():
    app = Flask(__name__)
    api = Api(app)
    return app

在 run.py 中：

from app import create_app

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

这种模块化方法有助于分离关注点，使应用程序更易于维护和扩展。

实现 CRUD 操作

API 开发的核心。以下是一个简单的“Item”资源的 CRUD 实现。

定义模型

在 app/models/item_model.py 中：

from flask_restx import fields

def get_item_model(api):
    return api.model('Item', {
        'id': fields.Integer(readOnly=True, description='唯一标识符'),
        'name': fields.String(required=True, description='项目名称'),
        'price': fields.Float(required=True, description='项目价格')
    })

实现资源

在 app/controllers/item_controller.py 中：

@ns.response(204, '项目已删除')

def delete(self, id):

    global items

    item = next((item for item in items if item['id'] == id), None)

    if item:

        items = [i for i in items if i['id'] != id]

        return '', 204

    ns.abort(404, message="项目未找到")

ns = Namespace('items', description='项目操作')

item_model = get_item_model(ns)



items = []

item_id_counter = 1



parser = reqparse.RequestParser()

parser.add_argument('name', type=str, required=True, help='项目名称')

parser.add_argument('price', type=float, required=True, help='项目价格')



@ns.route('/')

class ItemList(Resource):

    @ns.marshal_list_with(item_model)

    def get(self):

        return items



    @ns.expect(item_model)

    @ns.marshal_with(item_model, code=201)

    def post(self):

        global item_id_counter

        args = parser.parse_args()

        item = {'id': item_id_counter, 'name': args['name'], 'price': args['price']}

        items.append(item)

        item_id_counter += 1

        return item, 201



@ns.route('/')

@ns.response(404, '项目未找到')

@ns.param('id', '项目标识符')

class Item(Resource):

    @ns.marshal_with(item_model)

    def get(self, id):

        item = next((item for item in items if item['id'] == id), None)

        if item:

            return item

        ns.abort(404, message="项目未找到")



    @ns.expect(item_model)

    @ns.marshal_with(item_model)

    def put(self, id):

        item = next((item for item in items if item['id'] == id), None)

        if item:

            args = parser.parse_args()

            item.update({'name': args['name'], 'price': args['price']})

            return item

        ns.abort(404, message="项目未找到")



    @ns.response(204, '项目已删除')

    def delete(self, id):

        global items

        item = next((item for item in items if item['id'] == id), None)

        if item:

            items = [i for i in items if i['id'] != id]

            return '', 204

        ns.abort(404, message="项目未找到")

此实现提供了以下功能：

GET /items/：获取项目列表。
POST /items/：添加新项目。
GET /items/{id}：根据 ID 获取项目。
PUT /items/{id}：更新指定 ID 的项目。
DELETE /items/{id}：删除指定 ID 的项目。

使用 Flask-RESTX 验证和序列化数据

数据验证和序列化是确保 API 数据一致性的重要环节。Flask-RESTX 提供了模型和装饰器来简化这一过程。

使用模型验证输入

定义模型以指定输入和输出的格式：

item_model = api.model('Item', {
    'id': fields.Integer(readOnly=True, description='唯一标识符'),
    'name': fields.String(required=True, description='项目名称'),
    'price': fields.Float(required=True, description='项目价格')
})

通过 @ns.expect(item_model) 装饰器，Flask-RESTX 会自动验证传入的 JSON 数据。

序列化响应

使用 @ns.marshal_with 装饰器格式化返回数据：

@ns.marshal_with(item_model)
def get(self, id):

# 返回项目数据

这确保了响应数据符合预期的结构。

使用 Flask-RESTX 生成 API 文档

全面的 API 文档对于开发者和用户至关重要。Flask-RESTX 提供了内置的 Swagger UI 支持，自动生成交互式文档。

自动生成文档

运行 Flask-RESTX 应用后，访问根 URL（例如 http://localhost:5000/）即可查看自动生成的 Swagger 文档。

自定义文档

通过以下装饰器可以增强文档内容：

@api.doc()：添加额外的文档信息。
@api.response()：记录响应状态码及描述。
@api.marshal_with()：定义响应数据的格式。
@api.expect()：定义请求数据的格式。

总结

通过本文，我们学习了如何使用 Flask-RESTX CRUD 操作的实现也为构建高效、可维护的 API 提供了基础。希望本文能为您的 API 开发提供帮助！

原文链接: https://apidog.com/blog/flask-restx/