from __future__ import annotations from pathlib import Path from backend.app.api.errors import AppError from backend.app.api.schemas import DeleteResponse, MkdirResponse, RenameResponse from backend.app.fs.filesystem_adapter import FilesystemAdapter from backend.app.security.path_guard import PathGuard class FileOpsService: def __init__(self, path_guard: PathGuard, filesystem: FilesystemAdapter): self._path_guard = path_guard self._filesystem = filesystem def mkdir(self, parent_path: str, name: str) -> MkdirResponse: resolved_parent = self._path_guard.resolve_directory_path(parent_path) safe_name = self._path_guard.validate_name(name, field="name") target_relative = self._join_relative(resolved_parent.relative, safe_name) resolved_target = self._path_guard.resolve_path(target_relative) if resolved_target.absolute.exists(): raise AppError( code="already_exists", message="Target path already exists", status_code=409, details={"path": resolved_target.relative}, ) try: self._filesystem.make_directory(resolved_target.absolute) except FileExistsError: raise AppError( code="already_exists", message="Target path already exists", status_code=409, details={"path": resolved_target.relative}, ) except OSError as exc: raise AppError( code="io_error", message="Filesystem operation failed", status_code=500, details={"reason": str(exc)}, ) return MkdirResponse(path=resolved_target.relative) def rename(self, path: str, new_name: str) -> RenameResponse: resolved_source = self._path_guard.resolve_existing_path(path) safe_name = self._path_guard.validate_name(new_name, field="new_name") parent_relative = self._path_guard.entry_relative_path(resolved_source.alias, resolved_source.absolute.parent) target_relative = self._join_relative(parent_relative, safe_name) resolved_target = self._path_guard.resolve_path(target_relative) if resolved_target.absolute.exists(): raise AppError( code="already_exists", message="Target path already exists", status_code=409, details={"path": resolved_target.relative}, ) try: self._filesystem.rename_path(resolved_source.absolute, resolved_target.absolute) except FileNotFoundError: raise AppError( code="path_not_found", message="Requested path was not found", status_code=404, details={"path": path}, ) except FileExistsError: raise AppError( code="already_exists", message="Target path already exists", status_code=409, details={"path": resolved_target.relative}, ) except OSError as exc: raise AppError( code="io_error", message="Filesystem operation failed", status_code=500, details={"reason": str(exc)}, ) return RenameResponse(path=resolved_target.relative) def delete(self, path: str) -> DeleteResponse: resolved_target = self._path_guard.resolve_existing_path(path) try: if resolved_target.absolute.is_file(): self._filesystem.delete_file(resolved_target.absolute) elif resolved_target.absolute.is_dir(): if not self._filesystem.is_directory_empty(resolved_target.absolute): raise AppError( code="directory_not_empty", message="Directory is not empty", status_code=409, details={"path": resolved_target.relative}, ) self._filesystem.delete_empty_directory(resolved_target.absolute) else: raise AppError( code="type_conflict", message="Unsupported path type for delete", status_code=409, details={"path": resolved_target.relative}, ) except AppError: raise except FileNotFoundError: raise AppError( code="path_not_found", message="Requested path was not found", status_code=404, details={"path": path}, ) except OSError as exc: raise AppError( code="io_error", message="Filesystem operation failed", status_code=500, details={"reason": str(exc)}, ) return DeleteResponse(path=resolved_target.relative) @staticmethod def _join_relative(base: str, name: str) -> str: return f"{base}/{name}" if base else name