Processing PATCH Requests in FastAPI with SQLAlchemy

• Download source code - 33.3 KB

Background

The most commonly used HTTP methods are GET, POST, PUT and DELETE. There is a similar one like PUT known as PATCH. PATCH is used to indicate partial data updates. In this example, we will check how to update data partially in FastAPI using SQLAlchemy and Python.

Helper Class

This is a basic CRUDE helper class.

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: Gets the DB entity by id
• set_attrs: Sets new values from updated_attrs to entity DB model
• update: Updates Db row

set_attrs

We will focus more on the set_attrs the method. It will set new values to the existing data model.

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: db model
• updated_attrs: property/field wise new value dictionary
• throw_error_if_data_type_not_same: if true, the process will throw an error when the same-named field/property present in both entity and updated_attrs but datatypes are different.
• throw_error_if_attr_not_in_entity: If true, the process will throw an error when updated_attrs got any field/property that is not present in entity

I am considering some restrictions, so set default true for both of the flags, but it depends on the way we want it to be.

PATCH API

Models

DB Model

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 Request Model

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

Here, we are retrieving data row from DB by row id, applying request changes to the row, and saving the row back to the table.

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) converts the model to a dictionary with fields that were explicitly set or data that came as requested.

Using the Code

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

\backend> docker-compose up -d

project will run http://localhost:4003

Go to API Doc
http://localhost:4003/docs#/ 

Check PATCH API of Task Queue section.

Reference

• https://sqlmodel.tiangolo.com/tutorial/fastapi/update/

History

• 7^th July, 2022: Initial version