Support consumer-selected bundle store paths

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## 1.17.0
 
+- Add a `store_path(name, version)` callback to `VersionedDatasetRegistry` for
+  exact consumer-chosen managed-store locations. It is mutually exclusive with
+  root configuration, does not run at construction, and preserves immutable
+  generations and foreign-directory protection (#93).
+
 - Add `verify_files=False` bundle inspection and cache-hit checks that validate
   ownership, source identity, required file types, readability and recorded
   sizes without reading payloads. Fast results never claim hash verification.

datacache/bundles.py

@@ -367,17 +367,22 @@ class VersionedDatasetRegistry:
     Each dataset is {default_version, versions: {version: {asset: metadata}}}.
     The hitlist shape {filename, urls: {version: url}, default_version} is also
     accepted with verified=False. cache_root is a path; cache_dir optionally
-    accepts hitlist's zero-argument root callable. Construction never writes.
+    accepts hitlist's zero-argument root callable. Alternatively, store_path is
+    a two-argument (name, version) callback selecting an exact managed store.
+    Construction never writes or invokes either callback.
     """
 
-    def __init__(self, datasets, *, cache_root=None, cache_dir=None, verified=True):
-        if (cache_root is None) == (cache_dir is None):
-            raise ValueError('provide exactly one of cache_root or cache_dir')
+    def __init__(self, datasets, *, cache_root=None, cache_dir=None, store_path=None, verified=True):
+        if sum(value is not None for value in (cache_root, cache_dir, store_path)) != 1:
+            raise ValueError('provide exactly one of cache_root, cache_dir or store_path')
         if not isinstance(verified, bool):
             raise ValueError('verified must be a boolean')
         self._root = cache_dir if cache_dir is not None else lambda: cache_root
-        if not callable(self._root):
+        if cache_dir is not None and not callable(cache_dir):
             raise ValueError('cache_dir must be callable')
+        if store_path is not None and not callable(store_path):
+            raise ValueError('store_path must be callable')
+        self._store_path = store_path
         self.verified = verified
         self._datasets = {}
         for name, spec in datasets.items():
@@ -405,6 +410,8 @@ def resolve_version(self, name, version=None):
     def bundle_path(self, name, version=None):
         """Resolve the store path without checking it or creating directories."""
         version = self.resolve_version(name, version)
+        if self._store_path is not None:
+            return Path(self._store_path(name, version))
         parent = Path(self._root()) / name
         if path_present(parent):
             _directory(parent)

docs/api.md

@@ -1243,12 +1243,16 @@ generation. Paths remain usable across refreshes until explicitly removed.
 ## VersionedDatasetRegistry
 
 ```text
-VersionedDatasetRegistry(datasets, *, cache_root=None, cache_dir=None, verified=True)
+VersionedDatasetRegistry(datasets, *, cache_root=None, cache_dir=None, store_path=None, verified=True)
 ```
 
-Select exactly one root path or a zero-argument `cache_dir` callable. Each dataset
+Select exactly one root path, zero-argument `cache_dir` callable, or
+`store_path(name, version)` callback selecting the exact managed store path.
+Root strategies keep `<root>/<name>/<version>`; the exact-path callback owns
+the consumer layout but not the internal generation layout. Each dataset
 specifies a `default_version` and `versions`, mapping concrete versions to asset
-mappings. Construction validates metadata and performs no writes or networking.
+mappings. Construction validates metadata and performs no writes or networking,
+and does not invoke callbacks. Path callbacks should be pure path computations.
 The [bundle guide](bundles.md) includes an example and downstream migration notes.
 
 | Method | Result |

docs/bundles.md

@@ -73,6 +73,33 @@ ownership marker, even with `force=True`. Store generated indices and other deri
 outputs under a separate application-owned directory. Reinstalling sources does
 not visit those outputs.
 
+## Consumer-selected store paths
+
+Select exactly one of `cache_root`, a zero-argument `cache_dir` root callable,
+or an exact `store_path(name, version)` callback. Root strategies retain
+`<root>/<name>/<version>`; the callback can keep an application's semantic
+layout while placing new managed sources beside its existing derived indexes:
+
+```python
+registry = VersionedDatasetRegistry(
+    datasets,
+    store_path=lambda name, version: (
+        application_root / "GRCh38" / ("ensembl-" + version) / "sources" / name),
+)
+store = registry.bundle_path("reference")
+```
+
+The callback receives a validated concrete version, including when the caller
+omits it and selects the pinned default. Construction never calls it;
+`bundle_path` calls it without filesystem inspection or mutation. Callbacks
+should only compute paths. Inspection, installation, recovery and refresh all
+use the chosen store, while DataCache still owns its immutable generations.
+
+Do not point this callback at a populated legacy data/index directory: even
+`force=True` cannot adopt a foreign directory. Keep those old caches readable
+and deliberately install into a new managed source subdirectory instead. Model,
+biological naming and migration policy remain the application's responsibility.
+
 ## Status and trust
 
 `BundleInspection.status` is `available`, `missing`, `invalid`, `inaccessible`, or

tests/test_bundle_store_paths.py

@@ -0,0 +1,161 @@
+"""Consumer-selected paths use the ordinary owned bundle transaction."""
+
+from hashlib import sha256
+import os
+from pathlib import Path
+
+import pytest
+
+from datacache import FileValidationError, VersionedDatasetRegistry
+from datacache import bundles, download
+
+pytestmark = pytest.mark.skipif(os.name != 'posix', reason='POSIX bundle installation')
+
+
+@pytest.fixture
+def datasets(tmp_path):
+    source = tmp_path / 'source.fa'
+    payload = b'>reference\nACGT\n'
+    source.write_bytes(payload)
+    assets = {'records.fa': dict(url=source.as_uri(), sha256=sha256(payload).hexdigest(), size=len(payload))}
+    return {'reference': dict(default_version='110', versions={'110': assets, '109': assets})}
+
+
+def test_custom_path_resolution_is_pure_and_uses_concrete_versions(tmp_path, datasets, monkeypatch):
+    calls = []
+
+    def store_path(name, version):
+        calls.append((name, version))
+        return str(tmp_path / 'GRCh38' / ('ensembl-' + version) / 'sources' / name)
+
+    registry = VersionedDatasetRegistry(datasets, store_path=store_path)
+    assert calls == []
+
+    def forbidden(*args, **kwargs):
+        raise AssertionError('path resolution inspected or mutated the filesystem')
+
+    monkeypatch.setattr(bundles, 'path_present', forbidden)
+    monkeypatch.setattr(Path, 'mkdir', forbidden)
+    assert registry.bundle_path('reference') == tmp_path / 'GRCh38/ensembl-110/sources/reference'
+    assert registry.bundle_path('reference', '109') == tmp_path / 'GRCh38/ensembl-109/sources/reference'
+    assert calls == [('reference', '110'), ('reference', '109')]
+    with pytest.raises(ValueError):
+        registry.bundle_path('reference', 'latest')
+    assert len(calls) == 2
+    assert not (tmp_path / 'GRCh38').exists()
+
+
+def test_custom_layout_install_versions_refresh_and_read_only_reuse(tmp_path, datasets, monkeypatch):
+    root = tmp_path / 'application'
+    index = root / 'GRCh38' / 'ensembl-110' / 'index.sqlite'
+    index.parent.mkdir(parents=True)
+    index.write_bytes(b'application-owned derived index')
+    registry = VersionedDatasetRegistry(
+        datasets, store_path=lambda name, version: root / 'GRCh38' / ('ensembl-' + version) / 'sources' / name)
+    assert registry.inspect('reference').status == 'missing'
+    assert not registry.bundle_path('reference').exists()
+    current = registry.download('reference')
+    older = registry.download('reference', '109')
+    assert current != older
+    assert registry.local_path('reference') == Path(current['records.fa'])
+    assert registry.local_path('reference', '109') == Path(older['records.fa'])
+    refreshed = registry.download('reference', force=True)
+    assert refreshed != current
+    assert Path(current['records.fa']).read_bytes() == Path(refreshed['records.fa']).read_bytes()
+    assert index.read_bytes() == b'application-owned derived index'
+    store = registry.bundle_path('reference')
+    entries = [store, *store.rglob('*')]
+    modes = {path: path.stat().st_mode & 0o777 for path in entries}
+
+    def forbidden(*args, **kwargs):
+        raise AssertionError('offline reuse attempted a write or acquisition')
+
+    monkeypatch.setattr(download, 'fetch_file', forbidden)
+    monkeypatch.setattr(bundles, 'file_lock', forbidden)
+    monkeypatch.setattr(bundles, 'write_json', forbidden)
+    try:
+        for path in entries:
+            path.chmod(0o555 if path.is_dir() else 0o444)
+        assert registry.download('reference') == refreshed
+        assert registry.local_path('reference') == Path(refreshed['records.fa'])
+        assert registry.is_cached('reference')
+        assert registry.inspect('reference').verified
+    finally:
+        for path, mode in modes.items():
+            path.chmod(mode)
+
+
+def test_custom_layout_recovers_completed_generation_without_network(tmp_path, datasets, monkeypatch):
+    registry = VersionedDatasetRegistry(
+        datasets, store_path=lambda name, version: tmp_path / ('sources-' + version) / name)
+    original = bundles.write_json
+
+    def interrupted(path, value, **kwargs):
+        if Path(path).name == bundles.CURRENT:
+            raise KeyboardInterrupt('pointer publication interrupted')
+        return original(path, value, **kwargs)
+
+    monkeypatch.setattr(bundles, 'write_json', interrupted)
+    with pytest.raises(KeyboardInterrupt):
+        registry.download('reference')
+    state = registry.inspect('reference')
+    assert state.status == 'recovery-required'
+    monkeypatch.setattr(bundles, 'write_json', original)
+
+    def forbidden(*args, **kwargs):
+        raise AssertionError('recovery attempted acquisition')
+
+    monkeypatch.setattr(download, 'fetch_file', forbidden)
+    paths = registry.download('reference')
+    assert registry.inspect('reference').verified
+    assert registry.local_path('reference') == Path(paths['records.fa'])
+
+
+@pytest.mark.parametrize('force', [False, True])
+def test_custom_paths_never_take_over_foreign_directories(tmp_path, datasets, force):
+    foreign = tmp_path / 'foreign'
+    foreign.mkdir()
+    precious = foreign / 'index.sqlite'
+    precious.write_bytes(b'keep me')
+    registry = VersionedDatasetRegistry(datasets, store_path=lambda name, version: foreign)
+    with pytest.raises((FileValidationError, FileNotFoundError)):
+        registry.download('reference', force=force)
+    assert list(foreign.iterdir()) == [precious]
+    assert precious.read_bytes() == b'keep me'
+
+
+def test_custom_paths_reject_symlinked_stores(tmp_path, datasets):
+    foreign = tmp_path / 'foreign'
+    foreign.mkdir()
+    store = tmp_path / 'link'
+    store.symlink_to(foreign, target_is_directory=True)
+    registry = VersionedDatasetRegistry(datasets, store_path=lambda name, version: store)
+    assert registry.inspect('reference').status == 'invalid'
+    with pytest.raises(FileValidationError):
+        registry.download('reference', force=True)
+    assert list(foreign.iterdir()) == []
+
+
+@pytest.mark.parametrize('options', [{}, {'cache_root': '/cache', 'cache_dir': lambda: '/cache'},
+    {'cache_root': '/cache', 'store_path': lambda name, version: '/store'},
+    {'cache_dir': lambda: '/cache', 'store_path': lambda name, version: '/store'},
+    {'cache_root': '/cache', 'cache_dir': lambda: '/cache', 'store_path': lambda name, version: '/store'}])
+def test_destination_strategies_are_mutually_exclusive(datasets, options):
+    with pytest.raises(ValueError, match='exactly one'):
+        VersionedDatasetRegistry(datasets, **options)
+
+
+@pytest.mark.parametrize('key', ['cache_dir', 'store_path'])
+def test_path_callbacks_must_be_callable(datasets, key):
+    with pytest.raises(ValueError, match=key + ' must be callable'):
+        VersionedDatasetRegistry(datasets, **{key: '/not-a-callback'})
+
+
+def test_existing_root_strategies_remain_compatible(tmp_path, datasets):
+    explicit = VersionedDatasetRegistry(datasets, cache_root=tmp_path / 'root')
+    selected = [tmp_path / 'root']
+    dynamic = VersionedDatasetRegistry(datasets, cache_dir=lambda: selected[0])
+    assert explicit.bundle_path('reference') == dynamic.bundle_path('reference')
+    selected[0] = tmp_path / 'different-root'
+    assert dynamic.bundle_path('reference') == tmp_path / 'different-root/reference/110'
+    assert not selected[0].exists()