【问题标题】:Partial update in FastAPIFastAPI 中的部分更新
【发布时间】:2023-04-09 21:59:01
【问题描述】:

我想在支持部分更新的 FastAPI 中实现 put 或 patch 请求。 The official documentation 真的很混乱,我不知道如何提出请求。 (我不知道items 在文档中,因为我的数据将通过请求的正文而不是硬编码的字典传递)。

class QuestionSchema(BaseModel):
    title: str = Field(..., min_length=3, max_length=50)
    answer_true: str = Field(..., min_length=3, max_length=50)
    answer_false: List[str] = Field(..., min_length=3, max_length=50)
    category_id: int


class QuestionDB(QuestionSchema):
    id: int


async def put(id: int, payload: QuestionSchema):
    query = (
        questions
        .update()
        .where(id == questions.c.id)
        .values(**payload)
        .returning(questions.c.id)
    )
    return await database.execute(query=query)

@router.put("/{id}/", response_model=QuestionDB)
async def update_question(payload: QuestionSchema, id: int = Path(..., gt=0),):
    question = await crud.get(id)
    if not question:
        raise HTTPException(status_code=404, detail="question not found")

    ## what should be the stored_item_data, as documentation?
    stored_item_model = QuestionSchema(**stored_item_data)
    update_data = payload.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)

    response_object = {
        "id": question_id,
        "title": payload.title,
        "answer_true": payload.answer_true,
        "answer_false": payload.answer_false,
        "category_id": payload.category_id,
    }
    return response_object

如何完成我的代码以在此处获得成功的部分更新?

【问题讨论】:

  • stored_item_data 是您进入问题变量的数据。基本上,如果您的问题是带有旧值的 dict,请将旧值替换为新值(有效负载变量),并将整个行替换为数据库中的组合值(旧值和新值)。文档显示了一个一般情况,您应该自己在数据库上实现更新,而不是 fastapi

标签: python python-3.x sqlalchemy fastapi pydantic


【解决方案1】:

我在 FastAPI 的 Github 问题上得到了这个答案:

您可以在基类上将字段设为可选,并创建一个扩展 QuestionSchema 的新 QuestionCreate 模型。举个例子:

from typing import Optional

class Question(BaseModel):
    title: Optional[str] = None  # title is optional on the base schema
    ...

class QuestionCreate(Question):
   title: str  # Now title is required

cookiecutter 模板here 也提供了一些很好的洞察力。

【讨论】:

    【解决方案2】:

    为正在寻找直观解决方案以创建其 pydantic 模型的可选版本而无需重复代码的 google 用户发布此内容。

    假设我们有一个User 模型,我们希望允许 PATCH 请求更新用户。但是我们需要创建一个模式来告诉 FastApi 在内容主体中期望什么,特别是所有字段都是可选的(因为这是 PATCH 请求的本质)。我们可以在不重新定义所有字段的情况下这样做

    from pydantic import BaseModel
    from typing import Optional
    
    # Creating our Base User Model
    class UserBase(BaseModel):
       username: str
       email: str
       
    
    # And a Model that will be used to create an User
    class UserCreate(UserBase):
       password: str
    
    

    代码重复❌

    class UserOptional(UserCreate):
        username: Optional[str]
        email: Optional[str]
        password: Optional[str]
    

    一个班轮✅

    # Now we can make a UserOptional class that will tell FastApi that all the fields are optional. 
    # Doing it this way cuts down on the duplication of fields
    class UserOptional(UserCreate):
        __annotations__ = {k: Optional[v] for k, v in UserCreate.__annotations__.items()}
    

    注意:即使 Model 上的某个字段已经是 Optional,由于 Optional 的性质在后台是 typing.Union[type passed to Optional, None],它也不会产生影响。

    typing.Union[str, None] == typing.Optional[str]


    如果你要多次使用它,你甚至可以把它变成一个函数:

    def convert_to_optional(schema):
        return {k: Optional[v] for k, v in schema.__annotations__.items()}
    
    class UserOptional(UserCreate):
        __annotations__ = convert_to_optional(UserCreate)
    
    

    【讨论】:

      【解决方案3】:

      根据@cdraper的回答,我做了一个部分模型工厂:

      from typing import Mapping, Any, List, Type
      from pydantic import BaseModel
      
      def model_annotations_with_parents(model: BaseModel) -> Mapping[str, Any]:
          parent_models: List[Type] = [
              parent_model for parent_model in model.__bases__
              if (
                  issubclass(parent_model, BaseModel)
                  and hasattr(parent_model, '__annotations__')
              )
          ]
      
          annotations: Mapping[str, Any] = {}
      
          for parent_model in reversed(parent_models):
              annotations.update(model_annotations_with_parents(parent_model))
      
          annotations.update(model.__annotations__)
          return annotations
      
      
      def partial_model_factory(model: BaseModel, prefix: str = "Partial", name: str = None) -> BaseModel:
          if not name:
              name = f"{prefix}{model.__name__}"
      
          return type(
              name, (model,),
              dict(
                  __module__=model.__module__,
                  __annotations__={
                      k: Optional[v]
                      for k, v in model_annotations_with_parents(model).items()
                  }
              )
          )
      
      
      def partial_model(cls: BaseModel) -> BaseModel:
          return partial_model_factory(cls, name=cls.__name__)
      
      

      可以与函数partial_model_factory一起使用:

      PartialQuestionSchema = partial_model_factory(QuestionSchema)
      

      或者用装饰器partial_model:

      @partial_model
      class PartialQuestionSchema(QuestionSchema):
          pass
      

      【讨论】:

        猜你喜欢
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        相关资源
        最近更新 更多