(self, file: UploadFile, save_path: str)
| 119 | chunk = file.read(self.chunk_size) |
| 120 | |
| 121 | async def save_file(self, file: UploadFile, save_path: str): |
| 122 | path_obj = Path(save_path) |
| 123 | directory = str(path_obj.parent) |
| 124 | # 提取原始文件名并进行清理 |
| 125 | filename = await sanitize_filename(path_obj.name) |
| 126 | # 构建安全的完整保存路径 |
| 127 | safe_save_path = self.root_path / directory / filename |
| 128 | # 确保目录存在 |
| 129 | if not safe_save_path.parent.exists(): |
| 130 | safe_save_path.parent.mkdir(parents=True) |
| 131 | await asyncio.to_thread(self._save, file.file, safe_save_path) |
| 132 | |
| 133 | async def delete_file(self, file_code: FileCodes): |
| 134 | save_path = self.root_path / await file_code.get_file_path() |
nothing calls this directly
no test coverage detected