Registry and specs¶

def backend(self, kind: str | None = None) -> BackendSpec:
    """Get the backend spec of the given kind, or the default (the first)."""
    if kind is None:
        return self.backends[0]
    for b in self.backends:
        if b.kind == kind:
            return b
    available = ", ".join(b.kind for b in self.backends)
    raise ValueError(
        f"{self.label} is not served by the {kind!r} backend. Available: {available}."
    )

def capabilities_for(self, kind: str | None = None) -> frozenset[Cap]:
    """What this dataset can do *through this backend*.

    `self.capabilities` is what the data has. This is what you can actually
    reach, and it is the only one anyone should ask: a dataset object is always
    bound to one backend, so the unqualified set is a claim nobody can use.
    """
    b = self.backend(kind)
    return (
        (self.capabilities | b.extra_capabilities)
        - BACKEND_LIMITS[b.kind]
        - b.missing_capabilities
    )

def backends_with(self, cap: Cap) -> tuple[str, ...]:
    """Which of this dataset's backends can serve `cap`. Possibly none.

    Exists so a CapabilityError can end with "the cave backend does" instead of
    a flat no - the difference between a dead end and a next step.
    """
    return tuple(
        b.kind for b in self.backends if cap in self.capabilities_for(b.kind)
    )

def annotation_source(self, name: str | None = "auto") -> AnnotationSource | None:
    """Pick an annotation source by name.

    ``"auto"`` picks the first *public* source; ``None`` disables annotations.
    """
    if name is None or not self.annotation_sources:
        return None
    if name == "auto":
        for s in self.annotation_sources:
            if s.public:
                return s
        # No public source. That is a gated dataset whose only annotations are
        # lab-internal (aedes: FlyTable). Returning None would make
        # `.annotations.get()` claim "no source configured" - false, there is
        # one; it just needs credentials. Fall back to the first source and let
        # the fetch raise a clear missing-token error if the caller can't read it.
        return self.annotation_sources[0]
    for s in self.annotation_sources:
        if s.name == name:
            return s
    available = ", ".join(s.name for s in self.annotation_sources)
    raise ValueError(
        f"{self.label} has no annotation source {name!r}. Available: {available}."
    )

def evolve(self, **changes) -> DatasetSpec:
    """Return a copy with the given fields replaced."""
    return replace(self, **changes)