(
purpose: UploadImagePurpose = Form(...),
image: UploadFile = FastAPIFile(...),
auth_info: dict = Depends(auth_info_required),
)
| 19 | response_model=UploadImageResponse, |
| 20 | ) |
| 21 | async def upload_image( |
| 22 | purpose: UploadImagePurpose = Form(...), |
| 23 | image: UploadFile = FastAPIFile(...), |
| 24 | auth_info: dict = Depends(auth_info_required), |
| 25 | ): |
| 26 | purpose_info = image_purpose_dict.get(purpose) |
| 27 | ext = check_ext(purpose_info, image.filename.split(".")[-1].lower()) |
| 28 | |
| 29 | image_size_limit_mb = 5 |
| 30 | check_file_size(image_size_limit_mb, image.size) |
| 31 | |
| 32 | random_id = generate_random_id(8) |
| 33 | file_id = f"{ext}_{purpose_info.id_prefix}{random_id}" |
| 34 | |
| 35 | content_bytes = await image.read() |
| 36 | image_url = await boto3_client.upload_file_from_bytes( |
| 37 | bucket_name=CONFIG.S3_BUCKET_NAME, |
| 38 | purpose=purpose.value, |
| 39 | file_id=file_id, |
| 40 | content_bytes=content_bytes, |
| 41 | original_file_name=image.filename, |
| 42 | tenant_id=CONFIG.PROJECT_ID, |
| 43 | return_url=True, |
| 44 | ) |
| 45 | |
| 46 | if not image_url: |
| 47 | raise_http_error(ErrorCode.INTERNAL_SERVER_ERROR, "Failed to upload image.") |
| 48 | |
| 49 | return UploadImageResponse(data={"url": image_url}) |
nothing calls this directly
no test coverage detected