Pydantic-resolve 使用手册 / Manual

Pydantic-resolve 是一个轻量级的工具库， 用来构建多层嵌套结构的数据。

它能通过上层的数据字段获取并关联下层的数据。 也能在获取数据之后执行post方法进行额外计算。

安装

pip install pydantic-resolve

source: https://github.com/allmonday/pydantic-resolve

我们用FastAPI应用场景来举例， Tasks 接口

这个场景的目的是生成一个 task -> comment -> feedback 的三层嵌套数据

假设我们现在有一个API接口， get_tasks

class Task(Base):
   __tablename__ = "task"

   id: Mapped[int] = mapped_column(primary_key=True)
   name: Mapped[str]

class TaskSchema(BaseModel):
   id: int
   name: str

   class Config:
      orm_mode = True

@app.get('/tasks', response_model=List[TaskSchema])
async def get_tasks(private:bool= Query(default=True),
                  session: AsyncSession = Depends(db.get_session)):
   tasks = (await session.execute(select(md.Task))).scalars().all()
   return tasks

输出：

[
   { "id": 1, "name": "setup test environment" },
   { "id": 2, "name": "initial project" },
]

通过task_id查找 task 关联的 comments

现在我们需要为每个task 对象关联一些comments, 操作步骤

1. 添加 dataloader, 传入需要task ids, 返回每个task id 关联的comments

2. 在Task类型上添加 comments: List[Comment] = []

3. 使用Resolver执行解析

pydantic-resolve 会自动处理orm对象到schema对象的转换。只要字段内容一致，并且符合pydantic的用法。

async def comment_batch_load_fn(task_ids):
   async with db.async_session() as session:
         res = await session.execute(select(Comment).where(Comment.task_id.in_(task_ids)))
         rows = res.scalars().all()
         return build_list(rows, task_ids, lambda x: x.task_id)

class Comment(Base):
   __tablename__ = "comment"

   id: Mapped[int] = mapped_column(primary_key=True)
   task_id: Mapped[int] = mapped_column()
   content: Mapped[str]

class Task(Base):
   __tablename__ = "task"

   id: Mapped[int] = mapped_column(primary_key=True)
   name: Mapped[str]

class CommentSchema(BaseModel):
   id: int
   task_id: int
   content: str

   class Config:
      orm_mode = True

class TaskSchema(BaseModel):
   id: int
   name: str

   comments: List[CommentSchema] = []
   def resolve_comments(self, comment_loader=LoaderDepend(comment_batch_load_fn)):
      return comment_loader.load(self.id)

   class Config:
      orm_mode = True

@app.get('/tasks', response_model=List[TaskSchema])
async def get_tasks(private:bool= Query(default=True),
                  session: AsyncSession = Depends(db.get_session)):
   tasks = (await session.execute(select(Task))).scalars().all()
   tasks = await Resolver().resolve(tasks)
   return tasks

输出：

[
   { "id": 1, "name": "setup test environment", "comments": [
      { "id": 1, "task_id": 1, "content": "remember to config pipeline" },
      { "id": 2, "task_id": 1, "content": "DBA is OOO" },
   ] },
   { "id": 2, "name": "initial project", "comments": [
      { "id": 3, "task_id": 2, "content": "I need authority" },
   ] },
]

通过 comment_id 为 comment 添加关联的 feedback

我们照着这个模式，继续为每个comment对象关联一些 feedbacks.

1. 添加 dataloader, 传入需要comment ids, 返回每个comment id 关联的 feedbacks

2. 在 Comment类型上添加 feedbacks: List[Feedback] = []

async def comment_batch_load_fn(task_ids):
   async with db.async_session() as session:
         res = await session.execute(select(Comment).where(Comment.task_id.in_(task_ids)))
         rows = res.scalars().all()
         return build_list(rows, task_ids, lambda x: x.task_id)

async def feedback_batch_load_fn(comment_ids):
    async with db.async_session() as session:
         res = await session.execute(select(Feedback)
            .where(Feedback.comment_id.in_(comment_ids)))
         rows = res.scalars().all()
         return build_list(rows, comment_ids, lambda x: x.comment_id)

class Task(Base):
   __tablename__ = "task"

   id: Mapped[int] = mapped_column(primary_key=True)
   name: Mapped[str]

class Comment(Base):
   __tablename__ = "comment"

   id: Mapped[int] = mapped_column(primary_key=True)
   task_id: Mapped[int] = mapped_column()
   content: Mapped[str]

class Feedback(Base):
   __tablename__ = "feedback"

   id: Mapped[int] = mapped_column(primary_key=True)
   comment_id: Mapped[int] = mapped_column()
   content: Mapped[str]

class FeedbackSchema(BaseModel):
   id: int
   comment_id: int
   content: str

   class Config:
      orm_mode = True

class CommentSchema(BaseModel):
   id: int
   task_id: int
   content: str
   feedbacks: List[FeedbackSchema] = []
   def resolve_feedbacks(self, feedback_loader=LoaderDepend(feedback_batch_load_fn)):
      return feedback_loader.load(self.id)

   class Config:
      orm_mode = True

class TaskSchema(BaseModel):
   id: int
   name: str

   comments: List[CommentSchema] = []
   def resolve_comments(self, comment_loader=LoaderDepend(comment_batch_load_fn)):
      return comment_loader.load(self.id)

   class Config:
      orm_mode = True

@app.get('/tasks', response_model=List[TaskSchema])
async def get_tasks(private:bool= Query(default=True),
                  session: AsyncSession = Depends(db.get_session)):
   tasks = (await session.execute(select(Task))).scalars().all()
   tasks = await Resolver().resolve(tasks)
   return tasks

输出：

[
   { "id": 1, "name": "setup test environment", "comments": [
      { "id": 1, "task_id": 1, "content": "remember to config pipeline", "feedbacks": [
         { "id": 1, "comment_id": 1, "content": "roger"},
         { "id": 2, "comment_id": 1, "content": "done"},
      ] },
      { "id": 2, "task_id": 1, "content": "DBA is OOO", "feedbacks": [] },
   ] },
   { "id": 2, "name": "initial project", "comments": [
      { "id": 3, "task_id": 2, "content": "I need authority", "feedbacks": [
         { "id": 3, "comment_id": 3, "content": "received"},
         { "id": 4, "comment_id": 3, "content": "granted"},
      ] },
   ] },
]

Attention

从代码上我们能看到，所有的额外关联，都没有对代码的侵入和改动。

其他使用方法：

• 1. 组合查询数据, 处理数据

• 2. 通过Context传递参数

• 3. 通过 loader_filters 改变loader 查询条件

更多：

• Search Page