diff --git a/docs/concepts/file-hash-caching.md b/docs/concepts/file-hash-caching.md index 2ecfcb36..32e53439 100644 --- a/docs/concepts/file-hash-caching.md +++ b/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**. diff --git a/pyproject.toml b/pyproject.toml index ea2c34c7..98b1c86d 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/orcapod/contexts/__init__.py b/src/orcapod/contexts/__init__.py index a66ef3a6..69d58640 100644 --- a/src/orcapod/contexts/__init__.py +++ b/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)) diff --git a/src/orcapod/hashing/__init__.py b/src/orcapod/hashing/__init__.py index bd0866d7..ee0b094d 100644 --- a/src/orcapod/hashing/__init__.py +++ b/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", diff --git a/src/orcapod/hashing/hash_cachers.py b/src/orcapod/hashing/hash_cachers.py index feea722e..bd8ef5d1 100644 --- a/src/orcapod/hashing/hash_cachers.py +++ b/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()), ) diff --git a/src/orcapod/hashing/migrate_hash_cache.py b/src/orcapod/hashing/migrate_hash_cache.py new file mode 100644 index 00000000..fa1035e9 --- /dev/null +++ b/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() diff --git a/src/orcapod/hashing/postgres_hash_cacher.py b/src/orcapod/hashing/postgres_hash_cacher.py new file mode 100644 index 00000000..46fd2220 --- /dev/null +++ b/src/orcapod/hashing/postgres_hash_cacher.py @@ -0,0 +1,263 @@ +"""PostgreSQL-backed file hash cacher. + +Provides ``PostgresHashCacher`` — a network-accessible, concurrent-safe +implementation of ``CacherProtocol[FileHashKey, ContentHash]``. Requires +the optional ``psycopg`` driver (``pip install 'orcapod[postgresql]'``). +""" + +from __future__ import annotations + +import re +import threading + +from orcapod.hashing.file_hashers import FileHashKey +from orcapod.types import ContentHash + +try: + import psycopg +except ImportError as _exc: # pragma: no cover + raise ImportError( + "PostgresHashCacher requires psycopg. " + "Install it with: pip install 'orcapod[postgresql]'" + ) from _exc + +# Current schema version. Increment when the DDL changes. +_SCHEMA_VERSION = 1 + + +def _redact_conninfo(conninfo: str) -> str: + """Return ``conninfo`` with any password value replaced by ``***``. + + Handles both URL form (``postgresql://user:pass@host/db``) and + keyword DSN form (``host=... password=secret ...``). + + Args: + conninfo: Raw psycopg3 connection string. + + Returns: + Connection string safe for logging, with the password redacted. + """ + # URL form: postgresql://user:pass@host → postgresql://user:***@host + # Capture group 1: "://user:", group 2: "@" — replace the password between them. + redacted = re.sub(r"(://[^:@]*:)[^@]+(@)", r"\1***\2", conninfo) + # Keyword form: password=value or password='value with spaces' + redacted = re.sub( + r"(?i)\bpassword\s*=\s*(?:'[^']*'|\S+)", "password=***", redacted + ) + return redacted + + +class PostgresHashCacher: + """PostgreSQL-backed file hash cacher. + + Stores file hashes keyed on ``(path, mtime_ns, size)`` in a shared + PostgreSQL database. Uses thread-local connections for thread safety + and ``INSERT ... ON CONFLICT DO NOTHING`` for concurrent-insert safety. + + The hash is stored as a BYTEA in ``{method}:{raw_digest}`` format via + ``ContentHash.to_prefixed_digest()``. + + Schema versioning is tracked in a companion ``file_hash_cache_meta`` + table. If an existing database is detected with an old schema (e.g. + missing the ``cached_at`` column), a ``ValueError`` is raised explaining + the required migration DDL. + + Requires ``psycopg[binary]>=3.0`` (install with + ``pip install 'orcapod[postgresql]'``). + Minimum supported PostgreSQL version: **14**. + + Args: + conninfo: psycopg3 connection string, e.g. + ``"postgresql://user:pass@host:5432/dbname"`` or keyword DSN + ``"host=myhost dbname=mydb user=myuser password=mypass"``. + Any password in the conninfo is redacted in ``__repr__``. + read_only: When ``True``, all ``put()`` calls are silent no-ops. + ``get()`` still works normally. Defaults to ``False``. + min_cache_size_bytes: When set to a positive integer, files whose + ``key.size`` is strictly below this threshold are not inserted. + ``None`` and ``0`` disable the threshold (default behaviour). + Negative values raise ``ValueError``. Defaults to ``None``. + + Raises: + ValueError: If ``min_cache_size_bytes`` is negative, or if the + target database contains a ``file_hash_cache`` table with an + outdated schema (missing ``cached_at``). + """ + + def __init__( + self, + conninfo: str, + *, + read_only: bool = False, + min_cache_size_bytes: int | None = None, + ) -> None: + if min_cache_size_bytes is not None and min_cache_size_bytes < 0: + raise ValueError( + f"min_cache_size_bytes must be None or a non-negative integer, " + f"got {min_cache_size_bytes!r}" + ) + self._conninfo = conninfo + self._read_only = read_only + self._min_cache_size_bytes = min_cache_size_bytes + self._local = threading.local() + self._ensure_schema() + + def _ensure_schema(self) -> None: + """Create cache tables and verify the schema version. + + Uses a dedicated one-shot connection so schema setup happens once on + construction, independent of the thread-local connection pool. + + The companion ``file_hash_cache_meta`` table stores a + ``('schema_version', N)`` row. If the ``file_hash_cache`` table + already exists but is missing the ``cached_at`` column, a + ``ValueError`` is raised with the migration DDL to apply. + """ + with psycopg.connect(self._conninfo) as conn: + # Main cache table (idempotent). + conn.execute( + """ + 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) + ) + """ + ) + # Schema-version metadata table (idempotent). + conn.execute( + """ + CREATE TABLE IF NOT EXISTS file_hash_cache_meta ( + key TEXT NOT NULL PRIMARY KEY, + value TEXT NOT NULL + ) + """ + ) + + # Detect old schema: file_hash_cache exists but lacks cached_at. + # Constrain table_schema to the current search_path schemas so that + # a same-named table in a different schema does not shadow the check. + row = conn.execute( + """ + SELECT column_name + FROM information_schema.columns + WHERE table_name = 'file_hash_cache' + AND column_name = 'cached_at' + AND table_schema = ANY(current_schemas(false)) + """ + ).fetchone() + if row is None: + raise ValueError( + "PostgreSQL hash cache table exists but uses an outdated schema " + "(missing the 'cached_at' column, schema version 0). " + "Apply the following migration and then retry:\n\n" + " ALTER TABLE file_hash_cache\n" + " ADD COLUMN cached_at BIGINT NOT NULL\n" + " DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT;\n" + " INSERT INTO file_hash_cache_meta (key, value)\n" + " VALUES ('schema_version', '1')\n" + " ON CONFLICT (key) DO UPDATE SET value = '1';\n" + ) + + # Record (or update) schema version so the meta table always + # reflects what the running code has verified/initialized. + conn.execute( + """ + INSERT INTO file_hash_cache_meta (key, value) + VALUES ('schema_version', %s) + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value + """, + (str(_SCHEMA_VERSION),), + ) + + def _connection(self) -> "psycopg.Connection[tuple[object, ...]]": + """Return this thread's connection, opening it on first use.""" + conn = getattr(self._local, "conn", None) + if conn is None: + conn = psycopg.connect(self._conninfo) + self._local.conn = conn + return conn + + def get(self, key: FileHashKey) -> ContentHash | None: + """Return the cached ``ContentHash`` for ``key``, or ``None`` on miss. + + Args: + key: File hash cache key. + + Returns: + Cached ``ContentHash``, or ``None`` if not found. + """ + conn = self._connection() + row = conn.execute( + "SELECT hash FROM file_hash_cache " + "WHERE path=%s AND mtime_ns=%s AND size=%s", + (str(key.path), key.mtime_ns, key.size), + ).fetchone() + if row is None: + return None + blob: bytes = bytes(row[0]) + method_bytes, digest = blob.split(b":", 1) + return ContentHash(method=method_bytes.decode("ascii"), digest=digest) + + def put(self, key: FileHashKey, value: ContentHash) -> None: + """Store ``value`` under ``key``. + + No-ops silently when ``read_only=True`` or when ``key.size`` is below + ``min_cache_size_bytes``. Uses ``INSERT ... ON CONFLICT DO NOTHING`` + so concurrent inserts of the same key from multiple workers are safe. + + Args: + key: File hash cache key. + value: ``ContentHash`` to store. + """ + if self._read_only: + return + if ( + self._min_cache_size_bytes is not None + and self._min_cache_size_bytes > 0 + and key.size < self._min_cache_size_bytes + ): + return + conn = self._connection() + conn.execute( + """ + INSERT INTO file_hash_cache (path, mtime_ns, size, hash) + VALUES (%s, %s, %s, %s) + ON CONFLICT (path, mtime_ns, size) DO NOTHING + """, + (str(key.path), key.mtime_ns, key.size, value.to_prefixed_digest()), + ) + conn.commit() + + def clear(self) -> None: + """Delete all rows from the cache table.""" + conn = self._connection() + conn.execute("DELETE FROM file_hash_cache") + conn.commit() + + def close(self) -> None: + """Close this thread's database connection.""" + conn = getattr(self._local, "conn", None) + if conn is not None: + conn.close() + self._local.conn = None + + def __enter__(self) -> "PostgresHashCacher": + """Return self for use as a context manager.""" + return self + + def __exit__(self, *_: object) -> None: + """Close the thread-local connection on exit.""" + self.close() + + def __repr__(self) -> str: + return ( + f"PostgresHashCacher(" + f"conninfo={_redact_conninfo(self._conninfo)!r}, " + f"read_only={self._read_only!r}, " + f"min_cache_size_bytes={self._min_cache_size_bytes!r})" + ) diff --git a/superpowers/plans/2026-07-10-itl-520-postgres-hash-cacher.md b/superpowers/plans/2026-07-10-itl-520-postgres-hash-cacher.md new file mode 100644 index 00000000..83cebcb6 --- /dev/null +++ b/superpowers/plans/2026-07-10-itl-520-postgres-hash-cacher.md @@ -0,0 +1,1016 @@ +# PostgreSQL Hash Cacher (ITL-520) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use sensei:subagent-driven-development (recommended) or sensei:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add `PostgresHashCacher` — a Postgres-backed drop-in alternative to `SqliteHashCacher` — plus a `conninfo` parameter on `enable_file_hash_caching()` to activate it. + +**Architecture:** `PostgresHashCacher` lives in its own module (`postgres_hash_cacher.py`), uses thread-local psycopg3 connections, and stores hashes in a `file_hash_cache` table with `INSERT ... ON CONFLICT DO NOTHING` for concurrency safety. It is exported from `orcapod.hashing` under a `try/except ImportError` guard so users without `psycopg` installed pay nothing. Tests use `testcontainers.postgres.PostgresContainer` against a real Postgres. + +**Tech Stack:** Python 3.12, `psycopg[binary]>=3.0`, `testcontainers[postgres]>=4.0.0`, `uv run pytest` + +**Spec:** `superpowers/specs/2026-07-10-itl-520-postgres-hash-cacher-design.md` + +**Branch:** `eywalker/itl-520-filehasher-add-postgresql-backend-for-the-hash-cacher` + +--- + +## File Map + +| File | Action | Responsibility | +|------|--------|----------------| +| `src/orcapod/hashing/postgres_hash_cacher.py` | Create | `PostgresHashCacher` — full implementation | +| `tests/test_hashing/test_postgres_hash_cacher.py` | Create | All `PostgresHashCacher` tests via testcontainers | +| `src/orcapod/hashing/__init__.py` | Modify | Export `PostgresHashCacher` under `try/except ImportError` | +| `src/orcapod/contexts/__init__.py` | Modify | Add `conninfo` param to `enable_file_hash_caching()` | +| `tests/test_hashing/test_hash_cachers.py` | Modify | Add `ValueError` test for mutually-exclusive `conninfo`/`db_path` | +| `pyproject.toml` | Modify | Add `testcontainers[postgres]>=4.0.0` to dev deps | +| `docs/concepts/file-hash-caching.md` | Modify | Add "Choosing a backend: SQLite vs Postgres" section | + +--- + +## Task 0: Check out the feature branch + +**Files:** none + +- [ ] **Step 0.1: Check out the branch** + +```bash +git checkout main +git pull +git checkout -b eywalker/itl-520-filehasher-add-postgresql-backend-for-the-hash-cacher +git branch --show-current +``` + +Expected output: `eywalker/itl-520-filehasher-add-postgresql-backend-for-the-hash-cacher` + +--- + +## Task 1: Add `testcontainers[postgres]` to dev deps + +**Files:** +- Modify: `pyproject.toml` + +- [ ] **Step 1.1: Add the dev dependency** + +Open `pyproject.toml`. In the `[dependency-groups]` → `dev` list, add `testcontainers[postgres]` immediately after the existing `testcontainers[minio]` line: + +```toml + "testcontainers[minio]>=4.0.0", + "testcontainers[postgres]>=4.0.0", +``` + +- [ ] **Step 1.2: Sync and verify** + +```bash +uv sync +uv run python -c "from testcontainers.postgres import PostgresContainer; print('OK')" +``` + +Expected: `OK` + +- [ ] **Step 1.3: Commit** + +```bash +git add pyproject.toml uv.lock +git commit -m "chore(deps): add testcontainers[postgres] to dev deps (ITL-520)" +``` + +--- + +## Task 2: Write failing tests for `PostgresHashCacher` basic operations + +Write the tests first. They will fail with `ImportError` / `ModuleNotFoundError` because `postgres_hash_cacher.py` does not exist yet. + +**Files:** +- Create: `tests/test_hashing/test_postgres_hash_cacher.py` + +- [ ] **Step 2.1: Create the test file** + +Create `tests/test_hashing/test_postgres_hash_cacher.py` with this content: + +```python +"""Tests for PostgresHashCacher using a real Postgres via testcontainers.""" + +from __future__ import annotations + +import threading + +import pytest +from upath import UPath + +psycopg = pytest.importorskip("psycopg") + +from testcontainers.postgres import PostgresContainer # noqa: E402 + +from orcapod.hashing.file_hashers import FileHashKey # noqa: E402 +from orcapod.hashing.postgres_hash_cacher import PostgresHashCacher # noqa: E402 +from orcapod.types import ContentHash # noqa: E402 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_key( + path: str = "/a/b.txt", mtime_ns: int = 1000, size: int = 100 +) -> FileHashKey: + return FileHashKey(path=UPath(path), mtime_ns=mtime_ns, size=size) + + +def make_hash(digest: bytes = b"\xab" * 32) -> ContentHash: + return ContentHash(method="sha256", digest=digest) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def pg_conninfo(): + """Start one Postgres container for the entire module; yield a conninfo string.""" + with PostgresContainer("postgres:16") as pg: + yield ( + f"host={pg.get_container_host_ip()} " + f"port={pg.get_exposed_port(5432)} " + f"dbname={pg.dbname} " + f"user={pg.username} " + f"password={pg.password}" + ) + + +@pytest.fixture() +def cacher(pg_conninfo): + """Fresh PostgresHashCacher with a cleared table for each test.""" + c = PostgresHashCacher(pg_conninfo) + c.clear() + yield c + c.close() + + +# --------------------------------------------------------------------------- +# Basic operations +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherBasic: + def test_miss_returns_none(self, cacher): + assert cacher.get(make_key()) is None + + def test_put_then_get_returns_hit(self, cacher): + key = make_key() + value = make_hash() + cacher.put(key, value) + result = cacher.get(key) + assert result is not None + assert result.method == value.method + assert result.digest == value.digest + + def test_different_path_is_miss(self, cacher): + cacher.put(make_key("/a/b.txt"), make_hash(b"\xaa" * 32)) + assert cacher.get(make_key("/a/c.txt")) is None + + def test_different_mtime_ns_is_miss(self, cacher): + cacher.put(make_key(mtime_ns=1000), make_hash(b"\xaa" * 32)) + assert cacher.get(make_key(mtime_ns=2000)) is None + + def test_different_size_is_miss(self, cacher): + cacher.put(make_key(size=100), make_hash(b"\xaa" * 32)) + assert cacher.get(make_key(size=200)) is None + + def test_clear_empties_table(self, cacher): + key = make_key() + cacher.put(key, make_hash()) + cacher.clear() + assert cacher.get(key) is None + + def test_persistence_across_instances(self, pg_conninfo): + key = make_key("/persist.txt", mtime_ns=9999, size=512) + value = make_hash(b"\xcc" * 32) + + cacher1 = PostgresHashCacher(pg_conninfo) + cacher1.clear() + cacher1.put(key, value) + cacher1.close() + + cacher2 = PostgresHashCacher(pg_conninfo) + result = cacher2.get(key) + cacher2.close() + + assert result is not None + assert result.method == value.method + assert result.digest == value.digest + + def test_context_manager_closes_connection(self, pg_conninfo): + key = make_key("/ctx.txt", mtime_ns=7777, size=64) + value = make_hash(b"\xdd" * 32) + + with PostgresHashCacher(pg_conninfo) as c: + c.clear() + c.put(key, value) + + # Verify persistence (connection was closed but data committed) + cacher2 = PostgresHashCacher(pg_conninfo) + result = cacher2.get(key) + cacher2.close() + assert result is not None + assert result.digest == value.digest + + +# --------------------------------------------------------------------------- +# Write guards +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherReadOnly: + def test_put_is_noop_when_read_only(self, pg_conninfo): + key = make_key("/ro.txt") + value = make_hash() + + cacher = PostgresHashCacher(pg_conninfo, read_only=True) + cacher.clear() + cacher.put(key, value) + assert cacher.get(key) is None + cacher.close() + + def test_read_only_can_read_preexisting(self, pg_conninfo): + key = make_key("/ro_pre.txt", mtime_ns=5555, size=256) + value = make_hash(b"\xee" * 32) + + writable = PostgresHashCacher(pg_conninfo) + writable.clear() + writable.put(key, value) + writable.close() + + read_only = PostgresHashCacher(pg_conninfo, read_only=True) + result = read_only.get(key) + read_only.close() + + assert result is not None + assert result.digest == value.digest + + def test_default_is_not_read_only(self, cacher): + key = make_key() + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + + +@pytest.mark.postgres +class TestPostgresHashCacherThreshold: + def test_small_file_not_cached(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=100) + cacher.clear() + key = make_key(size=50) + cacher.put(key, make_hash()) + assert cacher.get(key) is None + cacher.close() + + def test_file_at_threshold_is_cached(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=100) + cacher.clear() + key = make_key(size=100) + value = make_hash() + cacher.put(key, value) + result = cacher.get(key) + cacher.close() + assert result is not None + assert result.digest == value.digest + + def test_large_file_is_cached(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=100) + cacher.clear() + key = make_key(size=200) + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + cacher.close() + + def test_none_threshold_caches_all(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=None) + cacher.clear() + key = make_key(size=1) + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + cacher.close() + + def test_zero_threshold_caches_all(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=0) + cacher.clear() + key = make_key(size=1) + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + cacher.close() + + +# --------------------------------------------------------------------------- +# Concurrency +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherConcurrency: + def test_concurrent_insert_no_error_one_row(self, pg_conninfo): + """Two threads inserting the same key simultaneously → no error, one row.""" + key = make_key("/concurrent.txt", mtime_ns=12345, size=999) + value = make_hash(b"\xff" * 32) + + # Pre-clear using a dedicated cacher + setup = PostgresHashCacher(pg_conninfo) + setup.clear() + setup.close() + + errors: list[Exception] = [] + + def insert() -> None: + try: + c = PostgresHashCacher(pg_conninfo) + c.put(key, value) + c.close() + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=insert) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [], f"Concurrent inserts raised errors: {errors}" + + # Verify exactly one row was written + with psycopg.connect(pg_conninfo) as conn: + count = conn.execute( + "SELECT COUNT(*) FROM file_hash_cache " + "WHERE path=%s AND mtime_ns=%s AND size=%s", + (str(key.path), key.mtime_ns, key.size), + ).fetchone()[0] + assert count == 1 + + +# --------------------------------------------------------------------------- +# Parity with SqliteHashCacher +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherParity: + def test_parity_with_sqlite(self, pg_conninfo, tmp_path): + """Same key/value → equivalent ContentHash from both backends.""" + from orcapod.hashing.hash_cachers import SqliteHashCacher + + key = make_key("/parity.txt", mtime_ns=111, size=222) + value = make_hash(b"\x12" * 32) + + pg_cacher = PostgresHashCacher(pg_conninfo) + pg_cacher.clear() + pg_cacher.put(key, value) + pg_result = pg_cacher.get(key) + pg_cacher.close() + + sq_cacher = SqliteHashCacher(tmp_path / "parity.db") + sq_cacher.put(key, value) + sq_result = sq_cacher.get(key) + + assert pg_result is not None + assert sq_result is not None + assert pg_result.method == sq_result.method + assert pg_result.digest == sq_result.digest + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherValidation: + def test_negative_min_cache_size_bytes_raises(self, pg_conninfo): + with pytest.raises(ValueError, match="non-negative"): + PostgresHashCacher(pg_conninfo, min_cache_size_bytes=-1) + + def test_repr_includes_conninfo_and_guards(self, pg_conninfo): + cacher = PostgresHashCacher( + pg_conninfo, read_only=True, min_cache_size_bytes=1024 + ) + r = repr(cacher) + assert "PostgresHashCacher" in r + assert "read_only=True" in r + assert "min_cache_size_bytes=1024" in r + cacher.close() +``` + +- [ ] **Step 2.2: Run tests to verify they fail at import** + +```bash +uv run pytest tests/test_hashing/test_postgres_hash_cacher.py -v 2>&1 | head -20 +``` + +Expected: error containing `ModuleNotFoundError` or `ImportError: cannot import name 'PostgresHashCacher'` + +--- + +## Task 3: Implement `PostgresHashCacher` + +**Files:** +- Create: `src/orcapod/hashing/postgres_hash_cacher.py` + +- [ ] **Step 3.1: Create the implementation file** + +Create `src/orcapod/hashing/postgres_hash_cacher.py` with this content: + +```python +"""PostgreSQL-backed file hash cacher. + +Provides ``PostgresHashCacher`` — a network-accessible, concurrent-safe +implementation of ``CacherProtocol[FileHashKey, ContentHash]``. Requires +the optional ``psycopg`` driver (``pip install 'orcapod[postgresql]'``). +""" + +from __future__ import annotations + +import threading +from typing import TYPE_CHECKING + +from orcapod.hashing.file_hashers import FileHashKey +from orcapod.types import ContentHash + +if TYPE_CHECKING: + import psycopg as _psycopg + + +class PostgresHashCacher: + """PostgreSQL-backed file hash cacher. + + Stores file hashes keyed on ``(path, mtime_ns, size)`` in a shared + PostgreSQL database. Uses thread-local connections for thread safety + and ``INSERT ... ON CONFLICT DO NOTHING`` for concurrent-insert safety. + + The hash is stored as a BYTEA in ``{method}:{raw_digest}`` format via + ``ContentHash.to_prefixed_digest()``. + + Requires ``psycopg[binary]>=3.0`` (install with + ``pip install 'orcapod[postgresql]'``). + + Args: + conninfo: psycopg3 connection string, e.g. + ``"postgresql://user:pass@host:5432/dbname"`` or keyword DSN + ``"host=myhost dbname=mydb user=myuser password=mypass"``. + read_only: When ``True``, all ``put()`` calls are silent no-ops. + ``get()`` still works normally. Defaults to ``False``. + min_cache_size_bytes: When set to a positive integer, files whose + ``key.size`` is strictly below this threshold are not inserted. + ``None`` and ``0`` disable the threshold (default behaviour). + Negative values raise ``ValueError``. Defaults to ``None``. + """ + + def __init__( + self, + conninfo: str, + *, + read_only: bool = False, + min_cache_size_bytes: int | None = None, + ) -> None: + try: + import psycopg # noqa: F401 + except ImportError: + raise ImportError( + "PostgresHashCacher requires psycopg. " + "Install it with: pip install 'orcapod[postgresql]'" + ) from None + if min_cache_size_bytes is not None and min_cache_size_bytes < 0: + raise ValueError( + f"min_cache_size_bytes must be None or a non-negative integer, " + f"got {min_cache_size_bytes!r}" + ) + self._conninfo = conninfo + self._read_only = read_only + self._min_cache_size_bytes = min_cache_size_bytes + self._local = threading.local() + self._ensure_schema() + + def _ensure_schema(self) -> None: + """Create the cache table if it does not exist. + + Uses a dedicated one-shot connection so schema setup happens once + on construction, independent of the thread-local connection pool. + """ + import psycopg + + with psycopg.connect(self._conninfo) as conn: + conn.execute( + """ + 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) + ) + """ + ) + + def _connection(self) -> "_psycopg.Connection[tuple[object, ...]]": + """Return this thread's connection, opening it on first use.""" + import psycopg + + conn = getattr(self._local, "conn", None) + if conn is None: + conn = psycopg.connect(self._conninfo) + self._local.conn = conn + return conn + + def get(self, key: FileHashKey) -> ContentHash | None: + """Return the cached ``ContentHash`` for ``key``, or ``None`` on miss. + + Args: + key: File hash cache key. + + Returns: + Cached ``ContentHash``, or ``None`` if not found. + """ + conn = self._connection() + row = conn.execute( + "SELECT hash FROM file_hash_cache " + "WHERE path=%s AND mtime_ns=%s AND size=%s", + (str(key.path), key.mtime_ns, key.size), + ).fetchone() + if row is None: + return None + blob: bytes = bytes(row[0]) + method_bytes, digest = blob.split(b":", 1) + return ContentHash(method=method_bytes.decode("ascii"), digest=digest) + + def put(self, key: FileHashKey, value: ContentHash) -> None: + """Store ``value`` under ``key``. + + No-ops silently when ``read_only=True`` or when ``key.size`` is below + ``min_cache_size_bytes``. Uses ``INSERT ... ON CONFLICT DO NOTHING`` + so concurrent inserts of the same key from multiple workers are safe. + + Args: + key: File hash cache key. + value: ``ContentHash`` to store. + """ + if self._read_only: + return + if ( + self._min_cache_size_bytes is not None + and self._min_cache_size_bytes > 0 + and key.size < self._min_cache_size_bytes + ): + return + conn = self._connection() + conn.execute( + """ + INSERT INTO file_hash_cache (path, mtime_ns, size, hash) + VALUES (%s, %s, %s, %s) + ON CONFLICT (path, mtime_ns, size) DO NOTHING + """, + (str(key.path), key.mtime_ns, key.size, value.to_prefixed_digest()), + ) + conn.commit() + + def clear(self) -> None: + """Delete all rows from the cache table.""" + conn = self._connection() + conn.execute("DELETE FROM file_hash_cache") + conn.commit() + + def close(self) -> None: + """Close this thread's database connection.""" + conn = getattr(self._local, "conn", None) + if conn is not None: + conn.close() + self._local.conn = None + + def __enter__(self) -> "PostgresHashCacher": + """Return self for use as a context manager.""" + return self + + def __exit__(self, *_: object) -> None: + """Close the thread-local connection on exit.""" + self.close() + + def __repr__(self) -> str: + return ( + f"PostgresHashCacher(" + f"conninfo={self._conninfo!r}, " + f"read_only={self._read_only!r}, " + f"min_cache_size_bytes={self._min_cache_size_bytes!r})" + ) +``` + +- [ ] **Step 3.2: Run all Postgres tests** + +```bash +uv run pytest tests/test_hashing/test_postgres_hash_cacher.py -v -m postgres +``` + +Expected: all tests PASS. (Container spin-up takes ~10–20 s on first run.) + +- [ ] **Step 3.3: Commit** + +```bash +git add src/orcapod/hashing/postgres_hash_cacher.py tests/test_hashing/test_postgres_hash_cacher.py +git commit -m "feat(hashing): add PostgresHashCacher with thread-local psycopg3 connections (ITL-520)" +``` + +--- + +## Task 4: Export `PostgresHashCacher` from `orcapod.hashing` + +**Files:** +- Modify: `src/orcapod/hashing/__init__.py` + +- [ ] **Step 4.1: Add the guarded import** + +Open `src/orcapod/hashing/__init__.py`. Find the existing `try/except` block for `legacy_core` near the bottom of the import section (around line 86). Add a new `try/except` block immediately after the `from orcapod.hashing.hash_cachers import ...` line (around line 53): + +```python +from orcapod.hashing.hash_cachers import InMemoryHashCacher, SqliteHashCacher + +try: + from orcapod.hashing.postgres_hash_cacher import PostgresHashCacher +except ImportError: + PostgresHashCacher = None # type: ignore[assignment,misc] +``` + +Then add `"PostgresHashCacher"` to the `__all__` list at the bottom of the file, after `"SqliteHashCacher"`: + +```python + "InMemoryHashCacher", + "SqliteHashCacher", + "PostgresHashCacher", +``` + +- [ ] **Step 4.2: Verify the export works** + +```bash +uv run python -c "from orcapod.hashing import PostgresHashCacher; print(PostgresHashCacher)" +``` + +Expected: `` + +- [ ] **Step 4.3: Verify existing tests still pass** + +```bash +uv run pytest tests/test_hashing/test_hash_cachers.py -v +``` + +Expected: all PASS (no regressions). + +- [ ] **Step 4.4: Commit** + +```bash +git add src/orcapod/hashing/__init__.py +git commit -m "feat(hashing): export PostgresHashCacher from orcapod.hashing (ITL-520)" +``` + +--- + +## Task 5: Extend `enable_file_hash_caching()` with `conninfo` parameter + +**Files:** +- Modify: `src/orcapod/contexts/__init__.py` +- Modify: `tests/test_hashing/test_hash_cachers.py` + +- [ ] **Step 5.1: Write the failing ValueError test first** + +Open `tests/test_hashing/test_hash_cachers.py`. Add this new class at the end of the file (after `TestEnableFileHashCaching`): + +```python +class TestEnableFileHashCachingConninfo: + def test_conninfo_and_db_path_raises(self, restore_default_file_handler, tmp_path): + """Providing both conninfo and db_path raises ValueError.""" + from orcapod.contexts import enable_file_hash_caching + + with pytest.raises(ValueError, match="not both"): + enable_file_hash_caching( + db_path=tmp_path / "x.db", + conninfo="postgresql://unused", + ) +``` + +- [ ] **Step 5.2: Run the failing test** + +```bash +uv run pytest tests/test_hashing/test_hash_cachers.py::TestEnableFileHashCachingConninfo -v +``` + +Expected: FAIL — `enable_file_hash_caching` does not yet accept `conninfo`. + +- [ ] **Step 5.3: Update `enable_file_hash_caching()`** + +Open `src/orcapod/contexts/__init__.py`. Find the `enable_file_hash_caching` function (around line 231). Replace the entire function signature and docstring with: + +```python +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 file hash caching on the default Orcapod context. + + When ``conninfo`` is provided, uses ``PostgresHashCacher`` for a shared, + network-accessible cache suitable for multi-machine pipelines. Otherwise + uses ``SqliteHashCacher`` for a local, per-machine cache (default). + + Call once at application startup before any file hashing occurs. + + If the handler already wraps a ``CachedFileHasher`` (i.e. this function + was already called), a warning is logged, all existing caching layers are + unwrapped to reach the original base hasher, and the new cacher is applied + around that base hasher. This keeps the system in a well-defined state + (exactly one caching layer) regardless of how many times this function + is called. + + For intentional multi-layer caching (e.g. in-memory L1 + SQLite L2), + construct a ``CachedFileHasher`` manually and register it directly via + the context's ``type_handler_registry`` instead. + + Also patches ``DirectoryHandler`` for ``orcapod.Directory``, using the + **same** ``CachedFileHasher`` instance. This means a file that was + cached via a direct ``op.File`` hash is also a cache hit when the same + file is encountered during directory traversal. + + Args: + db_path: Path to the SQLite cache database. Ignored when ``conninfo`` + is provided. Defaults to ``~/.orcapod/file_hash_cache.db`` or + the ``ORCAPOD_HASH_CACHE_DB`` environment variable. + conninfo: psycopg3 connection string for a PostgreSQL cache database, + e.g. ``"postgresql://user:pass@host:5432/db"``. When provided, + ``PostgresHashCacher`` is used instead of ``SqliteHashCacher``. + Mutually exclusive with ``db_path``. + 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." + ) +``` + +Then, inside the function body, replace the block that constructs `cached_file_hasher` (find the lines that build `SqliteHashCacher` and `CachedFileHasher`): + +```python + 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, + ) +``` + +The rest of the function body (registering `FileHandler` and `DirectoryHandler`) remains unchanged. + +- [ ] **Step 5.4: Run the ValueError test — should now pass** + +```bash +uv run pytest tests/test_hashing/test_hash_cachers.py::TestEnableFileHashCachingConninfo -v +``` + +Expected: PASS + +- [ ] **Step 5.5: Verify no regressions in existing enable_file_hash_caching tests** + +```bash +uv run pytest tests/test_hashing/test_hash_cachers.py -v +``` + +Expected: all PASS + +- [ ] **Step 5.6: Add Postgres-specific enable_file_hash_caching tests to the Postgres test file** + +Open `tests/test_hashing/test_postgres_hash_cacher.py`. Append this class at the end: + +```python +# --------------------------------------------------------------------------- +# enable_file_hash_caching(conninfo=...) integration +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def restore_default_file_handler(): + """Restore the default FileHandler and DirectoryHandler after each test.""" + from orcapod.contexts import get_default_context + from orcapod.extension_types.directory_type import Directory + from orcapod.extension_types.file_type import File + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + original_file_handler = registry.get_handler_for_type(File) + original_dir_handler = registry.get_handler_for_type(Directory) + yield + registry.register(File, original_file_handler) + registry.register(Directory, original_dir_handler) + + +@pytest.mark.postgres +class TestEnableFileHashCachingPostgres: + def test_conninfo_activates_postgres_cacher( + self, restore_default_file_handler, pg_conninfo + ): + """enable_file_hash_caching(conninfo=...) wires up PostgresHashCacher.""" + from orcapod.contexts import enable_file_hash_caching, get_default_context + from orcapod.extension_types.file_type import File + + enable_file_hash_caching(conninfo=pg_conninfo) + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + handler = registry.get_handler_for_type(File) + assert isinstance(handler.file_hasher.cacher, PostgresHashCacher) + + def test_conninfo_passes_read_only( + self, restore_default_file_handler, pg_conninfo + ): + """read_only=True is forwarded to PostgresHashCacher.""" + from orcapod.contexts import enable_file_hash_caching, get_default_context + from orcapod.extension_types.file_type import File + + enable_file_hash_caching(conninfo=pg_conninfo, read_only=True) + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + handler = registry.get_handler_for_type(File) + assert handler.file_hasher.cacher._read_only is True + + def test_conninfo_passes_min_cache_size_bytes( + self, restore_default_file_handler, pg_conninfo + ): + """min_cache_size_bytes is forwarded to PostgresHashCacher.""" + from orcapod.contexts import enable_file_hash_caching, get_default_context + from orcapod.extension_types.file_type import File + + enable_file_hash_caching(conninfo=pg_conninfo, min_cache_size_bytes=2048) + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + handler = registry.get_handler_for_type(File) + assert handler.file_hasher.cacher._min_cache_size_bytes == 2048 +``` + +- [ ] **Step 5.7: Run the new integration tests** + +```bash +uv run pytest tests/test_hashing/test_postgres_hash_cacher.py::TestEnableFileHashCachingPostgres -v -m postgres +``` + +Expected: all PASS + +- [ ] **Step 5.8: Commit** + +```bash +git add src/orcapod/contexts/__init__.py \ + tests/test_hashing/test_hash_cachers.py \ + tests/test_hashing/test_postgres_hash_cacher.py +git commit -m "feat(contexts): add conninfo param to enable_file_hash_caching() for Postgres backend (ITL-520)" +``` + +--- + +## Task 6: Update documentation + +**Files:** +- Modify: `docs/concepts/file-hash-caching.md` + +- [ ] **Step 6.1: Add the backend comparison section** + +Open `docs/concepts/file-hash-caching.md`. Find the end of the file and append the following section. (If there is already a section about "Controlling when the cache is written" from ITL-519, add the new section after it.) + +```markdown +## 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**. +``` + +- [ ] **Step 6.2: Commit** + +```bash +git add docs/concepts/file-hash-caching.md +git commit -m "docs(hashing): add Postgres backend section to file-hash-caching docs (ITL-520)" +``` + +--- + +## Task 7: Full test run and PR + +**Files:** none + +- [ ] **Step 7.1: Run the full hashing test suite** + +```bash +uv run pytest tests/test_hashing/ -v +``` + +Expected: all tests PASS (Postgres tests require Docker; SQLite/in-memory tests run unconditionally). + +- [ ] **Step 7.2: Run the broader test suite for regressions** + +```bash +uv run pytest tests/ -v --ignore=tests/test_hashing/test_postgres_hash_cacher.py -q +``` + +Expected: all tests PASS, no regressions. + +- [ ] **Step 7.3: Push and open PR** + +```bash +git push -u origin eywalker/itl-520-filehasher-add-postgresql-backend-for-the-hash-cacher +``` + +Then open a PR targeting `main` with a description referencing `Closes ITL-520`. diff --git a/superpowers/specs/2026-07-10-itl-520-postgres-hash-cacher-design.md b/superpowers/specs/2026-07-10-itl-520-postgres-hash-cacher-design.md new file mode 100644 index 00000000..e2b7c49e --- /dev/null +++ b/superpowers/specs/2026-07-10-itl-520-postgres-hash-cacher-design.md @@ -0,0 +1,269 @@ +# ITL-520: FileHasher — PostgreSQL Backend for the Hash Cacher + +**Date:** 2026-07-10 +**Issue:** [ITL-520](https://linear.app/enigma-metamorphic/issue/ITL-520) +**Status:** Approved + +--- + +## Overview + +Add `PostgresHashCacher` — a Postgres-backed implementation of the `CacherProtocol[FileHashKey, ContentHash]` that mirrors `SqliteHashCacher` in API and semantics but stores hashes in a shared, network-accessible Postgres database. Multiple machines and pipeline runs can consult and populate the cache concurrently without the per-machine limitation of a SQLite file. + +--- + +## Goals & Success Criteria + +- New `PostgresHashCacher` class implementing `CacherProtocol[FileHashKey, ContentHash]`. +- Constructor accepts a psycopg3 `conninfo` string (URL or keyword DSN). +- Idempotent schema creation on first use (`CREATE TABLE IF NOT EXISTS`). +- `INSERT ... ON CONFLICT DO NOTHING` for concurrency-safe writes. +- Respects `read_only` and `min_cache_size_bytes` write guards (same semantics as `SqliteHashCacher`). +- `enable_file_hash_caching()` extended with a `conninfo` parameter to activate Postgres caching. +- `PostgresHashCacher` exported from `orcapod.hashing` under a `try/except ImportError` guard. +- Tests against a real Postgres via `testcontainers` covering hit/miss, concurrency, read-only, threshold, and SQLite parity. +- `psycopg` remains an optional dependency (`orcapod[postgresql]`); users on SQLite pay nothing. + +--- + +## Design + +### 1. New module: `src/orcapod/hashing/postgres_hash_cacher.py` + +`PostgresHashCacher` is a close structural mirror of `SqliteHashCacher`: + +```python +class PostgresHashCacher: + def __init__( + self, + conninfo: str, + *, + read_only: bool = False, + min_cache_size_bytes: int | None = None, + ) -> None: +``` + +- **`conninfo`** — psycopg3 connection string, e.g. + `"postgresql://user:pass@host:5432/dbname"` or `"host=... dbname=... user=..."`. +- **`read_only`** / **`min_cache_size_bytes`** — identical semantics to `SqliteHashCacher`. +- Raises `ImportError` with a helpful install hint if `psycopg` is not available. +- Raises `ValueError` if `min_cache_size_bytes` is negative. +- Calls `_ensure_schema()` in `__init__` using a one-shot connection. + +**Connection management:** `threading.local()` — each thread opens its own +`psycopg.connect(conninfo)` on first use, same pattern as `SqliteHashCacher`. + +**Methods:** `get()`, `put()`, `clear()`, `close()`, `__enter__`/`__exit__`, `__repr__`. + +### 2. PostgreSQL schema + +```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) +) +``` + +- `PRIMARY KEY (path, mtime_ns, size)` creates a B-tree index — fast lookup with no + separate `CREATE INDEX` needed. +- `BIGINT` covers all realistic `mtime_ns` and `size` values. +- `BYTEA` matches the `{method}:{raw_digest}` blob format used by `SqliteHashCacher`. +- Table name `file_hash_cache` is the same as the SQLite version (they live in separate + databases, so there is no conflict). + +### 3. Concurrency-safe insert + +```sql +INSERT INTO file_hash_cache (path, mtime_ns, size, hash) +VALUES (%s, %s, %s, %s) +ON CONFLICT (path, mtime_ns, size) DO NOTHING +``` + +Two workers racing to insert the same key: exactly one row lands, the other's insert +silently completes with zero rows affected — no error, no duplicate. The miss → compute → +insert pattern is safe even if another worker inserts between the miss and the insert. + +Note: unlike `SqliteHashCacher` which uses `INSERT OR REPLACE` (last-writer wins), +`PostgresHashCacher` uses `DO NOTHING` (first-writer wins). For a hash cache the value +for any given key is always the same deterministic hash, so either policy is correct. + +### 4. `put()` write-guard order + +Identical to `SqliteHashCacher`: + +1. `read_only=True` → return immediately (no write, no error). +2. `min_cache_size_bytes` truthy and `key.size < min_cache_size_bytes` → return immediately. +3. Otherwise → execute `INSERT ... ON CONFLICT DO NOTHING`. + +`get()` is unaffected by either guard. + +### 5. `__repr__` + +```python +def __repr__(self) -> str: + return ( + f"PostgresHashCacher(" + f"conninfo={self._conninfo!r}, " + f"read_only={self._read_only!r}, " + f"min_cache_size_bytes={self._min_cache_size_bytes!r})" + ) +``` + +--- + +## Files Changed + +| File | Change | +|---|---| +| `src/orcapod/hashing/postgres_hash_cacher.py` | **New** — `PostgresHashCacher` implementation | +| `tests/test_hashing/test_postgres_hash_cacher.py` | **New** — testcontainers-based tests | +| `src/orcapod/hashing/__init__.py` | Export `PostgresHashCacher` under `try/except ImportError` | +| `src/orcapod/contexts/__init__.py` | Add `conninfo` param to `enable_file_hash_caching()` | +| `pyproject.toml` | Add `testcontainers[postgres]>=4.0.0` to `[dependency-groups] dev` | + +--- + +## `enable_file_hash_caching()` API change + +```python +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: +``` + +All parameters are keyword-only (note the leading `*`). + +- `conninfo` provided, `db_path` absent → use `PostgresHashCacher`. +- `db_path` provided (or both absent) and `conninfo` absent → use `SqliteHashCacher` as today. +- Both `conninfo` and `db_path` provided → raise `ValueError("Provide conninfo or db_path, not both")`. + +--- + +## `__init__.py` export + +```python +try: + from orcapod.hashing.postgres_hash_cacher import PostgresHashCacher +except ImportError: + PostgresHashCacher = None # type: ignore[assignment,misc] +``` + +`PostgresHashCacher` is added to `__all__`. Same pattern as the existing `legacy_core` +try/except block. + +--- + +## Tests + +**File:** `tests/test_hashing/test_postgres_hash_cacher.py` + +**Infrastructure:** +- `pytest.importorskip("psycopg")` at module level — skips the whole file if psycopg not installed. +- `@pytest.mark.postgres` marker on each test class. +- Module-scoped `pg_conninfo` fixture starts a `PostgresContainer("postgres:16")` once per + test session to avoid repeated container spin-up. +- Function-scoped `cacher` fixture creates a fresh `PostgresHashCacher` and calls `clear()` + before each test to prevent cross-test row contamination. + +```python +@pytest.fixture(scope="module") +def pg_conninfo(): + with PostgresContainer("postgres:16") as pg: + # Build a raw psycopg3 conninfo string (not a SQLAlchemy URL) + yield ( + f"host={pg.get_container_host_ip()} " + f"port={pg.get_exposed_port(5432)} " + f"dbname={pg.dbname} " + f"user={pg.username} " + f"password={pg.password}" + ) + +@pytest.fixture() +def cacher(pg_conninfo): + c = PostgresHashCacher(pg_conninfo) + c.clear() + yield c + c.close() +``` + +**Test cases:** + +| Test | What it verifies | +|---|---| +| `test_miss_returns_none` | `get()` on empty DB returns `None` | +| `test_put_then_get_returns_hit` | Round-trip: `put()` then `get()` returns same method and digest | +| `test_key_components_are_independent` | Different path / mtime_ns / size → miss | +| `test_persistence_across_instances` | Second `PostgresHashCacher(same_conninfo)` sees rows from first | +| `test_concurrent_insert_no_error_one_row` | Two threads insert same key simultaneously → no exception, exactly one DB row | +| `test_read_only_skips_writes` | `put()` with `read_only=True` → `get()` returns `None` | +| `test_read_only_can_read_preexisting` | Read-only instance `get()`s entries written by writable instance | +| `test_min_cache_size_bytes_small_file_not_cached` | File below threshold → not stored | +| `test_min_cache_size_bytes_boundary_is_inclusive` | File at threshold → stored | +| `test_parity_with_sqlite` | Same key/value → equivalent `ContentHash` from both backends | +| `test_clear_empties_table` | `clear()` → subsequent `get()` returns `None` | +| `test_context_manager_closes_connection` | `__exit__` closes thread-local connection | +| `test_negative_min_cache_size_bytes_raises` | `ValueError` on negative threshold | +| `test_repr_includes_conninfo_and_guards` | `__repr__` shows conninfo, read_only, min_cache_size_bytes | + +--- + +## Dependency changes + +**`pyproject.toml` — dev group only:** +```toml +"testcontainers[postgres]>=4.0.0", +``` +(The `orcapod[postgresql]` optional extra already declares `psycopg[binary]>=3.0` — no change.) + +--- + +## Documentation update + +**File:** `docs/concepts/file-hash-caching.md` + +Add a new section **"Choosing a backend: SQLite vs Postgres"** covering: + +- **SQLite**: zero-setup, per-machine, single-writer. Use for local development and + single-machine pipelines. +- **Postgres**: shared, network-accessible, concurrent multi-writer. Use when multiple + machines or pipeline workers share a cache. +- Minimum supported Postgres version: **14** (for `ON CONFLICT` and `EXTRACT EPOCH` support). +- Example connection config: + ```python + op.enable_file_hash_caching(conninfo="postgresql://user:pass@db-host:5432/orcapod_cache") + ``` +- Schema DDL reference (same as the `CREATE TABLE` above). + +--- + +## Scope & Boundaries + +In scope: +- `PostgresHashCacher` implementation and tests. +- `enable_file_hash_caching(conninfo=...)` extension. +- `orcapod.hashing.__init__` export. +- `testcontainers[postgres]` dev dep. +- Docs section on backend choice. + +Out of scope: +- Data migration between SQLite and Postgres. +- Multi-tenant / per-user table segmentation. +- TTL / eviction / size cap. +- Connection pooling (defer to follow-up). +- IAM / cloud auth schemes. +- Turso / other backends. + +--- + +## Dependencies + +- ITL-472 (FileHasher + `CacherProtocol` + `SqliteHashCacher`) — complete, merged. +- ITL-519 (read-only + min_cache_size_bytes write guards) — complete, merged. diff --git a/tests/test_hashing/test_function_info_extractors.py b/tests/test_hashing/test_function_info_extractors.py new file mode 100644 index 00000000..a89b9d85 --- /dev/null +++ b/tests/test_hashing/test_function_info_extractors.py @@ -0,0 +1,170 @@ +"""Tests for FunctionNameExtractor, FunctionSignatureExtractor, and FunctionInfoExtractorFactory.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + + +class TestFunctionNameExtractor: + def _make(self): + from orcapod.hashing.semantic_hashing.function_info_extractors import FunctionNameExtractor + return FunctionNameExtractor() + + def test_returns_function_name(self): + def my_func(): + pass + + result = self._make().extract_function_info(my_func) + assert result == {"name": "my_func"} + + def test_custom_function_name_overrides(self): + def my_func(): + pass + + result = self._make().extract_function_info(my_func, function_name="custom") + assert result["name"] == "custom" + + def test_non_callable_raises_type_error(self): + with pytest.raises(TypeError, match="not callable"): + self._make().extract_function_info("not a function") + + def test_lambda_name(self): + fn = lambda x: x # noqa: E731 + result = self._make().extract_function_info(fn) + assert "name" in result + + def test_ignores_typespec_args(self): + """input_typespec and output_typespec are accepted but not used.""" + def fn(x: int) -> str: + return str(x) + + result = self._make().extract_function_info(fn, input_typespec={"x": int}, output_typespec={"return": str}) + assert result == {"name": "fn"} + + +class TestFunctionSignatureExtractor: + def _make(self, **kwargs): + from orcapod.hashing.semantic_hashing.function_info_extractors import FunctionSignatureExtractor + return FunctionSignatureExtractor(**kwargs) + + def test_basic_extraction_returns_dict_with_name_and_params(self): + def my_func(x: int, y: str) -> bool: + return True + + result = self._make().extract_function_info(my_func) + assert "name" in result + assert result["name"] == "my_func" + assert "params" in result + assert "module" in result + + def test_non_callable_raises_type_error(self): + with pytest.raises(TypeError, match="not callable"): + self._make().extract_function_info(42) + + def test_include_module_false_omits_module(self): + def fn(): + pass + + result = self._make(include_module=False).extract_function_info(fn) + assert "module" not in result + + def test_include_module_true_adds_module(self): + def fn(): + pass + + result = self._make(include_module=True).extract_function_info(fn) + assert "module" in result + + def test_include_defaults_false_strips_defaults(self): + def fn(x: int = 42, y: str = "hi") -> None: + pass + + result = self._make(include_defaults=False).extract_function_info(fn) + # Default values should be stripped + assert "42" not in result["params"] + assert "hi" not in result["params"] + # Parameter names should still be present + assert "x" in result["params"] + assert "y" in result["params"] + + def test_include_defaults_true_keeps_defaults(self): + def fn(x: int = 42) -> None: + pass + + result = self._make(include_defaults=True).extract_function_info(fn) + assert "42" in result["params"] + + def test_return_annotation_present(self): + def fn() -> str: + return "hi" + + result = self._make().extract_function_info(fn) + assert "returns" in result + assert result["returns"] is str + + def test_no_return_annotation_omits_returns(self): + def fn(): + pass + + result = self._make().extract_function_info(fn) + assert "returns" not in result + + def test_function_name_override(self): + def fn(): + pass + + result = self._make().extract_function_info(fn, function_name="overridden") + assert result["name"] == "overridden" + + def test_union_annotation_canonicalized(self): + """Union annotations are canonicalized for order stability.""" + def fn1(x: str | Path) -> None: + pass + + def fn2(x: Path | str) -> None: + pass + + r1 = self._make(include_module=False).extract_function_info(fn1, function_name="fn") + r2 = self._make(include_module=False).extract_function_info(fn2, function_name="fn") + assert r1["params"] == r2["params"] + + def test_eval_str_fallback_on_unresolvable_annotation(self): + """When eval_str=True fails, falls back to unresolved signature.""" + # Create a function with an annotation that cannot be resolved at eval time + # by using exec with a forward reference that doesn't exist + code = "def fn(x: 'NonExistentType123') -> None: pass" + ns = {"__name__": "__test_module__"} + exec(compile(code, "", "exec"), ns) + fn = ns["fn"] + # eval_str=True will try to resolve 'NonExistentType123' from the function's + # __globals__, which won't have it → NameError → fallback + # This should not raise + result = self._make().extract_function_info(fn) + assert "params" in result + + +class TestFunctionInfoExtractorFactory: + def _factory(self): + from orcapod.hashing.semantic_hashing.function_info_extractors import FunctionInfoExtractorFactory + return FunctionInfoExtractorFactory + + def test_name_strategy_returns_function_name_extractor(self): + from orcapod.hashing.semantic_hashing.function_info_extractors import FunctionNameExtractor + extractor = self._factory().create_function_info_extractor("name") + assert isinstance(extractor, FunctionNameExtractor) + + def test_signature_strategy_returns_function_signature_extractor(self): + from orcapod.hashing.semantic_hashing.function_info_extractors import FunctionSignatureExtractor + extractor = self._factory().create_function_info_extractor("signature") + assert isinstance(extractor, FunctionSignatureExtractor) + + def test_default_strategy_is_signature(self): + from orcapod.hashing.semantic_hashing.function_info_extractors import FunctionSignatureExtractor + extractor = self._factory().create_function_info_extractor() + assert isinstance(extractor, FunctionSignatureExtractor) + + def test_invalid_strategy_raises_value_error(self): + with pytest.raises(ValueError, match="Unknown strategy"): + self._factory().create_function_info_extractor("invalid_strategy") diff --git a/tests/test_hashing/test_hash_cachers.py b/tests/test_hashing/test_hash_cachers.py index 5e78ff9a..2742d5f5 100644 --- a/tests/test_hashing/test_hash_cachers.py +++ b/tests/test_hashing/test_hash_cachers.py @@ -1,6 +1,7 @@ """Tests for InMemoryHashCacher and SqliteHashCacher.""" import sqlite3 +from pathlib import Path import pytest from upath import UPath @@ -304,3 +305,273 @@ def test_min_cache_size_bytes_kwarg_passes_through_to_cacher( assert isinstance(cacher, SqliteHashCacher) assert cacher._min_cache_size_bytes == 1024 + + +class TestEnableFileHashCachingConninfo: + def test_conninfo_and_db_path_raises(self, restore_default_file_handler, tmp_path): + """Providing both conninfo and db_path raises ValueError.""" + from orcapod.contexts import enable_file_hash_caching + + with pytest.raises(ValueError, match="not both"): + enable_file_hash_caching( + db_path=tmp_path / "x.db", + conninfo="postgresql://unused", + ) + + +# --------------------------------------------------------------------------- +# SqliteHashCacher schema version detection +# --------------------------------------------------------------------------- + + +def _make_v0_db(path: Path) -> None: + """Create a V0 SQLite cache database (no cached_at column).""" + import sqlite3 + + path.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(path) as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute( + """ + CREATE TABLE file_hash_cache ( + path TEXT NOT NULL, + mtime_ns INTEGER NOT NULL, + size INTEGER NOT NULL, + hash BLOB NOT NULL, + PRIMARY KEY (path, mtime_ns, size) + ) WITHOUT ROWID + """ + ) + # Leave user_version at 0 (default). + conn.commit() + + +class TestSqliteSchemaVersion: + def test_fresh_db_is_stamped_v1(self, tmp_path): + """A freshly created SQLite cache database gets user_version = 1.""" + import sqlite3 + from orcapod.hashing.hash_cachers import SqliteHashCacher + + db = tmp_path / "cache.db" + SqliteHashCacher(db) + + with sqlite3.connect(db) as conn: + version = conn.execute("PRAGMA user_version").fetchone()[0] + assert version == 1 + + def test_v0_db_raises_with_migration_hint(self, tmp_path): + """Opening a V0 database (no cached_at) raises ValueError with migration command.""" + from orcapod.hashing.hash_cachers import SqliteHashCacher + + db = tmp_path / "old_cache.db" + _make_v0_db(db) + + with pytest.raises(ValueError, match="migrate_hash_cache"): + SqliteHashCacher(db) + + def test_v0_db_error_mentions_db_path(self, tmp_path): + """The migration error message includes the database path.""" + from orcapod.hashing.hash_cachers import SqliteHashCacher + + db = tmp_path / "old_cache.db" + _make_v0_db(db) + + with pytest.raises(ValueError, match=str(db)): + SqliteHashCacher(db) + + def test_existing_v1_db_reopens_without_error(self, tmp_path): + """Opening a V1 database a second time succeeds.""" + from orcapod.hashing.hash_cachers import SqliteHashCacher + + db = tmp_path / "cache.db" + SqliteHashCacher(db).close() + # Should not raise. + SqliteHashCacher(db).close() + + def test_manually_bumped_version_with_missing_column_raises(self, tmp_path): + """A DB with user_version=1 but no cached_at column raises ValueError. + + Guards against manually-corrupted databases where user_version was + bumped without actually adding the cached_at column. + """ + import sqlite3 + from orcapod.hashing.hash_cachers import SqliteHashCacher + + db = tmp_path / "corrupt.db" + with sqlite3.connect(db) as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute( + """ + CREATE TABLE file_hash_cache ( + path TEXT NOT NULL, + mtime_ns INTEGER NOT NULL, + size INTEGER NOT NULL, + hash BLOB NOT NULL, + PRIMARY KEY (path, mtime_ns, size) + ) WITHOUT ROWID + """ + ) + # Manually stamp user_version = 1 without adding cached_at. + conn.execute("PRAGMA user_version = 1") + conn.commit() + + with pytest.raises(ValueError, match="cached_at"): + SqliteHashCacher(db) + + +# --------------------------------------------------------------------------- +# migrate_hash_cache script +# --------------------------------------------------------------------------- + + +class TestMigrateHashCache: + def test_migrates_v0_to_v1(self, tmp_path): + """migrate_sqlite_hash_cache adds cached_at and stamps version 1.""" + import sqlite3 + from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache + + db = tmp_path / "old.db" + _make_v0_db(db) + migrate_sqlite_hash_cache(db) + + with sqlite3.connect(db) as conn: + version = conn.execute("PRAGMA user_version").fetchone()[0] + columns = { + row[1] for row in conn.execute("PRAGMA table_info(file_hash_cache)") + } + assert version == 1 + assert "cached_at" in columns + + def test_migration_idempotent(self, tmp_path, capsys): + """Running migration twice on a V1 database prints a message and exits cleanly.""" + from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache + from orcapod.hashing.hash_cachers import SqliteHashCacher + + db = tmp_path / "cache.db" + SqliteHashCacher(db).close() # creates V1 DB + migrate_sqlite_hash_cache(db) + + out = capsys.readouterr().out + assert "nothing to migrate" in out + + def test_missing_db_raises_file_not_found(self, tmp_path): + """migrate_sqlite_hash_cache raises FileNotFoundError for a missing path.""" + from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache + + with pytest.raises(FileNotFoundError): + migrate_sqlite_hash_cache(tmp_path / "nonexistent.db") + + def test_non_cache_db_raises_value_error(self, tmp_path): + """migrate_sqlite_hash_cache raises ValueError if the expected table is absent.""" + import sqlite3 + from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache + + db = tmp_path / "other.db" + with sqlite3.connect(db) as conn: + conn.execute("CREATE TABLE unrelated (id INTEGER PRIMARY KEY)") + with pytest.raises(ValueError, match="file_hash_cache"): + migrate_sqlite_hash_cache(db) + + def test_preserved_rows_after_migration(self, tmp_path): + """Rows written before migration are readable after migration.""" + import sqlite3 + from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache + + db = tmp_path / "old.db" + _make_v0_db(db) + with sqlite3.connect(db) as conn: + conn.execute( + "INSERT INTO file_hash_cache (path, mtime_ns, size, hash) " + "VALUES (?, ?, ?, ?)", + ("/a/b.txt", 1000, 100, b"sha256:\xab" * 4), + ) + conn.commit() + + migrate_sqlite_hash_cache(db) + + with sqlite3.connect(db) as conn: + row = conn.execute( + "SELECT path, cached_at FROM file_hash_cache WHERE path=?", + ("/a/b.txt",), + ).fetchone() + assert row is not None + assert row[0] == "/a/b.txt" + assert row[1] == 0 # default for migrated rows + + def test_cached_at_column_present_but_unstamped(self, tmp_path, capsys): + """A DB that has cached_at but user_version=0 just gets stamped to V1.""" + import sqlite3 + from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache + + db = tmp_path / "unstamped.db" + # Build a DB that has the cached_at column but user_version still = 0 + with sqlite3.connect(db) as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute( + """ + CREATE TABLE file_hash_cache ( + path TEXT NOT NULL, + mtime_ns INTEGER NOT NULL, + size INTEGER NOT NULL, + hash BLOB NOT NULL, + cached_at INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (path, mtime_ns, size) + ) WITHOUT ROWID + """ + ) + # user_version stays at 0 (default) + conn.commit() + + migrate_sqlite_hash_cache(db) + + out = capsys.readouterr().out + assert "stamped schema version" in out + + with sqlite3.connect(db) as conn: + version = conn.execute("PRAGMA user_version").fetchone()[0] + assert version == 1 + + def test_put_after_migration_sets_nonzero_cached_at(self, tmp_path): + """put() on a migrated DB writes the current epoch, not 0. + + The migration adds cached_at with DEFAULT 0 (SQLite ALTER TABLE only + accepts literal defaults). SqliteHashCacher.put() must explicitly + supply cached_at so migrated databases get real timestamps on new rows. + """ + import sqlite3 + import time + from orcapod.hashing.hash_cachers import SqliteHashCacher + from orcapod.hashing.migrate_hash_cache import migrate_sqlite_hash_cache + + db = tmp_path / "migrated.db" + _make_v0_db(db) + migrate_sqlite_hash_cache(db) + + before = int(time.time()) - 1 + cacher = SqliteHashCacher(db) + cacher.put(make_key("/new.txt", mtime_ns=5000, size=300), make_hash(digest=b"\x99" * 32)) + cacher.close() + + with sqlite3.connect(db) as conn: + row = conn.execute( + "SELECT cached_at FROM file_hash_cache WHERE path='/new.txt'" + ).fetchone() + assert row is not None + assert row[0] > before, f"cached_at={row[0]} should be > {before}" + + def test_main_entry_point(self, tmp_path, capsys): + """main() parses argv, runs migration, exits cleanly.""" + import sys + from unittest.mock import patch + from orcapod.hashing.migrate_hash_cache import main + from orcapod.hashing.hash_cachers import SqliteHashCacher + + db = tmp_path / "v1.db" + SqliteHashCacher(db).close() # creates a V1 DB + + # Patch sys.argv to supply the db_path argument + with patch.object(sys, "argv", ["migrate_hash_cache", str(db)]): + main() + + out = capsys.readouterr().out + assert "nothing to migrate" in out diff --git a/tests/test_hashing/test_hash_utils.py b/tests/test_hashing/test_hash_utils.py index 28b5e61e..7de747e0 100644 --- a/tests/test_hashing/test_hash_utils.py +++ b/tests/test_hashing/test_hash_utils.py @@ -67,6 +67,18 @@ def test_typing_union_order_independent(self): class TestGetFunctionSignatureUnionCanonical: + def test_include_defaults_false_strips_defaults(self): + """include_defaults=False removes default values from parameter strings.""" + def fn(x: int = 42, y: str = "hello") -> None: + pass + + sig_with = get_function_signature(fn) + sig_without = get_function_signature(fn, include_defaults=False) + assert "42" in sig_with + assert "42" not in sig_without + assert "x" in sig_without + assert "y" in sig_without + def test_param_union_order_independent(self): """get_function_signature returns the same string for str|Path and Path|str params.""" def foo1(x: str | Path) -> str: @@ -230,3 +242,130 @@ def test_position_after_string_returns_false(self): def test_empty_prefix_returns_false(self): from orcapod.hashing.hash_utils import _is_in_string assert _is_in_string("# comment", 0) is False + + +class TestGetFunctionComponents: + """Tests for get_function_components() in hash_utils.""" + + def _get(self, *args, **kwargs): + from orcapod.hashing.hash_utils import get_function_components + return get_function_components(*args, **kwargs) + + def test_default_includes_name_module_source_annotations_code(self): + def my_func(x: int, y: str = "hi") -> bool: + return True + + components = self._get(my_func) + combined = " ".join(components) + assert any(c.startswith("name:") for c in components) + assert any(c.startswith("module:") for c in components) + assert any(c.startswith("source:") for c in components) + assert any(c.startswith("code_properties:") for c in components) + + def test_name_override(self): + def original_name(): + pass + + components = self._get(original_name, name_override="custom") + assert any(c == "name:custom" for c in components) + + def test_include_name_false(self): + def my_func(): + pass + + components = self._get(my_func, include_name=False) + assert not any(c.startswith("name:") for c in components) + + def test_include_module_false(self): + def my_func(): + pass + + components = self._get(my_func, include_module=False) + assert not any(c.startswith("module:") for c in components) + + def test_include_code_properties_false(self): + def my_func(x: int): + return x + + components = self._get(my_func, include_code_properties=False) + assert not any(c.startswith("code_properties:") for c in components) + + def test_include_annotations_false(self): + def my_func(x: int) -> str: + return str(x) + + components = self._get(my_func, include_annotations=False) + assert not any(c.startswith("annotations:") for c in components) + + def test_annotations_component_present_when_annotations_exist(self): + def my_func(x: int) -> str: + return str(x) + + components = self._get(my_func) + assert any(c.startswith("annotations:") for c in components) + + def test_preserve_whitespace_false(self): + def my_func(): + """Docstring.""" + pass + + components_ws = self._get(my_func, preserve_whitespace=True) + components_no_ws = self._get(my_func, preserve_whitespace=False) + # Both should have source; no_ws version should be cleandoc'd + source_ws = next(c for c in components_ws if c.startswith("source:")) + source_no_ws = next(c for c in components_no_ws if c.startswith("source:")) + # cleandoc strips leading indentation + assert len(source_no_ws) <= len(source_ws) + + def test_include_declaration_false(self): + def my_func(x: int) -> str: + return str(x) + + components = self._get(my_func, include_declaration=False) + source = next(c for c in components if c.startswith("source:")) + # The 'def my_func' line should be removed + assert "def my_func" not in source + + def test_include_comments_false(self): + # We can't write inline comments without them being included in the test + # source — use exec to create a function with a comment in its source. + import textwrap + code = textwrap.dedent(""" + def has_comment(x): + # this is a comment + return x + """) + ns = {} + exec(compile(code, "", "exec"), ns) + _ = ns["has_comment"] + + # NOTE: For dynamically-defined functions, inspect.getsource raises IOError, + # so include_comments only affects functions with real source. + # We test it on a real function instead. + def real_func(x): # inline comment + return x + + components_with = self._get(real_func, include_comments=True) + components_without = self._get(real_func, include_comments=False) + source_with = next(c for c in components_with if c.startswith("source:")) + source_without = next(c for c in components_without if c.startswith("source:")) + # The inline comment should be stripped + assert "inline comment" not in source_without + assert "inline comment" in source_with + + def test_no_source_falls_back_to_name_and_signature(self): + """For builtins (no source), falls back to name + signature components.""" + # Builtins don't have __code__, so disable include_code_properties to + # avoid AttributeError; focus on verifying the IOError/TypeError fallback path. + components = self._get(len, include_code_properties=False, include_annotations=False) + combined = " ".join(components) + # Should not have a source: component for a builtin + # but should have a name: (either from include_name or fallback) + assert "name:" in combined + + def test_return_is_a_list(self): + def my_func(): + pass + + result = self._get(my_func) + assert isinstance(result, list) diff --git a/tests/test_hashing/test_postgres_hash_cacher.py b/tests/test_hashing/test_postgres_hash_cacher.py new file mode 100644 index 00000000..53256e95 --- /dev/null +++ b/tests/test_hashing/test_postgres_hash_cacher.py @@ -0,0 +1,463 @@ +"""Tests for PostgresHashCacher using a real Postgres via testcontainers. + +These tests are marked ``@pytest.mark.postgres`` and are **skipped by default**. +To run them, pass ``--postgres`` and ensure Docker is available:: + + uv run pytest tests/test_hashing/test_postgres_hash_cacher.py --postgres + +The ``pg_conninfo`` fixture uses ``testcontainers[postgres]`` to spin up a +``postgres:16`` container automatically; no manual Postgres setup is needed. +Tests in ``TestPostgresReprRedaction`` do **not** carry the postgres marker and +run unconditionally because they test the pure ``_redact_conninfo()`` helper. +""" + +from __future__ import annotations + +import threading + +import pytest +from upath import UPath + +psycopg = pytest.importorskip("psycopg") + +from testcontainers.postgres import PostgresContainer # noqa: E402 + +from orcapod.hashing.file_hashers import FileHashKey # noqa: E402 +from orcapod.hashing.postgres_hash_cacher import PostgresHashCacher # noqa: E402 +from orcapod.types import ContentHash # noqa: E402 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_key( + path: str = "/a/b.txt", mtime_ns: int = 1000, size: int = 100 +) -> FileHashKey: + return FileHashKey(path=UPath(path), mtime_ns=mtime_ns, size=size) + + +def make_hash(digest: bytes = b"\xab" * 32) -> ContentHash: + return ContentHash(method="sha256", digest=digest) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def pg_conninfo(): + """Start one Postgres container for the entire module; yield a conninfo string.""" + with PostgresContainer("postgres:16") as pg: + yield ( + f"host={pg.get_container_host_ip()} " + f"port={pg.get_exposed_port(5432)} " + f"dbname={pg.dbname} " + f"user={pg.username} " + f"password={pg.password}" + ) + + +@pytest.fixture() +def cacher(pg_conninfo): + """Fresh PostgresHashCacher with a cleared table for each test.""" + c = PostgresHashCacher(pg_conninfo) + c.clear() + yield c + c.close() + + +# --------------------------------------------------------------------------- +# Basic operations +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherBasic: + def test_miss_returns_none(self, cacher): + assert cacher.get(make_key()) is None + + def test_put_then_get_returns_hit(self, cacher): + key = make_key() + value = make_hash() + cacher.put(key, value) + result = cacher.get(key) + assert result is not None + assert result.method == value.method + assert result.digest == value.digest + + def test_different_path_is_miss(self, cacher): + cacher.put(make_key("/a/b.txt"), make_hash(b"\xaa" * 32)) + assert cacher.get(make_key("/a/c.txt")) is None + + def test_different_mtime_ns_is_miss(self, cacher): + cacher.put(make_key(mtime_ns=1000), make_hash(b"\xaa" * 32)) + assert cacher.get(make_key(mtime_ns=2000)) is None + + def test_different_size_is_miss(self, cacher): + cacher.put(make_key(size=100), make_hash(b"\xaa" * 32)) + assert cacher.get(make_key(size=200)) is None + + def test_clear_empties_table(self, cacher): + key = make_key() + cacher.put(key, make_hash()) + cacher.clear() + assert cacher.get(key) is None + + def test_persistence_across_instances(self, pg_conninfo): + key = make_key("/persist.txt", mtime_ns=9999, size=512) + value = make_hash(b"\xcc" * 32) + + cacher1 = PostgresHashCacher(pg_conninfo) + cacher1.clear() + cacher1.put(key, value) + cacher1.close() + + cacher2 = PostgresHashCacher(pg_conninfo) + result = cacher2.get(key) + cacher2.close() + + assert result is not None + assert result.method == value.method + assert result.digest == value.digest + + def test_context_manager_closes_connection(self, pg_conninfo): + key = make_key("/ctx.txt", mtime_ns=7777, size=64) + value = make_hash(b"\xdd" * 32) + + with PostgresHashCacher(pg_conninfo) as c: + c.clear() + c.put(key, value) + + # Verify persistence (connection was closed but data committed) + cacher2 = PostgresHashCacher(pg_conninfo) + result = cacher2.get(key) + cacher2.close() + assert result is not None + assert result.digest == value.digest + + +# --------------------------------------------------------------------------- +# Write guards +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherReadOnly: + def test_put_is_noop_when_read_only(self, pg_conninfo): + key = make_key("/ro.txt") + value = make_hash() + + cacher = PostgresHashCacher(pg_conninfo, read_only=True) + cacher.clear() + cacher.put(key, value) + assert cacher.get(key) is None + cacher.close() + + def test_read_only_can_read_preexisting(self, pg_conninfo): + key = make_key("/ro_pre.txt", mtime_ns=5555, size=256) + value = make_hash(b"\xee" * 32) + + writable = PostgresHashCacher(pg_conninfo) + writable.clear() + writable.put(key, value) + writable.close() + + read_only = PostgresHashCacher(pg_conninfo, read_only=True) + result = read_only.get(key) + read_only.close() + + assert result is not None + assert result.digest == value.digest + + def test_default_is_not_read_only(self, cacher): + key = make_key() + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + + +@pytest.mark.postgres +class TestPostgresHashCacherThreshold: + def test_small_file_not_cached(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=100) + cacher.clear() + key = make_key(size=50) + cacher.put(key, make_hash()) + assert cacher.get(key) is None + cacher.close() + + def test_file_at_threshold_is_cached(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=100) + cacher.clear() + key = make_key(size=100) + value = make_hash() + cacher.put(key, value) + result = cacher.get(key) + cacher.close() + assert result is not None + assert result.digest == value.digest + + def test_large_file_is_cached(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=100) + cacher.clear() + key = make_key(size=200) + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + cacher.close() + + def test_none_threshold_caches_all(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=None) + cacher.clear() + key = make_key(size=1) + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + cacher.close() + + def test_zero_threshold_caches_all(self, pg_conninfo): + cacher = PostgresHashCacher(pg_conninfo, min_cache_size_bytes=0) + cacher.clear() + key = make_key(size=1) + value = make_hash() + cacher.put(key, value) + assert cacher.get(key) is not None + cacher.close() + + +# --------------------------------------------------------------------------- +# Concurrency +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherConcurrency: + def test_concurrent_insert_no_error_one_row(self, pg_conninfo): + """Two threads inserting the same key simultaneously → no error, one row.""" + key = make_key("/concurrent.txt", mtime_ns=12345, size=999) + value = make_hash(b"\xff" * 32) + + # Pre-clear using a dedicated cacher + setup = PostgresHashCacher(pg_conninfo) + setup.clear() + setup.close() + + errors: list[Exception] = [] + + def insert() -> None: + try: + c = PostgresHashCacher(pg_conninfo) + c.put(key, value) + c.close() + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=insert) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [], f"Concurrent inserts raised errors: {errors}" + + # Verify exactly one row was written + with psycopg.connect(pg_conninfo) as conn: + count = conn.execute( + "SELECT COUNT(*) FROM file_hash_cache " + "WHERE path=%s AND mtime_ns=%s AND size=%s", + (str(key.path), key.mtime_ns, key.size), + ).fetchone()[0] + assert count == 1 + + +# --------------------------------------------------------------------------- +# Parity with SqliteHashCacher +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherParity: + def test_parity_with_sqlite(self, pg_conninfo, tmp_path): + """Same key/value → equivalent ContentHash from both backends.""" + from orcapod.hashing.hash_cachers import SqliteHashCacher + + key = make_key("/parity.txt", mtime_ns=111, size=222) + value = make_hash(b"\x12" * 32) + + pg_cacher = PostgresHashCacher(pg_conninfo) + pg_cacher.clear() + pg_cacher.put(key, value) + pg_result = pg_cacher.get(key) + pg_cacher.close() + + sq_cacher = SqliteHashCacher(tmp_path / "parity.db") + sq_cacher.put(key, value) + sq_result = sq_cacher.get(key) + + assert pg_result is not None + assert sq_result is not None + assert pg_result.method == sq_result.method + assert pg_result.digest == sq_result.digest + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresHashCacherValidation: + def test_negative_min_cache_size_bytes_raises(self, pg_conninfo): + with pytest.raises(ValueError, match="non-negative"): + PostgresHashCacher(pg_conninfo, min_cache_size_bytes=-1) + + def test_repr_includes_conninfo_and_guards(self, pg_conninfo): + cacher = PostgresHashCacher( + pg_conninfo, read_only=True, min_cache_size_bytes=1024 + ) + r = repr(cacher) + assert "PostgresHashCacher" in r + assert "read_only=True" in r + assert "min_cache_size_bytes=1024" in r + cacher.close() + + +# --------------------------------------------------------------------------- +# enable_file_hash_caching(conninfo=...) integration +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def restore_default_file_handler(): + """Restore the default FileHandler and DirectoryHandler after each test.""" + from orcapod.contexts import get_default_context + from orcapod.extension_types.directory_type import Directory + from orcapod.extension_types.file_type import File + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + original_file_handler = registry.get_handler_for_type(File) + original_dir_handler = registry.get_handler_for_type(Directory) + yield + registry.register(File, original_file_handler) + registry.register(Directory, original_dir_handler) + + +@pytest.mark.postgres +class TestEnableFileHashCachingPostgres: + def test_conninfo_activates_postgres_cacher( + self, restore_default_file_handler, pg_conninfo + ): + """enable_file_hash_caching(conninfo=...) wires up PostgresHashCacher.""" + from orcapod.contexts import enable_file_hash_caching, get_default_context + from orcapod.extension_types.file_type import File + + enable_file_hash_caching(conninfo=pg_conninfo) + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + handler = registry.get_handler_for_type(File) + assert isinstance(handler.file_hasher.cacher, PostgresHashCacher) + + def test_conninfo_passes_read_only( + self, restore_default_file_handler, pg_conninfo + ): + """read_only=True is forwarded to PostgresHashCacher.""" + from orcapod.contexts import enable_file_hash_caching, get_default_context + from orcapod.extension_types.file_type import File + + enable_file_hash_caching(conninfo=pg_conninfo, read_only=True) + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + handler = registry.get_handler_for_type(File) + assert handler.file_hasher.cacher._read_only is True + + def test_conninfo_passes_min_cache_size_bytes( + self, restore_default_file_handler, pg_conninfo + ): + """min_cache_size_bytes is forwarded to PostgresHashCacher.""" + from orcapod.contexts import enable_file_hash_caching, get_default_context + from orcapod.extension_types.file_type import File + + enable_file_hash_caching(conninfo=pg_conninfo, min_cache_size_bytes=2048) + + context = get_default_context() + registry = context.semantic_hasher.type_handler_registry + handler = registry.get_handler_for_type(File) + assert handler.file_hasher.cacher._min_cache_size_bytes == 2048 + + +# --------------------------------------------------------------------------- +# Schema version and repr +# --------------------------------------------------------------------------- + + +@pytest.mark.postgres +class TestPostgresSchemaVersion: + def test_meta_table_contains_schema_version(self, pg_conninfo): + """After construction, file_hash_cache_meta stores schema_version = 1.""" + PostgresHashCacher(pg_conninfo).close() + + with psycopg.connect(pg_conninfo) as conn: + row = conn.execute( + "SELECT value FROM file_hash_cache_meta WHERE key = 'schema_version'" + ).fetchone() + assert row is not None + assert row[0] == "1" + + def test_old_schema_raises_with_migration_hint(self, pg_conninfo): + """A pre-existing table missing cached_at raises ValueError with DDL hint.""" + # Drop and re-create without cached_at to simulate old schema. + with psycopg.connect(pg_conninfo) as conn: + conn.execute("DROP TABLE IF EXISTS file_hash_cache CASCADE") + conn.execute("DROP TABLE IF EXISTS file_hash_cache_meta CASCADE") + conn.execute( + """ + CREATE TABLE file_hash_cache ( + path TEXT NOT NULL, + mtime_ns BIGINT NOT NULL, + size BIGINT NOT NULL, + hash BYTEA NOT NULL, + PRIMARY KEY (path, mtime_ns, size) + ) + """ + ) + + with pytest.raises(ValueError, match="cached_at"): + PostgresHashCacher(pg_conninfo) + + # Restore clean state for subsequent tests. + with psycopg.connect(pg_conninfo) as conn: + conn.execute("DROP TABLE IF EXISTS file_hash_cache CASCADE") + + +class TestPostgresReprRedaction: + def test_repr_redacts_url_password(self): + """__repr__ replaces the password in a URL-form conninfo with ***.""" + from orcapod.hashing.postgres_hash_cacher import _redact_conninfo + + raw = "postgresql://alice:s3cr3t@db.example.com:5432/orcapod" + redacted = _redact_conninfo(raw) + assert "s3cr3t" not in redacted + assert "alice" in redacted + assert "***" in redacted + + def test_repr_redacts_keyword_password(self): + """__repr__ replaces the password in a keyword-form conninfo with ***.""" + from orcapod.hashing.postgres_hash_cacher import _redact_conninfo + + raw = "host=db.example.com dbname=orcapod user=alice password=s3cr3t" + redacted = _redact_conninfo(raw) + assert "s3cr3t" not in redacted + assert "***" in redacted + + def test_repr_no_password_unchanged(self): + """__repr__ does not alter a conninfo string that has no password.""" + from orcapod.hashing.postgres_hash_cacher import _redact_conninfo + + raw = "host=localhost dbname=orcapod user=alice" + assert _redact_conninfo(raw) == raw diff --git a/uv.lock b/uv.lock index 7994a779..f5114413 100644 --- a/uv.lock +++ b/uv.lock @@ -2431,6 +2431,7 @@ dev = [ { name = "redis", specifier = ">=6.2.0" }, { name = "ruff", specifier = ">=0.14.4" }, { name = "testcontainers", extras = ["minio"], specifier = ">=4.0.0" }, + { name = "testcontainers", extras = ["postgres"], specifier = ">=4.0.0" }, { name = "tqdm", specifier = ">=4.67.1" }, ]