Remove a directory tree.
(self, request: DirectoryDeleteRequest)
| 375 | ) |
| 376 | |
| 377 | async def rmtree(self, request: DirectoryDeleteRequest) -> ActionResult: |
| 378 | """Remove a directory tree.""" |
| 379 | try: |
| 380 | absolute = self._policy.resolve_relative(request.path) |
| 381 | relative = self._policy.to_relative(absolute) |
| 382 | |
| 383 | async with self._locks.acquire(self._key(relative)): |
| 384 | if request.recursive: |
| 385 | await self._storage.rmtree(absolute) |
| 386 | else: |
| 387 | # For non-recursive, we need to check if directory is empty |
| 388 | entries = await self._storage.listdir(absolute) |
| 389 | if entries: |
| 390 | raise ConflictError(f"Directory not empty: {relative}") |
| 391 | await self._storage.remove(absolute) |
| 392 | |
| 393 | return ActionResult( |
| 394 | success=True, |
| 395 | message=f"Successfully deleted directory {relative}", |
| 396 | extra={"path": str(relative), "recursive": request.recursive} |
| 397 | ) |
| 398 | except Exception as e: |
| 399 | return ActionResult( |
| 400 | success=False, |
| 401 | message=f"Failed to delete directory: {str(e)}", |
| 402 | extra={"error": str(e), "path": str(request.path)} |
| 403 | ) |
| 404 | |
| 405 | async def stat(self, request: FileStatRequest) -> ActionResult: |
| 406 | """Get file statistics.""" |
nothing calls this directly
no test coverage detected