feat: raise FileOrDirectoryNotFoundError if mtime or size queries fail and object does not exist remotely

.github/workflows/test.yml

@@ -21,10 +21,9 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Install Pixi
-        uses: prefix-dev/setup-pixi@v0.8.3
+        uses: prefix-dev/setup-pixi@v0
         with:
           environments: dev
-          pixi-version: v0.42.1
 
       - name: Ruff Format
         if: always()
@@ -49,10 +48,9 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: Install Pixi
-        uses: prefix-dev/setup-pixi@v0.8.3
+        uses: prefix-dev/setup-pixi@v0
         with:
           environments: dev
-          pixi-version: v0.42.1
 
       - name: Run tests
         run: pixi run --environment dev test --show-capture=all -s -vv

pyproject.toml

@@ -22,11 +22,10 @@ homepage = "https://github.com/snakemake/snakemake-interface-storage-plugins"
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
 
-[tool.pixi.project]
+[tool.pixi.workspace]
 channels = ["conda-forge"]
 platforms = ["osx-arm64", "linux-64"]
 
-
 [tool.pixi.tasks]
 
 [tool.pixi.dependencies]

snakemake_interface_storage_plugins/exceptions.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+from typing import Optional
+
+
+class FileOrDirectoryNotFoundError(Exception):
+    def __init__(self, local_path: Path, query: Optional[str]):
+        self.query: Optional[str] = query
+        self.local_path: Path = local_path
+        msg = (
+            f"Storage object {query} not found in storage (local path: {local_path!s})."
+            if query
+            else f"File or directory not found: {local_path!s}"
+        )
+        super().__init__(msg)
+
+    def is_for_path(self, path: Path) -> bool:
+        return self.local_path.resolve() == path.resolve()

snakemake_interface_storage_plugins/storage_object.py

@@ -19,6 +19,7 @@
 from snakemake_interface_common.logging import get_logger
 from snakemake_interface_storage_plugins.common import Operation, get_disk_free
 
+from snakemake_interface_storage_plugins.exceptions import FileOrDirectoryNotFoundError
 from snakemake_interface_storage_plugins.io import IOCacheStorageInterface
 from snakemake_interface_storage_plugins.storage_provider import StorageProviderBase
 
@@ -173,18 +174,24 @@ def retrieve_object(self):
         """
         ...
 
+    def _raise_object_not_found_if_not_exists(self):
+        if not self.exists():
+            raise FileOrDirectoryNotFoundError(self.print_query, self.local_path())
+
     async def managed_size(self) -> int:
         try:
             async with self._rate_limiter(Operation.SIZE):
                 return self.size()
         except Exception as e:
+            self._raise_object_not_found_if_not_exists()
             raise WorkflowError(f"Failed to get size of {self.print_query}", e)
 
     async def managed_mtime(self) -> float:
         try:
             async with self._rate_limiter(Operation.MTIME):
                 return self.mtime()
         except Exception as e:
+            self._raise_object_not_found_if_not_exists()
             raise WorkflowError(f"Failed to get mtime of {self.print_query}", e)
 
     async def managed_exists(self) -> bool: