stores

Candidate dataclass

Candidate(label: str, value: T)

Bases: Generic[T]

A record together with the label a store shows when asking the user to pick it.

CompositeStore

CompositeStore(
    *,
    default: Store | None = None,
    routes: Mapping[str, Store] | None = None,
)

Bases: StoreBase

Routes each kind to a backend by record name, falling back to a default store.

This is what composition buys over subclassing: a deployment keeping rigs and tasks on a network share but trainer state in Dataverse is a routing table, not a class.

Example

from clabe.stores.dataverse import DataverseStore

store = CompositeStore(
    default=LocalFileStore(root=VR_LIB),
    routes={"trainer_state": DataverseStore()},
)

Parameters:

Name	Type	Description	Default
default	Store | None	Serves any kind not named in routes. If None, unrouted kinds raise.	None
routes	Mapping[str, Store] | None	Maps a :attr:Kind.name to the store serving it.	None

Source code in src/clabe/stores/_base.py

def __init__(self, *, default: Store | None = None, routes: Mapping[str, Store] | None = None) -> None:
    """
    Args:
        default: Serves any kind not named in ``routes``. If ``None``,
            unrouted kinds raise.
        routes: Maps a :attr:`Kind.name` to the store serving it.
    """
    super().__init__()
    self._default = default
    self._routes = dict(routes or {})

scope property

scope: Scope

The scope every call on this store is narrowed by.

resolve

resolve(
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T

Delegates to the routed backend, so its own presentation is used.

Source code in src/clabe/stores/_base.py

def resolve(
    self,
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T:
    """Delegates to the routed backend, so its own presentation is used."""
    _kind = as_kind(kind)
    return self._route(_kind).resolve(_kind, scope=scope, pick_kwargs=pick_kwargs)

list

list(
    kind: KindLike[T], *, scope: Scope | None = None
) -> Sequence[T]

Delegates to the routed backend.

Source code in src/clabe/stores/_base.py

def list(self, kind: KindLike[T], *, scope: Scope | None = None) -> Sequence[T]:
    """Delegates to the routed backend."""
    _kind = as_kind(kind)
    return self._route(_kind).list(_kind, scope=scope)

write

write(
    kind: KindLike[T],
    value: T,
    *,
    scope: Scope | None = None,
) -> None

Delegates to the routed backend.

Source code in src/clabe/stores/_base.py

def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """Delegates to the routed backend."""
    _kind = as_kind(kind)
    self._route(_kind).write(_kind, value, scope=scope)

scoped

scoped(**scope: str) -> CompositeStore

Returns a composite whose every backend is narrowed by the given scope.

Source code in src/clabe/stores/_base.py

def scoped(self, **scope: str) -> "CompositeStore":
    """Returns a composite whose every backend is narrowed by the given scope."""
    return CompositeStore(
        default=self._default.scoped(**scope) if self._default is not None else None,
        routes={name: store.scoped(**scope) for name, store in self._routes.items()},
    )

Kind

Kind(
    model: type[T],
    name: str | None = None,
    *,
    validators: Callable[[T], T]
    | Iterable[Callable[[T], T]]
    | None = None,
)

Bases: Generic[T]

A record name bound to the pydantic model that parses it.

The name routes the record to a backend and determines its storage layout; the model types the return value of every store call made with this kind.

Canonical kinds should be built through :meth:from_rig, :meth:from_task, :meth:from_session or :meth:from_trainer_state, which fix the framework name. The bare constructor derives the name from the model class instead, so Kind(AindVrForagingRig) is named "aind_vr_foraging_rig", not "rig" -- it is for custom records only.

Parameters:

Name	Type	Description	Default
model	type[T]	The pydantic model records of this kind deserialize to.	required
name	str | None	The record name. Defaults to the snake-cased model class name.	None
validators	Callable[[T], T] | Iterable[Callable[[T], T]] | None	One callable, an iterable of callables, or None. Each is applied in order to every record read through a store.	None

Source code in src/clabe/stores/_base.py

def __init__(
    self,
    model: type[T],
    name: str | None = None,
    *,
    validators: Callable[[T], T] | Iterable[Callable[[T], T]] | None = None,
) -> None:
    """
    Args:
        model: The pydantic model records of this kind deserialize to.
        name: The record name. Defaults to the snake-cased model class name.
        validators: One callable, an iterable of callables, or ``None``.
            Each is applied in order to every record read through a store.
    """
    self.model = model
    self.name = name if name is not None else to_snake(model.__name__)
    self.validators: list[Callable[[T], T]] = (
        []
        if validators is None
        else [validators]  # type: ignore[list-item]
        if callable(validators)
        else list(validators)  # type: ignore[arg-type]
    )

adapter cached property

adapter: TypeAdapter[T]

Serializes and parses records of this kind. Also covers models that are not BaseModel.

from_rig staticmethod

from_rig(
    model: type[TRig],
    *,
    validators: Callable[[TRig], TRig]
    | Iterable[Callable[[TRig], TRig]]
    | None = validators,
) -> Kind[TRig]

Builds the canonical "rig" kind for a rig model.

Source code in src/clabe/stores/_base.py

@staticmethod
def from_rig(
    model: type[TRig],
    *,
    validators: Callable[[TRig], TRig] | Iterable[Callable[[TRig], TRig]] | None = _CANONICAL_KINDS[Rig].validators,
) -> "Kind[TRig]":
    """Builds the canonical ``"rig"`` kind for a rig model."""
    return Kind(model, _CANONICAL_KINDS[Rig].name, validators=validators)

from_task staticmethod

from_task(model: type[TTask]) -> Kind[TTask]

Builds the canonical "task" kind for a task model.

Source code in src/clabe/stores/_base.py

@staticmethod
def from_task(model: type[TTask]) -> "Kind[TTask]":
    """Builds the canonical ``"task"`` kind for a task model."""
    return Kind(model, _CANONICAL_KINDS[Task].name)

from_session staticmethod

from_session() -> Kind[Session]

Builds the canonical "session" kind for a session model.

Source code in src/clabe/stores/_base.py

@staticmethod
def from_session() -> "Kind[Session]":
    """Builds the canonical ``"session"`` kind for a session model."""
    return Kind(Session, _CANONICAL_KINDS[Session].name)

from_trainer_state staticmethod

from_trainer_state() -> Kind[TrainerState]

Builds the canonical "trainer_state" kind.

Source code in src/clabe/stores/_base.py

@staticmethod
def from_trainer_state() -> "Kind[TrainerState]":
    """Builds the canonical ``"trainer_state"`` kind."""
    spec = _CANONICAL_KINDS[TrainerState]
    return Kind(TrainerState, spec.name, validators=spec.validators)

Store

Bases: Protocol

Reads, resolves and writes records of a given :class:Kind.

Which kinds a store actually serves is a deployment fact, not a type-level one: an unsupported kind raises when the call is made.

resolve

resolve(
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T

Selects a single record, prompting the user when the choice is ambiguous.

Source code in src/clabe/stores/_base.py

def resolve(
    self,
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: "PickRequestKwargs | None" = None,
) -> T:
    """Selects a single record, prompting the user when the choice is ambiguous."""
    ...

list

list(
    kind: KindLike[T], *, scope: Scope | None = None
) -> Sequence[T]

Returns every record of this kind in scope, without prompting.

Source code in src/clabe/stores/_base.py

def list(self, kind: KindLike[T], *, scope: Scope | None = None) -> Sequence[T]:
    """Returns every record of this kind in scope, without prompting."""
    ...

write

write(
    kind: KindLike[T],
    value: T,
    *,
    scope: Scope | None = None,
) -> None

Persists a record. Whether that overwrites, versions or appends is the store's contract.

Source code in src/clabe/stores/_base.py

def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """Persists a record. Whether that overwrites, versions or appends is the store's contract."""
    ...

scoped

scoped(**scope: str) -> Store

Returns a view of this store narrowed by the given scope.

Source code in src/clabe/stores/_base.py

def scoped(self, **scope: str) -> "Store":
    """Returns a view of this store narrowed by the given scope."""
    ...

StoreBase

StoreBase(*, scope: Scope | None = None)

Bases: ABC

Base for stores, supplying scope narrowing and the default resolution policy.

Subclasses implement :meth:_candidates and :meth:write, and may override :meth:resolve when their backend affords a better presentation than a flat pick list.

Source code in src/clabe/stores/_base.py

def __init__(self, *, scope: Scope | None = None) -> None:
    self._scope: dict[str, str] = dict(scope or {})

scope property

scope: Scope

The scope every call on this store is narrowed by.

scoped

scoped(**scope: str) -> Self

Returns a view sharing this store's data, narrowed by the given scope.

The view is the same concrete store type, so narrowing never loses backend-specific typing.

Source code in src/clabe/stores/_base.py

def scoped(self, **scope: str) -> Self:
    """Returns a view sharing this store's data, narrowed by the given scope.

    The view is the same concrete store type, so narrowing never loses backend-specific typing.
    """
    clone = object.__new__(type(self))
    clone.__dict__.update(self.__dict__)
    clone._scope = {**self._scope, **scope}
    return clone

write abstractmethod

write(
    kind: KindLike[T],
    value: T,
    *,
    scope: Scope | None = None,
) -> None

Persists a record. See :meth:Store.write.

Source code in src/clabe/stores/_base.py

@abc.abstractmethod
def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """Persists a record. See :meth:`Store.write`."""

list

list(
    kind: KindLike[T], *, scope: Scope | None = None
) -> Sequence[T]

Returns every record of this kind in scope, validated but unprompted.

Source code in src/clabe/stores/_base.py

def list(self, kind: KindLike[T], *, scope: Scope | None = None) -> Sequence[T]:
    """Returns every record of this kind in scope, validated but unprompted."""
    _kind = as_kind(kind)
    return [self._validated(_kind, c.value) for c in self._candidates(_kind, self._merge_scope(scope))]

resolve

resolve(
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T

Selects a single record: raises on none, auto-selects a lone candidate, prompts otherwise.

Parameters:

Name	Type	Description	Default
kind	KindLike[T]	The kind of record to resolve.	required
scope	Scope | None	Narrows which records are considered. Layered over the store's own scope.	None
pick_kwargs	PickRequestKwargs | None	Optional overrides for the :class:ui.PickRequest shown to the user. Any key present here takes precedence over resolve's own defaults. options is always derived from the store's candidates and is not overridable.	None

Raises:

Type	Description
LookupError	If no record of this kind exists in scope, or the user declined to pick one.
NoFrontendError	If a choice must be made and no frontend is registered.

Source code in src/clabe/stores/_base.py

def resolve(
    self,
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T:
    """
    Selects a single record: raises on none, auto-selects a lone candidate, prompts otherwise.

    Args:
        kind: The kind of record to resolve.
        scope: Narrows which records are considered. Layered over the store's own scope.
        pick_kwargs: Optional overrides for the :class:`ui.PickRequest` shown to the user.
            Any key present here takes precedence over ``resolve``'s own defaults.
            ``options`` is always derived from the store's candidates and is not overridable.

    Raises:
        LookupError: If no record of this kind exists in scope, or the user declined to pick one.
        ui.NoFrontendError: If a choice must be made and no frontend is registered.
    """
    _kind = as_kind(kind)
    candidates = self._candidates(_kind, self._merge_scope(scope))
    if not candidates:
        raise LookupError(f"No {_kind.name!r} records found in {self}.")
    if len(candidates) == 1:
        ui.notify(f"Found a single {_kind.name}. Using {candidates[0].label}.")
        return self._validated(_kind, candidates[0].value)
    by_label = {c.label: c.value for c in candidates}
    _request_kwargs: dict[str, Any] = {
        "label": f"Choose a {_kind.name} for {_kind.model.__name__}:",
        "allow_none": False,
        "field": _kind.name,
    }
    _request_kwargs.update(pick_kwargs or {})
    picked = ui.prompt_pick(
        ui.PickRequest(
            options=_recent_first(_kind.name, sorted(by_label)),
            **_request_kwargs,
        )
    )
    if picked is None:
        raise LookupError(f"No {_kind.name} was selected.")
    CacheManager.get_instance().add_to_cache(_kind.name, picked)
    return self._validated(_kind, by_label[picked])

DefaultLayout

The config library tree as it exists on the shared drive today.

Rig/<computer_name>/*.json, Task/*.json and Subjects/<subject>/<kind>.json. Tasks read from the subject folder first and fall back to the shared library, which is the ordered scope chain that replaces the nested lookup the picker used to do inline.

Kinds other than rig and task are per-subject state, living beside the subject's task; with no subject in scope they sit at the library root, which is what makes a one-off store pointed at a session directory work.

read

read(kind_name: str, scope: Scope) -> Sequence[str]

Returns the glob patterns for this kind, subject folder before shared library.

Source code in src/clabe/stores/_local.py

def read(self, kind_name: str, scope: Scope) -> Sequence[str]:
    """Returns the glob patterns for this kind, subject folder before shared library."""
    subject = scope.get("subject")
    if kind_name == "rig":
        return [f"{RIG_DIR}/{scope['computer_name']}/*.json"]
    if kind_name == "task":
        library = [f"{TASK_DIR}/*.json"]
        return [f"{SUBJECT_DIR}/{subject}/task.json", *library] if subject else library
    return [self._subject_scoped(kind_name, subject)]

write

write(kind_name: str, scope: Scope) -> str

Returns the single path this kind is written to.

Source code in src/clabe/stores/_local.py

def write(self, kind_name: str, scope: Scope) -> str:
    """Returns the single path this kind is written to."""
    if kind_name == "rig":
        return f"{RIG_DIR}/{scope['computer_name']}/rig.json"
    return self._subject_scoped(kind_name, scope.get("subject"))

Layout

Bases: Protocol

Maps a record kind and scope onto paths within a config library.

read

read(kind_name: str, scope: Scope) -> Sequence[str]

Returns library-relative glob patterns to read from, highest priority first.

Source code in src/clabe/stores/_local.py

def read(self, kind_name: str, scope: Scope) -> Sequence[str]:
    """Returns library-relative glob patterns to read from, highest priority first."""
    ...

write

write(kind_name: str, scope: Scope) -> str

Returns the library-relative path to write to.

Source code in src/clabe/stores/_local.py

def write(self, kind_name: str, scope: Scope) -> str:
    """Returns the library-relative path to write to."""
    ...

LocalFileStore

LocalFileStore(
    root: PathLike | str,
    *,
    layout: Layout | None = None,
    scope: Scope | None = None,
)

Bases: StoreBase

A store over a directory of JSON files, i.e. the config library.

Writes overwrite, and create the directories they need. Files that fail to parse are skipped with a warning rather than failing the whole listing.

Example

store = LocalFileStore(root=r"\\allen\aind\scratch\AindBehavior.db\AindVrForaging")
rig = store.resolve(Kind.from_rig(AindVrForagingRig))

Parameters:

Name	Type	Description	Default
root	PathLike | str	The config library directory.	required
layout	Layout | None	Maps kinds onto paths within root. Defaults to :class:DefaultLayout.	None
scope	Scope | None	Initial scope. computer_name defaults to this machine's.	None

Source code in src/clabe/stores/_local.py

def __init__(self, root: os.PathLike | str, *, layout: Layout | None = None, scope: Scope | None = None) -> None:
    """
    Args:
        root: The config library directory.
        layout: Maps kinds onto paths within ``root``. Defaults to :class:`DefaultLayout`.
        scope: Initial scope. ``computer_name`` defaults to this machine's.
    """
    super().__init__(scope={"computer_name": os.environ.get("COMPUTERNAME", platform.node()), **(scope or {})})
    self._root = Path(root)
    self._layout = layout or DefaultLayout()

scope property

scope: Scope

The scope every call on this store is narrowed by.

root property

root: Path

The config library directory.

scoped

scoped(**scope: str) -> Self

Returns a view sharing this store's data, narrowed by the given scope.

The view is the same concrete store type, so narrowing never loses backend-specific typing.

Source code in src/clabe/stores/_base.py

def scoped(self, **scope: str) -> Self:
    """Returns a view sharing this store's data, narrowed by the given scope.

    The view is the same concrete store type, so narrowing never loses backend-specific typing.
    """
    clone = object.__new__(type(self))
    clone.__dict__.update(self.__dict__)
    clone._scope = {**self._scope, **scope}
    return clone

list

list(
    kind: KindLike[T], *, scope: Scope | None = None
) -> Sequence[T]

Returns every record of this kind in scope, validated but unprompted.

Source code in src/clabe/stores/_base.py

def list(self, kind: KindLike[T], *, scope: Scope | None = None) -> Sequence[T]:
    """Returns every record of this kind in scope, validated but unprompted."""
    _kind = as_kind(kind)
    return [self._validated(_kind, c.value) for c in self._candidates(_kind, self._merge_scope(scope))]

resolve

resolve(
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T

Selects a single record: raises on none, auto-selects a lone candidate, prompts otherwise.

Parameters:

Name	Type	Description	Default
kind	KindLike[T]	The kind of record to resolve.	required
scope	Scope | None	Narrows which records are considered. Layered over the store's own scope.	None
pick_kwargs	PickRequestKwargs | None	Optional overrides for the :class:ui.PickRequest shown to the user. Any key present here takes precedence over resolve's own defaults. options is always derived from the store's candidates and is not overridable.	None

Raises:

Type	Description
LookupError	If no record of this kind exists in scope, or the user declined to pick one.
NoFrontendError	If a choice must be made and no frontend is registered.

Source code in src/clabe/stores/_base.py

def resolve(
    self,
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T:
    """
    Selects a single record: raises on none, auto-selects a lone candidate, prompts otherwise.

    Args:
        kind: The kind of record to resolve.
        scope: Narrows which records are considered. Layered over the store's own scope.
        pick_kwargs: Optional overrides for the :class:`ui.PickRequest` shown to the user.
            Any key present here takes precedence over ``resolve``'s own defaults.
            ``options`` is always derived from the store's candidates and is not overridable.

    Raises:
        LookupError: If no record of this kind exists in scope, or the user declined to pick one.
        ui.NoFrontendError: If a choice must be made and no frontend is registered.
    """
    _kind = as_kind(kind)
    candidates = self._candidates(_kind, self._merge_scope(scope))
    if not candidates:
        raise LookupError(f"No {_kind.name!r} records found in {self}.")
    if len(candidates) == 1:
        ui.notify(f"Found a single {_kind.name}. Using {candidates[0].label}.")
        return self._validated(_kind, candidates[0].value)
    by_label = {c.label: c.value for c in candidates}
    _request_kwargs: dict[str, Any] = {
        "label": f"Choose a {_kind.name} for {_kind.model.__name__}:",
        "allow_none": False,
        "field": _kind.name,
    }
    _request_kwargs.update(pick_kwargs or {})
    picked = ui.prompt_pick(
        ui.PickRequest(
            options=_recent_first(_kind.name, sorted(by_label)),
            **_request_kwargs,
        )
    )
    if picked is None:
        raise LookupError(f"No {_kind.name} was selected.")
    CacheManager.get_instance().add_to_cache(_kind.name, picked)
    return self._validated(_kind, by_label[picked])

write

write(
    kind: KindLike[T],
    value: T,
    *,
    scope: Scope | None = None,
) -> None

Overwrites the file this kind maps to, creating parent directories.

Source code in src/clabe/stores/_local.py

def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """Overwrites the file this kind maps to, creating parent directories."""
    _kind = as_kind(kind)
    path = self._root / self._layout.write(_kind.name, self._merge_scope(scope))
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(_kind.adapter.dump_json(value, indent=2))
    ui.notify(f"Saved {_kind.name} to {path}", ui.MessageLevel.SUCCESS)

MemoryStore

MemoryStore(*, scope: Scope | None = None)

Bases: StoreBase

A store holding records in memory, for tests and for assembling records in-process.

Writes append, as Dataverse does, so several records of one kind can be offered as candidates. A record is visible to a read whose scope agrees with every narrowing the record was written under, so a record written unscoped is library-wide.

Source code in src/clabe/stores/_memory.py

def __init__(self, *, scope: Scope | None = None) -> None:
    super().__init__(scope=scope)
    self._records: list[tuple[str, dict[str, str], Any]] = []

scope property

scope: Scope

The scope every call on this store is narrowed by.

scoped

scoped(**scope: str) -> Self

Returns a view sharing this store's data, narrowed by the given scope.

The view is the same concrete store type, so narrowing never loses backend-specific typing.

Source code in src/clabe/stores/_base.py

def scoped(self, **scope: str) -> Self:
    """Returns a view sharing this store's data, narrowed by the given scope.

    The view is the same concrete store type, so narrowing never loses backend-specific typing.
    """
    clone = object.__new__(type(self))
    clone.__dict__.update(self.__dict__)
    clone._scope = {**self._scope, **scope}
    return clone

list

list(
    kind: KindLike[T], *, scope: Scope | None = None
) -> Sequence[T]

Returns every record of this kind in scope, validated but unprompted.

Source code in src/clabe/stores/_base.py

def list(self, kind: KindLike[T], *, scope: Scope | None = None) -> Sequence[T]:
    """Returns every record of this kind in scope, validated but unprompted."""
    _kind = as_kind(kind)
    return [self._validated(_kind, c.value) for c in self._candidates(_kind, self._merge_scope(scope))]

resolve

resolve(
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T

Selects a single record: raises on none, auto-selects a lone candidate, prompts otherwise.

Parameters:

Name	Type	Description	Default
kind	KindLike[T]	The kind of record to resolve.	required
scope	Scope | None	Narrows which records are considered. Layered over the store's own scope.	None
pick_kwargs	PickRequestKwargs | None	Optional overrides for the :class:ui.PickRequest shown to the user. Any key present here takes precedence over resolve's own defaults. options is always derived from the store's candidates and is not overridable.	None

Raises:

Type	Description
LookupError	If no record of this kind exists in scope, or the user declined to pick one.
NoFrontendError	If a choice must be made and no frontend is registered.

Source code in src/clabe/stores/_base.py

def resolve(
    self,
    kind: KindLike[T],
    *,
    scope: Scope | None = None,
    pick_kwargs: PickRequestKwargs | None = None,
) -> T:
    """
    Selects a single record: raises on none, auto-selects a lone candidate, prompts otherwise.

    Args:
        kind: The kind of record to resolve.
        scope: Narrows which records are considered. Layered over the store's own scope.
        pick_kwargs: Optional overrides for the :class:`ui.PickRequest` shown to the user.
            Any key present here takes precedence over ``resolve``'s own defaults.
            ``options`` is always derived from the store's candidates and is not overridable.

    Raises:
        LookupError: If no record of this kind exists in scope, or the user declined to pick one.
        ui.NoFrontendError: If a choice must be made and no frontend is registered.
    """
    _kind = as_kind(kind)
    candidates = self._candidates(_kind, self._merge_scope(scope))
    if not candidates:
        raise LookupError(f"No {_kind.name!r} records found in {self}.")
    if len(candidates) == 1:
        ui.notify(f"Found a single {_kind.name}. Using {candidates[0].label}.")
        return self._validated(_kind, candidates[0].value)
    by_label = {c.label: c.value for c in candidates}
    _request_kwargs: dict[str, Any] = {
        "label": f"Choose a {_kind.name} for {_kind.model.__name__}:",
        "allow_none": False,
        "field": _kind.name,
    }
    _request_kwargs.update(pick_kwargs or {})
    picked = ui.prompt_pick(
        ui.PickRequest(
            options=_recent_first(_kind.name, sorted(by_label)),
            **_request_kwargs,
        )
    )
    if picked is None:
        raise LookupError(f"No {_kind.name} was selected.")
    CacheManager.get_instance().add_to_cache(_kind.name, picked)
    return self._validated(_kind, by_label[picked])

write

write(
    kind: KindLike[T],
    value: T,
    *,
    scope: Scope | None = None,
) -> None

Appends a record, so repeated writes accumulate as candidates.

Source code in src/clabe/stores/_memory.py

def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """Appends a record, so repeated writes accumulate as candidates."""
    self._records.append((as_kind(kind).name, self._merge_scope(scope), value))

as_kind

as_kind(kind: KindLike[T]) -> Kind[T]

Normalizes a kind-or-model argument into a :class:Kind.

Source code in src/clabe/stores/_base.py

def as_kind(kind: KindLike[T]) -> Kind[T]:
    """Normalizes a kind-or-model argument into a :class:`Kind`."""
    return kind if isinstance(kind, Kind) else Kind(kind)