feat(hashing): add PostgresHashCacher with psycopg3 backend (ITL-520)

docs/concepts/file-hash-caching.md

@@ -174,3 +174,68 @@ sqlite3 ~/.orcapod/file_hash_cache.db "DELETE FROM file_hash_cache;"
   different key, so every access is a cache miss.
 - It is the first run on a new machine or a freshly cleared cache. Every file
   is a miss on a cold cache; you pay both the lookup cost and the file read.
+
+## Choosing a backend: SQLite vs Postgres
+
+Orcapod ships two hash cache backends. Pick based on your deployment:
+
+| | SQLite | Postgres |
+|---|---|---|
+| Setup | Zero — file on disk | Requires a running Postgres server |
+| Scope | Per-machine | Shared across machines and pipeline runs |
+| Concurrency | Single-writer | Multi-writer safe (`ON CONFLICT DO NOTHING`) |
+| Best for | Local development, single-machine pipelines | Distributed pipelines, shared caches |
+
+### Using the SQLite backend (default)
+
+```python
+import orcapod as op
+
+op.enable_file_hash_caching()  # uses ~/.orcapod/file_hash_cache.db
+# or specify a path:
+op.enable_file_hash_caching(db_path="/shared/nfs/cache.db")
+```
+
+### Using the Postgres backend
+
+Requires `psycopg` (install with `pip install 'orcapod[postgresql]'`):
+
+```python
+import orcapod as op
+
+op.enable_file_hash_caching(
+    conninfo="postgresql://user:pass@db-host:5432/orcapod_cache"
+)
+```
+
+You can also construct `PostgresHashCacher` directly and pass it to
+`CachedFileHasher` for advanced configurations:
+
+```python
+from orcapod.hashing import CachedFileHasher, FileHasher, PostgresHashCacher
+
+cacher = PostgresHashCacher(
+    conninfo="postgresql://user:pass@db-host:5432/orcapod_cache",
+    read_only=True,           # lookup only, no writes
+    min_cache_size_bytes=1_048_576,  # skip files < 1 MB
+)
+hasher = CachedFileHasher(file_hasher=FileHasher(), cacher=cacher)
+```
+
+### Schema
+
+`PostgresHashCacher` creates the following table on first use
+(`CREATE TABLE IF NOT EXISTS`):
+
+```sql
+CREATE TABLE IF NOT EXISTS file_hash_cache (
+    path      TEXT   NOT NULL,
+    mtime_ns  BIGINT NOT NULL,
+    size      BIGINT NOT NULL,
+    hash      BYTEA  NOT NULL,
+    cached_at BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT,
+    PRIMARY KEY (path, mtime_ns, size)
+)
+```
+
+Minimum supported Postgres version: **14**.

pyproject.toml

