生成S3预签名上传URL :param save_path: 文件保存路径 :param expires_in: URL过期时间(秒),默认15分钟 :return: 预签名PUT URL
(self, save_path: str, expires_in: int = 900)
| 514 | logger.info(f"清理 S3 分片数据时出错: {e}") |
| 515 | |
| 516 | async def generate_presigned_upload_url(self, save_path: str, expires_in: int = 900) -> Optional[str]: |
| 517 | """ |
| 518 | 生成S3预签名上传URL |
| 519 | :param save_path: 文件保存路径 |
| 520 | :param expires_in: URL过期时间(秒),默认15分钟 |
| 521 | :return: 预签名PUT URL |
| 522 | """ |
| 523 | async with self._client() as s3: |
| 524 | return await s3.generate_presigned_url( |
| 525 | "put_object", |
| 526 | Params={ |
| 527 | "Bucket": self.bucket_name, |
| 528 | "Key": save_path, |
| 529 | }, |
| 530 | ExpiresIn=expires_in, |
| 531 | ) |
| 532 | |
| 533 | async def file_exists(self, save_path: str) -> bool: |
| 534 | """ |
no test coverage detected