ditto.backends¶

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:

66
67
68
69
70
71
72
73

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
    self._local = isinstance(fs, LocalFileSystem)

 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110

def __setitem__(self, key: str, value: bytes) -> None:
    """Store `value` under `key`, replacing any previous value whole.

    On the local filesystem, opening the destination for writing would
    truncate it before any new bytes arrive, so the value is written
    atomically instead: a failed write leaves the previous value intact.
    Object stores (S3, GCS, Azure) already replace an object whole.
    """
    p = self._full_path(key)
    parent = posixpath.dirname(p)
    self._fs.makedirs(parent, exist_ok=True)
    if self._local:
        write_atomically(Path(p), value)
        return
    with self._fs.open(p, "wb") as f:
        f.write(value)  # type: ignore[arg-type]  # fsspec binary mode

136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

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) or (self._local and _is_leftover_temp(path)):
            continue
        yield path[len(prefix) :], int(info.get("size") or 0), _info_mtime(info)  # type: ignore[arg-type]

Bases: MutableMapping[str, BackendFactory]

Backend factories set in code, which take precedence over installed ones.

Behaves like a dict, except that setting a scheme checks the factory and the scheme first. Nothing here loads or depends on an installed backend, so monkeypatch.setitem works for any scheme, including one whose installed backend fails to load or conflicts.

44
45

def __init__(self) -> None:
    self._factories: dict[str, BackendFactory] = {}

50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

def __setitem__(self, scheme: str, factory: BackendFactory) -> None:
    """Handle `scheme` with `factory`, overriding any installed factory.

    Raises
    ------
    TypeError
        If `factory` is not callable.
    DittoBackendConflictError
        If no target URI can reach `scheme`.
    """
    if not callable(factory):
        raise TypeError(f"{type(factory).__name__} is not callable")
    problems = find_scheme_problems([Registration(scheme, LOCAL_REGISTRATION)])
    if problems:
        raise DittoBackendConflictError(
            " ".join(problem.message for problem in problems)
        )
    self._factories[scheme] = factory

Bases: Mapping[str, BackendFactory]

Backend factories keyed by URI scheme, imported only when first looked up.

Each factory has the signature factory(uri: str, **storage_options) -> MutableMapping[str, bytes].

Schemes come from ditto_backends entry-point metadata, which needs no import, so membership tests, iteration and problems never import a plugin. Every registration is checked against the plugin contract when the registry is built. Looking up a scheme a problem affects raises DittoBackendConflictError, so the registry never picks one of the conflicting registrations.

The registry is read-only. Set a scheme on overrides to handle it in code: an override takes precedence over any installed factory for the scheme, and resolves its conflict. Deleting the override reveals the installed factory again. Iteration preserves discovery order, followed by schemes set only on overrides.

Looking up values (including through get, items or values) can load plugins and raise DittoBackendLoadError or DittoBackendConflictError.

Parameters:

108
109
110
111
112
113
114
115
116

def __init__(self, entry_points: Iterable[EntryPoint] | None = None) -> None:
    if entry_points is None:
        entry_points = importlib.metadata.entry_points(group="ditto_backends")
    self._discovered = LazyEntryPoints(entry_points, _check_factory, _load_error)
    self._problems = tuple(find_scheme_problems(self._discovered.registrations))
    self._conflicts = {
        scheme: problem for problem in self._problems for scheme in problem.names
    }
    self._overrides = BackendOverrides()