How to use FastAPI UploadFile with Pydantic model
This is a short tutorial to guide you on how to use Pydantic models and UploadFile together in FastAPI.
I had to figure out a way to achieve this after all sorts of validation errors I had. And since there were no guides online on how best to do this, I decided to write one myself, and I hope it helps someone.
I was trying to build an API endpoint that can handle profile picture uploads, and user information updates at the same time. So I will just go straight to the point. I broke down my models in the classes below so my users can have an idea of what to expect for each route.
To use the model with UploadFile I am using the UserUpdate model so I can update it when no file has been uploaded.
Here is the FastAPI route handling the update with None as default values of user fields.
@router.put(“/{user_name}”, response_model=User)
async def update_user(request: Request, user_name: constr(to_lower=True), email: str = Form(None), username: constr(to_lower=True) = Form(None),
password: SecretStr = Form(None), first_name: str = Form(None),
last_name: str = Form(None), is_active: str = Form(None),
is_superuser: bool = Form(None), is_deleted: bool = Form(None),
terms_of_service_accepted: bool = Form(None),
profile_picture: UploadFile = File(None)):# assemble user data
user = UserUpdate(email=email,
first_name=first_name,
last_name=last_name,
username=username,
password=password,
is_active=is_active,
is_superuser=is_superuser,
is_deleted=is_deleted,
terms_of_service_accepted=terms_of_service_accepted)
This would give you this view in your openapi interface
You can do whatever you like with the user variable up there. Here is what I did with mine. I excluded all unset fields and all fields with a None value. So when none of the fields is updated, I return an exception
update_data = user.dict(exclude_unset=True, exclude_none=True)
update_data = jsonable_encoder(update_data)try:
if “password” in update_data:
update_data[“password”] = get_password_hash(
user.password.get_secret_value())
except:
pass
if profile_picture is not None:
marketplace_domain = await get_request_source(request)
upload_path = marketplace_domain + “/users/” + instance.username
if not await upload_file(profile_picture, upload_path, settings.ALLOWED_IMAGE_TYPES, settings.FILE_SERVICE):
raise HTTPException(detail=”file could not be uploaded”,
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)update_data[“profile_picture”] = upload_path + “/” + profile_picture.filename
if not update_data:
raise HTTPException(detail=”empty user data in request”,
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
If everything works fine, we can now save the details to whatever database we intend to work with and return a response with the empty exception
or a successful response which would be up to you how you intend to respond.
And Voila!! we have an API endpoint that works. Good luck
