In this tip, you will see an example to check how to update data partially 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):
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
class AppBaseModelOrm:
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
is_active = Column(Boolean, default=True)
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)
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