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.

Pydantic User models

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

Handle user password field

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.

Successful response

And Voila!! we have an API endpoint that works. Good luck