US01-05: Generate and Serve Managed Thumbnails (#50)

photo_pipeline/api/app.py

@@ -12,7 +12,7 @@ from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
 
-from photo_pipeline.api.routes import health
+from photo_pipeline.api.routes import health, thumbnails
 from photo_pipeline.config import Config
 from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
 from photo_pipeline.logging import configure_logging
@@ -38,4 +38,5 @@ def create_app(config: Config | None = None) -> FastAPI:
 
     app = FastAPI(title="Photo Pipeline", version="0.1.0", lifespan=lifespan)
     app.include_router(health.router, prefix="/api/v1")
+    app.include_router(thumbnails.router, prefix="/api/v1")
     return app

photo_pipeline/api/routes/thumbnails.py

@@ -0,0 +1,35 @@
+"""Thumbnail endpoint: previews addressed only by asset ID and bounded size.
+
+The browser never supplies a filesystem path. Errors return a consistent JSON
+envelope with the service's typed code and status; successful responses carry the
+image with an immutable cache header (the URL's identity comes from the versioned,
+content-keyed cache).
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Query, Request
+from fastapi.responses import FileResponse, JSONResponse
+
+from photo_pipeline.services.thumbnails import ThumbnailError, ThumbnailService
+
+router = APIRouter(tags=["thumbnails"])
+
+
+@router.get("/assets/{asset_id}/thumbnail")
+def get_thumbnail(asset_id: str, request: Request, size: int = Query(512)):
+    service = ThumbnailService(
+        request.app.state.session_factory, request.app.state.config
+    )
+    try:
+        path = service.generate(asset_id, size)
+    except ThumbnailError as error:
+        return JSONResponse(
+            status_code=error.http_status,
+            content={"error": {"code": error.code, "message": str(error)}},
+        )
+    return FileResponse(
+        path,
+        media_type="image/webp",
+        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+    )

photo_pipeline/config.py

@@ -30,6 +30,11 @@ class Config(BaseModel):
     log_level: str = "INFO"
     log_format: str = "json"  # "json" or "text"
 
+    # Library boundary for path validation (os.pathsep-separated in the env var).
+    library_roots: tuple[Path, ...] = ()
+    thumbnail_cache_quota_bytes: int = 500_000_000
+    thumbnail_max_pixels: int = 100_000_000
+
     vision_api_key: SecretStr | None = None
     immich_api_key: SecretStr | None = None
 
@@ -41,12 +46,17 @@ class Config(BaseModel):
     def database_url(self) -> str:
         return f"sqlite:///{self.database_path}"
 
+    @property
+    def thumbnail_cache_dir(self) -> Path:
+        return self.data_dir / "cache" / "thumbs"
+
     @classmethod
     def from_env(cls, environ: Mapping[str, str] | None = None) -> "Config":
         env = os.environ if environ is None else environ
-        data = {
-            name: env[ENV_PREFIX + name.upper()]
-            for name in cls.model_fields
-            if env.get(ENV_PREFIX + name.upper())
-        }
+        data: dict = {}
+        for name in cls.model_fields:
+            raw = env.get(ENV_PREFIX + name.upper())
+            if not raw:
+                continue
+            data[name] = raw.split(os.pathsep) if name == "library_roots" else raw
         return cls(**data)

photo_pipeline/models/__init__.py

@@ -10,6 +10,7 @@ from photo_pipeline.models.duplicates import (
     DuplicateMember,
     DuplicateNegativeLink,
 )
+from photo_pipeline.models.thumbnails import Thumbnail
 
 __all__ = [
     "Asset",
@@ -17,4 +18,5 @@ __all__ = [
     "DuplicateCluster",
     "DuplicateMember",
     "DuplicateNegativeLink",
+    "Thumbnail",
 ]

photo_pipeline/models/thumbnails.py

@@ -0,0 +1,37 @@
+"""Managed thumbnail cache records.
+
+One row per ``cache_key`` (pixel hash + size + algorithm version, or the source
+byte hash for undecodable inputs). The row records whether generation succeeded or
+failed and why, so a broken original is not retried on every request. Because the
+key contains the normalized pixel hash, an EXIF-only change reuses the thumbnail
+and a genuine pixel change invalidates it automatically.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class Thumbnail(Base):
+    __tablename__ = "thumbnails"
+
+    cache_key: Mapped[str] = mapped_column(String, primary_key=True)
+    asset_id: Mapped[str] = mapped_column(ForeignKey("assets.id"), nullable=False)
+    size: Mapped[int] = mapped_column(Integer, nullable=False)
+    state: Mapped[str] = mapped_column(String, nullable=False)  # ready | error
+    error_code: Mapped[str | None] = mapped_column(String)
+    path: Mapped[str | None] = mapped_column(String)
+    width: Mapped[int | None] = mapped_column(Integer)
+    height: Mapped[int | None] = mapped_column(Integer)
+    format: Mapped[str | None] = mapped_column(String)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )

