backend/app/storage/backend.py

"""StorageBackend interface + Local FS default (spec §5, §16, A9).

The DB stores a relative ``key`` (path) returned by ``save``; an S3 adapter can drop in behind
the same interface by implementing these four methods.
"""
from __future__ import annotations

from abc import ABC, abstractmethod
from pathlib import Path

from ..config import settings


class StorageBackend(ABC):
    @abstractmethod
    def save(self, key: str, data: bytes) -> str:
        """Persist ``data`` under ``key``; return the stored relative key."""

    @abstractmethod
    def open(self, key: str) -> bytes:
        ...

    @abstractmethod
    def delete(self, key: str) -> None:
        ...

    @abstractmethod
    def abs_path(self, key: str) -> str | None:
        """Absolute filesystem path if the backend is local; else None."""


class LocalStorage(StorageBackend):
    def __init__(self, root: Path | None = None):
        self.root = (root or settings.media_path)
        self.root.mkdir(parents=True, exist_ok=True)

    def _full(self, key: str) -> Path:
        # Prevent path traversal outside the media root.
        full = (self.root / key).resolve()
        if not str(full).startswith(str(self.root.resolve())):
            raise ValueError(f"Refusing to access key outside media root: {key}")
        return full

    def save(self, key: str, data: bytes) -> str:
        full = self._full(key)
        full.parent.mkdir(parents=True, exist_ok=True)
        full.write_bytes(data)
        return key

    def open(self, key: str) -> bytes:
        return self._full(key).read_bytes()

    def delete(self, key: str) -> None:
        full = self._full(key)
        if full.exists():
            full.unlink()

    def abs_path(self, key: str) -> str | None:
        return str(self._full(key))


class S3Storage(StorageBackend):  # pragma: no cover - stub for future (spec §5/A9)
    """Stub. Implement with boto3 when STORAGE_BACKEND=s3 is needed."""

    def __init__(self, *_, **__):
        raise NotImplementedError("S3 storage backend not implemented yet (see DECISIONS.md).")

    def save(self, key: str, data: bytes) -> str: ...  # noqa: D401
    def open(self, key: str) -> bytes: ...
    def delete(self, key: str) -> None: ...
    def abs_path(self, key: str) -> str | None: ...


_backend: StorageBackend | None = None


def get_storage() -> StorageBackend:
    global _backend
    if _backend is None:
        if settings.storage_backend == "s3":
            _backend = S3Storage()
        else:
            _backend = LocalStorage()
    return _backend