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