ditto.backends¶

Bases: MutableMapping

MutableMapping that serialises/deserialises values via save/load callables.

Wraps an underlying MutableMapping[str, bytes] backend. Values written via __setitem__ are passed through save (Any → bytes) before storage; values read via __getitem__ are passed through load (bytes → Any) after retrieval.

Examples:

Partial instances (mapping-only or save/load-only) are combined via |:

store = TransformMapping(mapping=backend) | TransformMapping(
    save=gzip.compress, load=gzip.decompress
)

28
29
30
31
32
33
34
35
36
37

def __init__(
    self,
    *,
    save: Callable[[Any], bytes] | None = None,
    load: Callable[[bytes], Any] | None = None,
    mapping: MutableMapping[str, bytes] | None = None,
) -> None:
    self._save = save
    self._load = load
    self._mapping = mapping

Bases: MutableMapping[str, bytes]

Scopes all operations on a flat MutableMapping under a fixed key prefix.

Use for flat key-value stores (Redis, DynamoDB, etc.) where namespace isolation must come from the key rather than a directory/root path.

The prefix is prepended to keys on write and stripped on read. Keys exposed to callers are always prefix-free.

iter filters by prefix. For backends where full-keyspace iteration is expensive (large shared Redis DB), either: - Use a dedicated DB instance, OR - Configure the inner mapping to do prefix-scoped scanning natively, OR - Raise NotImplementedError on iter to opt out of pruning.

Context management is propagated to the inner store when it implements enter/exit. ditto enters the backend via the session ExitStack (in plugin.py) and closes it in pytest_unconfigure.

32
33
34
35
36

def __init__(self, store: MutableMapping[str, bytes], prefix: str) -> None:
    if not prefix:
        raise ValueError("prefix must be non-empty")
    self._store = store
    self._prefix = prefix

Bases: MutableMapping[str, bytes]

MutableMapping backed by any fsspec AbstractFileSystem.

Uses fs.find() for iteration and direct fs.open / fs.rm / fs.isfile for individual key access — never fs.glob(). This avoids fsspec's fnmatch-based glob expansion, which silently drops keys whose names contain bracket characters (e.g. parametrised test names like test_foo[bar]).

Works with any fsspec backend: local, S3 (s3fs), GCS (gcsfs), Azure (adlfs), in-memory, etc.

Parameters:

52
53
54
55
56
57
58

def __init__(self, fs: AbstractFileSystem, root: str) -> None:
    self._fs = fs
    self._root = root.rstrip("/")
    # .root mirrors the attribute LocalMapping exposed for fsspec compatibility;
    # plugin.py's _get_root() uses it to register this backend's root directory
    # so ghost-detection (Pass 2) doesn't flag its own .ditto/ as unused.
    self.root = self._root

111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

def stat_entries(self) -> Iterator[tuple[str, int, float | None]]:
    """Yield (key, size_bytes, modified) for each stored snapshot.

    Uses one `fs.find(detail=True)` listing, so size and mtime come from a
    single round trip — no per-key reads. `modified` is None when the
    filesystem reports no mtime. Mirrors `__iter__`'s root-prefix stripping.
    """
    if not self._fs.exists(self._root):
        return
    prefix = self._root + "/"
    entries: dict[str, dict[str, object]] = self._fs.find(self._root, detail=True)  # type: ignore[assignment]
    for path, info in entries.items():
        if not path.startswith(prefix):
            continue
        yield path[len(prefix) :], int(info.get("size") or 0), _info_mtime(info)  # type: ignore[arg-type]