Prepare v0.1.0 release

src/govoplan_files/backend/storage/backends.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Iterable, Protocol
 
@@ -26,33 +26,43 @@ class StorageBackend(Protocol):
 @dataclass(slots=True)
 class LocalFilesystemStorageBackend:
     root: Path
+    fallback_roots: tuple[Path, ...] = field(default_factory=tuple)
     name: str = "local"
 
     def __post_init__(self) -> None:
         self.root = self.root.expanduser().resolve()
+        self.fallback_roots = tuple(root.expanduser().resolve() for root in self.fallback_roots if root)
         self.root.mkdir(parents=True, exist_ok=True)
 
-    def _path(self, key: str) -> Path:
-        path = (self.root / key).resolve()
-        if not path.is_relative_to(self.root):
+    def _path_for_root(self, root: Path, key: str) -> Path:
+        path = (root / key).resolve()
+        if not path.is_relative_to(root):
             raise StorageBackendError("Storage key escapes local storage root")
         return path
 
+    def _path(self, key: str) -> Path:
+        return self._path_for_root(self.root, key)
+
+    def _readable_path(self, key: str) -> Path:
+        primary = self._path(key)
+        if primary.exists() and primary.is_file():
+            return primary
+        for root in self.fallback_roots:
+            candidate = self._path_for_root(root, key)
+            if candidate.exists() and candidate.is_file():
+                return candidate
+        raise StorageBackendError("Stored object does not exist")
+
     def put_bytes(self, key: str, data: bytes, *, content_type: str | None = None) -> None:
         path = self._path(key)
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_bytes(data)
 
     def get_bytes(self, key: str) -> bytes:
-        path = self._path(key)
-        if not path.exists() or not path.is_file():
-            raise StorageBackendError("Stored object does not exist")
-        return path.read_bytes()
+        return self._readable_path(key).read_bytes()
 
     def iter_bytes(self, key: str, *, chunk_size: int = 1024 * 1024) -> Iterable[bytes]:
-        path = self._path(key)
-        if not path.exists() or not path.is_file():
-            raise StorageBackendError("Stored object does not exist")
+        path = self._readable_path(key)
         with path.open("rb") as handle:
             while True:
                 chunk = handle.read(chunk_size)
@@ -66,8 +76,11 @@ class LocalFilesystemStorageBackend:
             path.unlink()
 
     def exists(self, key: str) -> bool:
-        path = self._path(key)
-        return path.exists() and path.is_file()
+        try:
+            self._readable_path(key)
+        except StorageBackendError:
+            return False
+        return True
 
 
 @dataclass(slots=True)
@@ -125,10 +138,15 @@ class S3StorageBackend:
             return False
 
 
+def _fallback_roots() -> tuple[Path, ...]:
+    raw = getattr(settings, "file_storage_local_fallback_roots", "") or ""
+    return tuple(Path(item.strip()) for item in str(raw).split(",") if item.strip())
+
+
 def get_storage_backend() -> StorageBackend:
     backend = settings.file_storage_backend.lower().strip()
     if backend in {"local", "filesystem", "fs"}:
-        return LocalFilesystemStorageBackend(Path(settings.file_storage_local_root))
+        return LocalFilesystemStorageBackend(Path(settings.file_storage_local_root), fallback_roots=_fallback_roots())
     if backend in {"s3", "garage"}:
         return S3StorageBackend(
             bucket=settings.file_storage_s3_bucket or settings.s3_bucket,