FastAPI 中使用 SQLAlchemy 处理 PATCH 请求

• 下载源代码 - 33.3 KB

背景

最常用的 HTTP 方法是 GET、POST、PUT 和 DELETE。 还有一个类似于 PUT 的方法，称为 PATCH。 PATCH 用于指示部分数据更新。 在本示例中，我们将检查如何使用 SQLAlchemy 和 Python 在 FastAPI 中部分更新数据。

辅助类

这是一个基本的 CRUDE 辅助类。

table_repo.py

class TableRepository:

    entity:object = NotImplementedError
    db:Session = NotImplementedError

    def __init__(self, db:Session, entity:object):
        self.db = db
        self.entity = entity

    def find_by_id(self, id:int):
        return self.db.query(self.entity).filter(self.entity.id==id).first()

    def set_attrs(self, entity, updated_attrs, 
    throw_error_if_data_type_not_same:bool = True, 
    throw_error_if_attr_not_in_entity:bool = True):

        # simple one
        # for attr in updated_attrs:
        #     has_attr = hasattr(entity, attr)
        #     if has_attr:
        #         setattr(entity, attr, updated_attrs[attr])

        # complex one
        attrs = []
        for attr in updated_attrs:
            has_attr = hasattr(entity, attr)
            if has_attr:
                expected_type = type(getattr(entity, attr))
                inputed_type = type(updated_attrs[attr])
                is_same_type =  inputed_type == expected_type
                if is_same_type:
                    attrs.append(attr)
                else:
                    if throw_error_if_data_type_not_same:
                        raise TypeError(f"The expected value type of attr 
                        '{attr}' is '{expected_type}' of entity, 
                        where inputted value type is '{inputed_type}'.")
            else:
                if throw_error_if_attr_not_in_entity:
                    raise TypeError(f"attr '{attr}' is not found in entity.")
                  
        for attr in attrs:
            setattr(entity, attr, updated_attrs[attr])   
        return attrs
  
    def update(self, entity, updated_by_user_id:int = None):
        entity.updated_by = updated_by_user_id

• find_by_id：根据 ID 获取数据库实体
• set_attrs：将 updated_attrs 中的新值设置到 entity 数据库模型
• update：更新数据库行

set_attrs

我们将更多地关注 set_attrs 方法。 它会将新值设置到现有的数据模型中。

set_attrs(self, entity, updated_attrs, throw_error_if_data_type_not_same:bool = True, \
throw_error_if_attr_not_in_entity:bool = True)

• entity：数据库模型
• updated_attrs：属性/字段明智的新值字典
• throw_error_if_data_type_not_same：如果为 true，当 entity 和 updated_attrs 中存在同名的字段/属性但数据类型不同时，该过程将抛出错误。
• throw_error_if_attr_not_in_entity：如果为 true，当 updated_attrs 包含 entity 中不存在的任何字段/属性时，该过程将抛出错误

我考虑了一些限制，因此将两个标志的默认值设置为 true，但这取决于我们希望的方式。

PATCH API

模型

数据库模型

models.py

# common fields for all entities
class AppBaseModelOrm:
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    is_active = Column(Boolean, default=True)  # soft delete
    created_by = Column(Integer)
    updated_by = Column(Integer, default=None)
    created_datetime = Column(DateTime(timezone=True), default=datetime.datetime.utcnow)
    updated_datetime = Column(DateTime(timezone=True), 
                       default=None, onupdate=datetime.datetime.utcnow)

    account_id = Column(Integer)

# tables
class TaskQueue(AppBaseModelOrm, Base):
    __tablename__ = "task_queues"
    name = Column(String, index=True) 

API 请求模型

schemas.py

class TaskQueueSchema(CamelModel):
    id: int
    account_id: int
    name: str
    is_active:bool
    created_by:Optional[int] = None
    updated_by:Optional[int] = None
    created_datetime:Optional[datetime.datetime] = None
    updated_datetime:Optional[datetime.datetime] = None

    class Config:
        orm_mode = True

class TaskQueuePatch(CamelModel):
    name: str = None 

API

在这里，我们通过行 ID 从数据库中检索数据行，将请求更改应用于该行，并将该行保存回表中。

task_queue.py

@cbv(router)
class TaskQueue:
    db: Session = Depends(get_db)
    current_user:CurrentUser = Depends(get_current_user)

    @router.patch("/{id}", response_model=schemas.TaskQueueSchema)
    def patch_item(self, id:int, model: schemas.TaskQueuePatch):
        '''can be null'''
        repo = TableRepository(self.db, models.TaskQueue)
        item = repo.find_by_id(id)
        if item:
            update_data = model.dict(exclude_unset=True)
            repo.set_attrs(item, update_data)
            repo.update(item, self.current_user.id)
            self.db.commit()
            self.db.refresh(item)
        return item 

model.dict(exclude_unset=True) 将模型转换为一个字典，其中包含显式设置的字段或作为请求接收到的数据。

Using the Code

Go to backend folder
Open cmd 
Type docker-compose up -d

\backend> docker-compose up -d

project will run https://:4003

Go to API Doc
https://:4003/docs#/ 

请查看 任务队列 部分的 PATCH API。

参考

• https://sqlmodel.cn/tutorial/fastapi/update/

历史

• 2022 年 7 月 7 日：初始版本