Processing PATCH Requests in FastAPI with SQLAlchemy





5.00/5 (1 vote)
This example is to partially update data in FastAPI using SQLAlchemy and Python
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 idset_attrs
: Sets new values fromupdated_attrs
toentity
DB modelupdate
: 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 modelupdated_attrs
: property/field wise new value dictionarythrow_error_if_data_type_not_same
: iftrue
, the process will throw an error when the same-named field/property present in bothentity
andupdated_attrs
but datatypes are different.throw_error_if_attr_not_in_entity
: Iftrue
, the process will throw an error whenupdated_attrs
got any field/property that is not present inentity
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
History
- 7th July, 2022: Initial version