直传确认,客户端完成S3直传后调用获取分享码
(upload_id: str, ip: str = Depends(ip_limit["upload"]))
| 752 | "/upload/confirm/{upload_id}", dependencies=[Depends(share_required_login)] |
| 753 | ) |
| 754 | async def presign_upload_confirm(upload_id: str, ip: str = Depends(ip_limit["upload"])): |
| 755 | """直传确认,客户端完成S3直传后调用获取分享码""" |
| 756 | session = await _get_valid_session(upload_id, expected_mode="direct") |
| 757 | |
| 758 | storage: FileStorageInterface = storages[settings.file_storage]() |
| 759 | if not await storage.file_exists(session.save_path): |
| 760 | raise HTTPException(404, "文件未上传或上传失败") |
| 761 | |
| 762 | code = await FileUploadService.create_file_record( |
| 763 | session.file_name, |
| 764 | session.file_size, |
| 765 | os.path.dirname(session.save_path), |
| 766 | session.expire_value, |
| 767 | session.expire_style, |
| 768 | ) |
| 769 | |
| 770 | await session.delete() |
| 771 | ip_limit["upload"].add_ip(ip) |
| 772 | return APIResponse(detail={"code": code, "name": session.file_name}) |
| 773 | |
| 774 | |
| 775 | @presign_api.get( |
nothing calls this directly
no test coverage detected