APIFlask（或者apispec等）中怎么为Schema和route信息添加OpenAPI的spec并在Swagger UI/Redoc中呈现？

HelloFlask APIFlask
1

在写Djask过程中，在app类中添加了一个models属性app.models。其中存储了所有用户定义的SQLAlchemy模型类，并为之实现了app.get_model_by_name()方法，将URL中的模型类的名称字符串转换为具体的类。
现在我想要为app自动生成一个可以面向所有模型服务的api，因此我定义了一个新的蓝本admin_api

# ...
@admin_api.route("/<model>/<int:model_id>")
class ModelAPI(MethodView):
    def get(self, model: str, model_id: int):
        return current_app.get_model_by_name(model).query.get_or_404(model_id)

    def put(self, model: str, model_id: int, data: t.Dict[str, t.Any]):
        model = current_app.get_model_by_name(model)
        instance = model.query.get_or_404(model_id)
        for attr, value in data.items():
            try:
                setattr(instance, attr, value)
            except Exception as e:
                abort(400, str(e))
        db.session.commit()
        return instance

    def delete(self, model: str, model_id: int):
        model: t.Type[db.Model] = current_app.get_model_by_name(model)
        instance = model.query.get_or_404(model_id)
        db.session.delete(instance)
        db.session.commit()

post方法还没有实现。
我想为这个API实现OpenAPI的spec以便能够在Swagger UI中呈现出来。
但是由于模型的名称和ID是通过URL参数传递的，无法通过@input和@output装饰器直接生成。

目前大概思路如下：

  1. 可以用marshmallow_sqlalchemy为每个模型类生成Schema
  2. 再将这些Schema逐个添加到spec中，并添加对应的route（理论上可以实现因为模型类的名称已知）

上述第一步的实现可以参照文档，但是请问第二步应当怎么进行具体实现，目前没有想到好的解决办法

2

上段代码中仍然存在BUG，例如return current_app.get_model_by_name(model).query.get_or_404(model_id)没有考虑model不存在的情况，但是这暂时不是非常重要。

另外，如果需要查看Djask的其他代码来进行代码追溯的话，项目地址在
https://github.com/z-t-y/Djask

3

我有思路了：
将APIFlask._generate_spec()方法重写一遍，在调用super()._generate_spec()之后再使用APISpec和marshmallow_sqlalchemy将各个模型手动添加到spec中

4

贴一下具体代码：

class Djask:
    #...
    def _generate_spec(self) -> APISpec:
        """
        Add data models to the spec.

        .. versionadded:: 0.3.0
        """
        # call parental _generate_spec
        spec = super()._generate_spec()
        # get the prefix
        custom_prefix = self.config.get("ADMIN_PREFIX")
        prefix = custom_prefix if isinstance(custom_prefix, str) else "/admin"

        for m in self.models:
            m_name = m.__name__
            if m_name == "User":
                continue

            # register the schema to spec
            spec.components.schema(m_name, schema=m.to_schema())

            # define some common parameters, responses, etc.
            not_found = {
                "content": {"application/json": {"schema": "HTTPError"}},
                "description": "Not found",
            }
            bad_request = {
                "content": {"application/json": {"schema": "ValidationError"}},
                "description": "Validation error",
            }
            parameter_model_id = {
                "in": "path",
                "name": f"{m_name.lower()}_id",
                "schema": {"type": "integer"},
                "required": True,
            }
            response_model_schema = {
                "content": {"application/json": {"schema": m_name}}
            }
            # register the url route
            spec.path(
                path="{0}/api/{1}/{{{1}_id}}".format(prefix, m_name.lower()),
                operations=dict(
                    get=dict(
                        parameters=[parameter_model_id],
                        responses={
                            "200": response_model_schema,
                            "404": not_found,
                            "400": bad_request,
                        },
                        tags=["Admin.Admin_Api"],
                        summary=f"returns a {m_name.lower()}",
                    ),
                    put=dict(
                        parameters=[parameter_model_id],
                        responses={
                            "200": response_model_schema,
                            "404": not_found,
                            "400": bad_request,
                        },
                        requestBody={
                            "content": {"application/json": {"schema": m_name}}
                        },
                        tags=["Admin.Admin_Api"],
                        summary=f"updates a {m_name.lower()}",
                    ),
                    delete=dict(
                        parameters=[parameter_model_id],
                        responses={
                            "204": {"description": "Sucessful response"},
                            "404": not_found,
                        },
                        tags=["Admin.Admin_Api"],
                        summary=f"deletes a {m_name.lower()}",
                    ),
                ),
                description="Operate on {0}".format(m_name),
            )
        return spec
1 个赞