photo_pipeline/services/thumbnails.py

@@ -0,0 +1,274 @@
+"""ThumbnailService — safe, cached, correctly oriented previews by asset ID.
+
+Thumbnails are requested by asset ID (never a browser-supplied path), resolved to
+the current path through the database, and validated against the configured library
+boundary. Generation applies EXIF orientation before resizing, preserves
+transparency, and encodes WebP. The cache key is ``pixel_hash + size + version``, so
+an EXIF-only edit reuses the file while a real pixel change invalidates it; writes
+are atomic and the cache is bounded by an LRU quota. Failures are persisted as typed
+errors so a broken original is not retried on every request.
+
+Reuses photo_analyzer.prepare_image decode/resize/HEIC handling, adding the missing
+EXIF-orientation step, WebP output, and a managed cache (donor_ledger.yaml:
+pa-imaging).
+"""
+
+from __future__ import annotations
+
+import uuid
+from pathlib import Path
+
+from PIL import Image, ImageOps, UnidentifiedImageError
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline import path_policy
+from photo_pipeline.config import Config
+from photo_pipeline.models import Asset, Thumbnail
+
+# Best-effort HEIC support: registered only if the optional decoder is installed.
+try:  # pragma: no cover - depends on an optional native dependency
+    import pillow_heif
+
+    pillow_heif.register_heif_opener()
+except Exception:  # pragma: no cover
+    pass
+
+SIZES = (256, 512, 1280)
+THUMB_VERSION = 1
+THUMB_FORMAT = "webp"
+
+
+class ThumbnailError(RuntimeError):
+    code = "thumbnail_error"
+    http_status = 500
+
+
+class InvalidSize(ThumbnailError):
+    code = "invalid_size"
+    http_status = 422
+
+
+class ThumbnailNotFound(ThumbnailError):
+    code = "not_found"
+    http_status = 404
+
+
+class ThumbnailUnavailable(ThumbnailError):
+    code = "unavailable"
+    http_status = 409
+
+
+class UnsupportedImage(ThumbnailError):
+    code = "unsupported_image"
+    http_status = 415
+
+
+class ImageTooLarge(ThumbnailError):
+    code = "image_too_large"
+    http_status = 413
+
+
+class PathNotAllowed(ThumbnailError):
+    code = "path_not_allowed"
+    http_status = 403
+
+
+# Errors that describe a broken source and are persisted to avoid retry loops.
+_PERSISTED_ERRORS = {UnsupportedImage, ImageTooLarge}
+_ERROR_BY_CODE = {cls.code: cls for cls in _PERSISTED_ERRORS}
+
+
+class ThumbnailService:
+    def __init__(self, session_factory: sessionmaker, config: Config) -> None:
+        self._session_factory = session_factory
+        self._config = config
+        self._cache_dir = config.thumbnail_cache_dir
+
+    def generate(self, asset_id: str, size: int) -> Path:
+        if size not in SIZES:
+            raise InvalidSize(f"size must be one of {SIZES}")
+
+        with self._session_factory() as session:
+            asset = session.get(Asset, asset_id)
+            if asset is None:
+                raise ThumbnailNotFound(f"unknown asset {asset_id}")
+            if asset.availability_state != "active" or not asset.current_path:
+                raise ThumbnailUnavailable(f"asset {asset_id} has no active file")
+            self._validate_path(asset.current_path)
+            cache_key = self._cache_key(asset, size)
+
+            row = session.get(Thumbnail, cache_key)
+            if row is not None:
+                if row.state == "error":
+                    raise _ERROR_BY_CODE[row.error_code](
+                        f"cached failure for {asset_id}: {row.error_code}"
+                    )
+                if row.path and Path(row.path).exists():
+                    _touch(row.path)
+                    return Path(row.path)
+
+            source = asset.current_path
+
+        # Rendering happens outside the DB session (no transaction held during I/O).
+        try:
+            rendered = self._render(source, size, cache_key)
+        except ThumbnailError as error:
+            if type(error) in _PERSISTED_ERRORS:
+                self._record_error(cache_key, asset_id, size, error.code)
+            raise
+
+        self._record_ready(cache_key, asset_id, size, rendered)
+        self._enforce_quota(keep=rendered["path"])
+        return Path(rendered["path"])
+
+    # ── path safety ──────────────────────────────────────────────────────────
+    def _validate_path(self, current_path: str) -> None:
+        path = Path(current_path)
+        if path_policy.is_excluded(path):
+            raise PathNotAllowed(f"excluded path: {current_path}")
+        roots = self._config.library_roots
+        if not roots:
+            return
+        for root in roots:
+            try:
+                path_policy.resolve_within(Path(root), path)
+                return
+            except path_policy.PathPolicyError:
+                continue
+        raise PathNotAllowed(f"path outside configured roots: {current_path}")
+
+    # ── cache key + rendering ──────────────────────────────────────────────────
+    @staticmethod
+    def _cache_key(asset: Asset, size: int) -> str:
+        if asset.pixel_sha256:
+            return f"{asset.pixel_sha256}-{size}-v{THUMB_VERSION}"
+        # Undecodable at inventory time: key by source bytes so a fixed file retries.
+        return f"src:{asset.current_sha256 or asset.id}-{size}-v{THUMB_VERSION}"
+
+    def _cache_path(self, cache_key: str) -> Path:
+        safe = cache_key.replace(":", "_")
+        return self._cache_dir / safe[:2] / f"{safe}.{THUMB_FORMAT}"
+
+    def _render(self, source: str, size: int, cache_key: str) -> dict:
+        try:
+            with Image.open(source) as image:
+                width, height = image.size
+                if width * height > self._config.thumbnail_max_pixels:
+                    raise ImageTooLarge(
+                        f"{width}x{height} exceeds {self._config.thumbnail_max_pixels} px"
+                    )
+                oriented = ImageOps.exif_transpose(image)
+                mode = "RGBA" if _has_alpha(oriented) else "RGB"
+                converted = oriented.convert(mode)
+                converted.thumbnail((size, size), Image.LANCZOS)
+                out_width, out_height = converted.size
+                destination = self._cache_path(cache_key)
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                tmp = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp")
+                converted.save(tmp, format="WEBP", quality=82, method=4)
+        except ImageTooLarge:
+            raise
+        except (UnidentifiedImageError, OSError, ValueError) as error:
+            raise UnsupportedImage(f"cannot decode {source}: {error}") from error
+
+        import os
+
+        os.replace(tmp, destination)
+        return {
+            "path": str(destination),
+            "width": out_width,
+            "height": out_height,
+            "format": THUMB_FORMAT,
+        }
+
+    # ── persistence ────────────────────────────────────────────────────────────
+    def _record_ready(self, cache_key: str, asset_id: str, size: int, rendered: dict) -> None:
+        with self._session_factory() as session:
+            session.merge(
+                Thumbnail(
+                    cache_key=cache_key,
+                    asset_id=asset_id,
+                    size=size,
+                    state="ready",
+                    error_code=None,
+                    path=rendered["path"],
+                    width=rendered["width"],
+                    height=rendered["height"],
+                    format=rendered["format"],
+                )
+            )
+            try:
+                session.commit()
+            except IntegrityError:
+                session.rollback()  # another worker recorded it first; the file stands
+
+    def _record_error(self, cache_key: str, asset_id: str, size: int, code: str) -> None:
+        with self._session_factory() as session:
+            session.merge(
+                Thumbnail(
+                    cache_key=cache_key,
+                    asset_id=asset_id,
+                    size=size,
+                    state="error",
+                    error_code=code,
+                )
+            )
+            try:
+                session.commit()
+            except IntegrityError:
+                session.rollback()
+
+    # ── quota / eviction ───────────────────────────────────────────────────────
+    def _enforce_quota(self, keep: str | None = None) -> None:
+        quota = self._config.thumbnail_cache_quota_bytes
+        if not self._cache_dir.exists():
+            return
+        files = [f for f in self._cache_dir.rglob(f"*.{THUMB_FORMAT}") if f.is_file()]
+        total = sum(f.stat().st_size for f in files)
+        if total <= quota:
+            return
+        files.sort(key=lambda f: f.stat().st_mtime)  # least-recently-used first
+        keep_path = str(Path(keep)) if keep else None
+        evicted: list[str] = []
+        for f in files:
+            if total <= quota:
+                break
+            if str(f) == keep_path:
+                continue
+            size = f.stat().st_size
+            try:
+                f.unlink()
+                total -= size
+                evicted.append(str(f))
+            except OSError:
+                pass
+        if evicted:
+            self._forget(evicted)
+
+    def _forget(self, paths: list[str]) -> None:
+        with self._session_factory() as session:
+            rows = session.execute(
+                select(Thumbnail).where(Thumbnail.path.in_(paths))
+            ).scalars()
+            for row in rows:
+                session.delete(row)
+            session.commit()
+
+
+def _has_alpha(image: Image.Image) -> bool:
+    return image.mode in ("RGBA", "LA") or (
+        image.mode == "P" and "transparency" in image.info
+    )
+
+
+def _touch(path: str) -> None:
+    import os
+    import time
+
+    try:
+        now = time.time()
+        os.utime(path, (now, now))
+    except OSError:
+        pass