从 restful 到 graphql， 从 rpc 到 pydantic-resolve， 谈谈前后端紧耦合项目接口的（可能）演进方向

首先阐述一下当前对 restful, gql 和 rpc 的主流表述和看法

• restful 接口“普遍”是扁平的，于是前端需要调用多个接口来拼装数据
• gql 解决的是这种多接口数据拼接的需求，通过单一接口 + 查询体，让前端描述正好所需的数据来获取所需数据。
• rpc 解决的是类型和调用方法，构建方式不限形式的话，通过 openapi 生成 ts sdk 是很方便的一种手段。前端无需关心查询，直接获取展示用的数据。（trpc 很火，但是后端只能 node, 这就局限很大）

优势

restful 和 gql，功能上适合用来提供稳定的 public API 接口，比如 github，confluence 等，可以从接口和文档获取到相关信息。（所以往往需要版本号）

非常适合基于这些公共接口做二次开发，这些接口扮演了第三方 api 的角色，可以等价 db 查询之类的数据查询。

rpc 适合用在 前后端关系紧密的项目，表现为前后端修改是相互联动的，对这些接口来说，通常是不需要考虑版本号之类的需求的，后端改了，前端也要对应作出修改。

rpc 可以通过同步类型和方法来快速通知前端变更，使两边信息维持同步，降低了前端获取数据的复杂度（专心负责展现）。

问题

如果把 restful 用在这种类型项目上，因为后端总面向资源设计 API，导致前端无法舒舒服服的使用数据，要操心数据拼接，另一方面数据溯源也会变麻烦。

如果把 gql 用在这类项目中，前端拼接数据的场景少了，但是后端需要构建一个大而全的综合查询接口，工作量就上去了。另外 gql 虽然能方便的构建树形关联的数据，但它只能层层往下获取数据，如果前端存在层级数据的聚合或者转换的需求，依然会比较麻烦。更不用说前端还需要维护好一套 query 语句，在后端修改之后还需要连带着修改 query。此外还有引入 gql 相关框架的成本。

比如 comment_count，让后端处理就会比较麻烦，无法充分利用已查询到的同级数据 comments, 只能另外发请求来计算。

query {
    MyBlogSite {
        name
        blogs {
            id
            title
            comments {
                id
                content
            }
            comment_count  # comments count for each blog
        }
        comment_count  # overall comments count
    }
}

rpc 可以简化前后端沟通成本，但构建视图数据上并没有额外帮助。

所以麻烦事最终落在了构建前端视图上 ，精准构建前端视图数据往往不太方便，这种杂活往往比较琐碎，容易变化，如果遇上层级间的数据转换，也会很麻烦。 这也是后端不愿意负责的原因。

常见做法一类是把前端多 API / 多查询的数据拼接杂活在后端用过程式处理代办了，另一类是借助 ORM 来获取关联数据，借助 ORM 和 借助 gql 的本质差不多，都会遇到对获取数据的后处理不方便，以及重新调整层级结构比较麻烦的问题。

那么是否有好的方案，可以让这种麻烦事变简单呢？

方案

思路藏在 gql 中，既通过申明的方式来描述数据：

基于 pydantic 实现了一个 python 版本的方案：pydantic-resolve, 具体如下：

class MyBlogSite(BaseModel):
    blogs: list[Blog] = []
    async def resolve_blogs(self):
        return await get_blogs()

    comment_count: int = 0
    def post_comment_count(self):
        return sum([b.comment_count for b in self.blogs])

class Blog(BaseModel):
    id: int
    title: str

    comments: list[Comment] = []
    def resolve_comments(self, loader=LoaderDepend(blog_to_comments_loader)):
        return loader.load(self.id)

    comment_count: int = 0
    def post_comment_count(self):
        return len(self.comments)

class Comment(BaseModel):
    id: int
    content: str
    
async def main():
    my_blog_site = MyBlogSite(name: "tangkikodo's blog")
    my_blog_site = await Resolver().resolve(my_blog_site)

忽略 resolve_ 和 post_ 方法的话，上述代码只是描述了 Site -> blog -> comment 的层级结构。

加上 resolve_ 方法之后，他就能从方法返回值获取到数据，获取数据的过程是递归的，resolve_blogs 的过程中会触发 resolve_comments.

直到 blogs 的子孙信息都被获取完毕之后才会结束。（用来解决 N+1 query 的 dataloader 和 gql 里面用的是一样的）

加上 post_ 方法之后，每个层级的 resolve_ 获取完数据之后，可以在 post_ 方法中对该层的数据做处理，每个 blog 的 comments 长度就能在此时计算出来，最终到顶层的 comment_count 汇总到一起。

在这么两个简单的方法的加持下，gql 不擅长的后处理环节就解决了。

{
  "blogs": [
    {
      "id": 1,
      "title": "what is pydantic-resolve",
      "comments": [
        {
          "id": 1,
          "content": "its interesting"
        },
        {
          "id": 2,
          "content": "i need more example"
        }
      ],
      "comment_count": 2
    },
    {
      "id": 2,
      "title": "what is composition oriented development pattarn",
      "comments": [
        {
          "id": 3,
          "content": "what problem does it solved?"
        },
        {
          "id": 4,
          "content": "interesting"
        }
      ],
      "comment_count": 2
    }
  ],
  "comment_count": 4
}

借助 pydantic + fastapi, 可以生成 openapi.json, 然后可以用 openapi-typescript-codegen 来创建 rpc 风格的前端 sdk。

而这，也许是处理 前后端关系紧密的项目 的一种新的思路。

1. 申明式让数据结构始终保持清晰
2. resolve 负责获取数据，post 负责后处理，利用好层级关系。
3. 还有其他一系列功能，用来构建数据，比如 读取祖先字段，收集子孙字段等。（可用于调整层级）
4. 使用 context 来提供参数
5. schema 可复用（类似 fragment)

如果你看到了这里，我表示深深的感谢，然后贴上 API 文档~：https://allmonday.github.io/pydantic-resolve/reference_api/

这个库的概念并不复杂，但鉴于 python web 相对小众，也许能发挥的作用并不大。

因此想开发一些基于 java 或者 js 的版本，故发帖来收集一下大家的意见和反馈。

请多多指教。

restful 本身也能做到返回多层的嵌套数据，这里只是为了方便比较，故特此说明。

pydantic-resolve 和 gql 的概念区别是，它从数据来做展开，gql 则都是从查询来展开。

彩蛋：
附上一个计算 tree count 总和的 snippet.

class Tree(BaseModel):
    count: int
    children: List[Tree] = []

    total: int = 0
    def post_total(self):
        return self.count + sum([c.total for c in self.children])


tree = dict(count=10, children=[dict(count=9, children=[]),
    dict(count=1, children=[dict(count=20, children=[])
    ])
])

async def main():
    t = await Resolver().resolve(Tree(**tree))
    print(t.json(indent=2))


asyncio.run(main())

{
  "count": 10,
  "children": [
    {
      "count": 9,
      "children": [],
      "total": 9
    },
    {
      "count": 1,
      "children": [
        {
          "count": 20,
          "children": [],
          "total": 20
        }
      ],
      "total": 21
    }
  ],
  "total": 40
}