@@ -88,6 +88,7 @@ dev = [
     "redis>=6.2.0",
     "ruff>=0.14.4",
     "testcontainers[minio]>=4.0.0",
+    "testcontainers[postgres]>=4.0.0",
     "pip-licenses>=5.0.0",
     "tqdm>=4.67.1",
     "mkdocs-material>=9.7.5",

src/orcapod/contexts/__init__.py

@@ -229,15 +229,22 @@ def create_registry(
 
 
 def enable_file_hash_caching(
+    *,
     db_path: "Path | None" = None,
+    conninfo: str | None = None,
     read_only: bool = False,
     min_cache_size_bytes: int | None = None,
 ) -> None:
-    """Enable SQLite-backed file hash caching on the default Orcapod context.
+    """Enable file hash caching on the default Orcapod context.
 
-    Wraps the existing ``FileHandler``'s hasher in a ``CachedFileHasher``
-    backed by a ``SqliteHashCacher`` and re-registers it for ``orcapod.File``
-    in the default context's semantic hasher registry.
+    Exactly one backend must be chosen:
+
+    * ``conninfo`` provided → ``PostgresHashCacher``, shared across machines.
+    * ``db_path`` provided (or neither) → ``SqliteHashCacher``, local to this
+      machine.
+
+    ``conninfo`` and ``db_path`` are mutually exclusive; providing both raises
+    ``ValueError``.
 
     Call once at application startup before any file hashing occurs.
 
@@ -258,19 +265,29 @@ def enable_file_hash_caching(
     file is encountered during directory traversal.
 
     Args:
-        db_path: Path to the SQLite cache database. Defaults to
+        db_path: Path to the SQLite cache database. Mutually exclusive with
+            ``conninfo``; providing both raises ``ValueError``. Defaults to
             ``~/.orcapod/file_hash_cache.db`` or the
             ``ORCAPOD_HASH_CACHE_DB`` environment variable.
-        read_only: When ``True``, the underlying ``SqliteHashCacher`` will
-            not insert new entries. Lookups still work normally. Defaults
-            to ``False``.
+        conninfo: psycopg3 connection string for a PostgreSQL cache database,
+            e.g. ``"postgresql://user:pass@host:5432/db"``. Mutually exclusive
+            with ``db_path``; providing both raises ``ValueError``.
+        read_only: When ``True``, the underlying cacher will not insert new
+            entries. Lookups still work normally. Defaults to ``False``.
         min_cache_size_bytes: When set, files smaller than this byte count
             are not inserted into the cache. ``None`` and ``0`` disable the
             threshold. Defaults to ``None``.
+
+    Raises:
+        ValueError: If both ``conninfo`` and ``db_path`` are provided.
     """
+    if conninfo is not None and db_path is not None:
+        raise ValueError(
+            "enable_file_hash_caching(): provide conninfo or db_path, not both."
+        )
+
     from orcapod.extension_types.file_type import File
     from orcapod.hashing.file_hashers import CachedFileHasher
-    from orcapod.hashing.hash_cachers import SqliteHashCacher
     from orcapod.hashing.semantic_hashing.builtin_handlers import FileHandler
 
     context = get_default_context()
@@ -303,13 +320,28 @@ def enable_file_hash_caching(
     from orcapod.hashing.directory_hashers import BasicDirectoryHasher
     from orcapod.hashing.semantic_hashing.builtin_handlers import DirectoryHandler
 
-    cached_file_hasher = CachedFileHasher(
-        file_hasher=base_hasher,
-        cacher=SqliteHashCacher(
+    if conninfo is not None:
+        from orcapod.hashing.postgres_hash_cacher import PostgresHashCacher
+
+        cacher = PostgresHashCacher(
+            conninfo,
+            read_only=read_only,
+            min_cache_size_bytes=min_cache_size_bytes,
+        )
+    else:
+        # SqliteHashCacher import was previously at the top of the function body;
+        # it now lives inside this branch only.
+        from orcapod.hashing.hash_cachers import SqliteHashCacher
+
+        cacher = SqliteHashCacher(
             db_path,
             read_only=read_only,
             min_cache_size_bytes=min_cache_size_bytes,
-        ),
+        )
+
+    cached_file_hasher = CachedFileHasher(
+        file_hasher=base_hasher,
+        cacher=cacher,
     )
 
     registry.register(File, FileHandler(cached_file_hasher))

src/orcapod/hashing/__init__.py

@@ -51,6 +51,12 @@
     populate_hash_cache,
 )
 from orcapod.hashing.hash_cachers import InMemoryHashCacher, SqliteHashCacher
+
+try:
+    from orcapod.hashing.postgres_hash_cacher import PostgresHashCacher
+except ImportError:  # pragma: no cover
+    PostgresHashCacher = None  # type: ignore[assignment,misc]
+
 from orcapod.hashing.directory_hashers import BasicDirectoryHasher
 from orcapod.hashing.hash_utils import hash_file
 from orcapod.hashing.semantic_hashing.builtin_handlers import (
@@ -134,6 +140,7 @@
     "CachedFileHasher",
     "InMemoryHashCacher",
     "SqliteHashCacher",
+    "PostgresHashCacher",
     "CachePopulationStats",
     "FileOutcome",
     "ProgressCallback",

src/orcapod/hashing/hash_cachers.py

@@ -10,6 +10,11 @@
 import threading
 from pathlib import Path
 
+# Current SQLite schema version.  Stored in PRAGMA user_version.
+# V0 (default) = legacy schema without the cached_at column.
+# V1            = current schema: added cached_at column.
+_SQLITE_SCHEMA_VERSION = 1
+
 from orcapod.hashing.file_hashers import FileHashKey
 from orcapod.types import ContentHash
 
@@ -134,14 +139,28 @@ def __init__(
         self._ensure_schema()
 
     def _ensure_schema(self) -> None:
-        """Create the cache table and enable WAL mode.
+        """Create the cache table, enable WAL mode, and verify schema version.
+
+        Uses a dedicated one-shot connection so schema setup happens once on
+        construction, independent of the thread-local connection pool.
+
+        Schema version is tracked via ``PRAGMA user_version``:
+
+        * **V0** (default SQLite value) — legacy schema without ``cached_at``.
+        * **V1** — current schema with ``cached_at``.
 
-        Uses a dedicated one-shot connection so schema setup happens once
-        on construction, independent of the thread-local connection pool.
+        If the database already contains a ``file_hash_cache`` table that is
+        missing the ``cached_at`` column (V0), a ``ValueError`` is raised
+        with instructions to run the bundled migration script.
+
+        Raises:
+            ValueError: If an existing database uses an outdated schema.
         """
         self.db_path.parent.mkdir(parents=True, exist_ok=True)
         with sqlite3.connect(self.db_path) as conn:
             conn.execute("PRAGMA journal_mode=WAL")
+
+            # Create table with current schema (no-op if already exists).
             conn.execute(
                 """
                 CREATE TABLE IF NOT EXISTS file_hash_cache (
@@ -154,6 +173,33 @@ def _ensure_schema(self) -> None:
                 ) WITHOUT ROWID
                 """
             )
+
+            version = conn.execute("PRAGMA user_version").fetchone()[0]
+
+            # Always validate the cached_at column is present, regardless of
+            # user_version.  A manually-bumped user_version without the matching
+            # schema change would otherwise cause a cryptic INSERT failure in
+            # put() rather than a clear diagnostic here.
+            columns = {
+                row[1]
+                for row in conn.execute("PRAGMA table_info(file_hash_cache)")
+            }
+            if "cached_at" not in columns:
+                raise ValueError(
+                    f"SQLite hash cache at '{self.db_path}' uses an outdated "
+                    f"schema (version {version}, missing 'cached_at' column). "
+                    f"Run the migration script to upgrade:\n\n"
+                    f"    python -m orcapod.hashing.migrate_hash_cache "
+                    f"{self.db_path}\n"
+                )
+
+            if version < _SQLITE_SCHEMA_VERSION:
+                # Version stamp is missing or outdated — stamp it now.
+                # (Reaches here when the table was created by code before
+                # schema versioning was introduced, so cached_at exists
+                # but user_version is still 0.)
+                conn.execute(f"PRAGMA user_version = {_SQLITE_SCHEMA_VERSION}")
+
             conn.commit()
 
     def _connection(self) -> sqlite3.Connection:
@@ -204,8 +250,8 @@ def put(self, key: FileHashKey, value: ContentHash) -> None:
         conn = self._connection()
         conn.execute(
             """
-            INSERT OR REPLACE INTO file_hash_cache (path, mtime_ns, size, hash)
-            VALUES (?, ?, ?, ?)
+            INSERT OR REPLACE INTO file_hash_cache (path, mtime_ns, size, hash, cached_at)
+            VALUES (?, ?, ?, ?, strftime('%s', 'now'))
             """,
             (str(key.path), key.mtime_ns, key.size, value.to_prefixed_digest()),
         )

src/orcapod/hashing/migrate_hash_cache.py

@@ -0,0 +1,115 @@
+"""SQLite hash cache schema migration script.
+
+Upgrades a ``file_hash_cache`` SQLite database from schema V0 (no
+``cached_at`` column) to schema V1 (``cached_at`` column added).
+
+Usage::
+
+    python -m orcapod.hashing.migrate_hash_cache /path/to/cache.db
+
+Or from Python code::
+
+    from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache
+    migrate_sqlite_hash_cache("/path/to/cache.db")
+
+What the migration does:
+
+* Adds a ``cached_at INTEGER NOT NULL DEFAULT 0`` column to the
+  ``file_hash_cache`` table (existing rows receive ``cached_at = 0``).
+* Sets ``PRAGMA user_version = 1`` to stamp the new schema version.
+
+The migration is idempotent: running it on an already-migrated database
+prints a message and exits without making changes.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sqlite3
+from pathlib import Path
+
+
+def migrate_sqlite_hash_cache(db_path: "Path | str") -> None:
+    """Upgrade a SQLite hash cache database from schema V0 to V1.
+
+    Adds the ``cached_at`` column (epoch seconds; defaults to ``0`` for
+    existing rows) and stamps ``PRAGMA user_version = 1``.
+
+    Args:
+        db_path: Path to the SQLite database file to migrate.
+
+    Raises:
+        FileNotFoundError: If ``db_path`` does not exist.
+        ValueError: If the database does not contain the expected
+            ``file_hash_cache`` table.
+    """
+    db_path = Path(db_path)
+    if not db_path.exists():
+        raise FileNotFoundError(f"Database not found: {db_path}")
+
+    with sqlite3.connect(db_path) as conn:
+        # Check whether the table exists at all.
+        table_exists = conn.execute(
+            "SELECT name FROM sqlite_master "
+            "WHERE type='table' AND name='file_hash_cache'"
+        ).fetchone()
+        if not table_exists:
+            raise ValueError(
+                f"No 'file_hash_cache' table found in {db_path}. "
+                "Is this an orcapod hash cache database?"
+            )
+
+        version = conn.execute("PRAGMA user_version").fetchone()[0]
+        if version >= 1:
+            print(
+                f"{db_path}: already at schema V{version} — nothing to migrate."
+            )
+            return
+
+        columns = {
+            row[1] for row in conn.execute("PRAGMA table_info(file_hash_cache)")
+        }
+        if "cached_at" in columns:
+            # Table already has cached_at (was created with current code before
+            # version stamping was introduced).  Just stamp the version.
+            conn.execute("PRAGMA user_version = 1")
+            conn.commit()
+            print(
+                f"{db_path}: 'cached_at' column already present; "
+                "stamped schema version to V1."
+            )
+            return
+
+        print(f"{db_path}: migrating from schema V0 → V1 …")
+        conn.execute(
+            "ALTER TABLE file_hash_cache "
+            "ADD COLUMN cached_at INTEGER NOT NULL DEFAULT 0"
+        )
+        conn.execute("PRAGMA user_version = 1")
+        conn.commit()
+        print(
+            f"{db_path}: migration complete. "
+            "Existing rows have cached_at = 0 (epoch origin)."
+        )
+
+
+def main() -> None:
+    """Entry point for ``python -m orcapod.hashing.migrate_hash_cache``."""
+    parser = argparse.ArgumentParser(
+        prog="python -m orcapod.hashing.migrate_hash_cache",
+        description=(
+            "Upgrade an orcapod SQLite hash cache database "
+            "from schema V0 to V1 (adds the cached_at column)."
+        ),
+    )
+    parser.add_argument(
+        "db_path",
+        metavar="DB_PATH",
+        help="Path to the SQLite hash cache database file.",
+    )
+    args = parser.parse_args()
+    migrate_sqlite_hash_cache(args.db_path)
+
+
+if __name__ == "__main__":
+    main()