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))

FicusClient ¶

FicusClient(
    base_url: str = DEFAULT_BASE_URL,
    *,
    default_extension: str = ".json",
    timeout: float = 10.0,
    session: Session | None = None,
)

An HTTP client for ficus, independent of clabe's store vocabulary.

Example

client = FicusClient("http://eng-tools/ficus-dev")
result = client.get_merged("aind-behavior-vr-foraging", scopes=[COMPUTERS("DT201256")])
print(result.sources)  # contributing documents, lowest precedence first

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Ficus' root URL.	`DEFAULT_BASE_URL`
`default_extension`	`str`	Names a document ficus does not have yet, and is tried first when reading. Reads fall back across :data:`SUPPORTED_EXTENSIONS`, and an existing document keeps the extension it has.	`'.json'`
`timeout`	`float`	Seconds to wait on any single request.	`10.0`
`session`	`Session \| None`	A `requests` session to reuse. Defaults to `None`, in which case one is created.	`None`

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

def __init__(
    self,
    base_url: str = DEFAULT_BASE_URL,
    *,
    default_extension: str = ".json",
    timeout: float = 10.0,
    session: requests.Session | None = None,
) -> None:
    """
    Args:
        base_url: Ficus' root URL.
        default_extension: Names a document ficus does not have yet, and is tried first when
            reading. Reads fall back across :data:`SUPPORTED_EXTENSIONS`, and an existing
            document keeps the extension it has.
        timeout: Seconds to wait on any single request.
        session: A ``requests`` session to reuse. Defaults to ``None``, in which case one is
            created.
    """
    self.base_url = base_url.rstrip("/")
    self.default_extension = default_extension
    self.timeout = timeout
    self._session = session or requests.Session()

get_layer ¶

get_layer(
    namespace: str,
    *,
    filename: str = DEFAULT_FILENAME,
    scope: ScopeRef | None = None,
) -> Layer | None

Fetches exactly one document, without any server-side merging.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	The ficus namespace.	required
`filename`	`str`	The document to fetch.	`DEFAULT_FILENAME`
`scope`	`ScopeRef \| None`	The scope to read from. `None` reads the `defaults` layer.	`None`

Returns:

Type	Description
`Layer \| None`	Layer \| None: The document, or `None` if it does not exist.

Raises:

Type	Description
`HTTPError`	For any failure other than the document being absent.

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

def get_layer(
    self,
    namespace: str,
    *,
    filename: str = DEFAULT_FILENAME,
    scope: ScopeRef | None = None,
) -> Layer | None:
    """Fetches exactly one document, without any server-side merging.

    Args:
        namespace: The ficus namespace.
        filename: The document to fetch.
        scope: The scope to read from. ``None`` reads the ``defaults`` layer.

    Returns:
        Layer | None: The document, or ``None`` if it does not exist.

    Raises:
        requests.HTTPError: For any failure other than the document being absent.
    """
    params: dict[str, str] = {"filename": filename, "merge": "false"}
    if scope is not None:
        params[scope.param] = scope.identifier

    response = self._session.get(self._read_url(namespace), params=params, timeout=self.timeout)
    if response.status_code == 404 and self._is_absent(response):
        logger.debug("No ficus document at namespace=%r filename=%r scope=%r", namespace, filename, scope)
        return None
    response.raise_for_status()

    payload = response.json()
    files = payload.get("details", {}).get("files") or []
    return Layer(
        data=payload["data"],
        source=files[0] if files else filename,
        key=LayerKey(scope=scope, filename=filename),
    )

locate ¶

locate(
    namespace: str,
    stem: str,
    *,
    scope: ScopeRef | None = None,
) -> tuple[str, Layer | None]

Finds which extension a document is stored under, and fetches it.

A read matches the extension exactly, so a document stored as rig.yml is absent to a request for rig.json. :attr:default_extension is tried first, then the rest of :data:SUPPORTED_EXTENSIONS. Ficus allows a stem only one extension at a time, so at most one can exist.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	The ficus namespace.	required
`stem`	`str`	The document name without its extension, e.g. `"default"`.	required
`scope`	`ScopeRef \| None`	The scope to look in. `None` is the `defaults` layer.	`None`

Returns:

Type	Description
`tuple[str, Layer \| None]`	tuple[str, Layer \| None]: The filename to address -- the extension found, or `stem` plus :attr:`default_extension` when none was -- and the layer, if it exists.

Raises:

Type	Description
`HTTPError`	For any failure other than a document being absent.

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

def locate(self, namespace: str, stem: str, *, scope: ScopeRef | None = None) -> tuple[str, Layer | None]:
    """Finds which extension a document is stored under, and fetches it.

    A read matches the extension exactly, so a document stored as ``rig.yml`` is absent to a
    request for ``rig.json``. :attr:`default_extension` is tried first, then the rest of
    :data:`SUPPORTED_EXTENSIONS`. Ficus allows a stem only one extension at a time, so at most
    one can exist.

    Args:
        namespace: The ficus namespace.
        stem: The document name without its extension, e.g. ``"default"``.
        scope: The scope to look in. ``None`` is the ``defaults`` layer.

    Returns:
        tuple[str, Layer | None]: The filename to address -- the extension found, or ``stem``
            plus :attr:`default_extension` when none was -- and the layer, if it exists.

    Raises:
        requests.HTTPError: For any failure other than a document being absent.
    """
    for extension in dict.fromkeys((self.default_extension, *SUPPORTED_EXTENSIONS)):
        filename = f"{stem}{extension}"
        layer = self.get_layer(namespace, filename=filename, scope=scope)
        if layer is not None:
            return filename, layer
    return f"{stem}{self.default_extension}", None

get_merged ¶

get_merged(
    namespace: str,
    *,
    filename: str | None = None,
    scopes: Iterable[ScopeRef] = (),
    policy: MergePolicy | None = None,
) -> MergeResult

Fetches every layer in the chain and merges them.

The chain runs defaults first, then each scope in the order given, and within each layer the default document before any extra filename. This is ficus' documented precedence.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	The ficus namespace.	required
`filename`	`str \| None`	An extra document to layer on top of `default` at every level. `None` uses `default` alone. Taken literally, extension included; the `default` document's extension is resolved per layer by :meth:`locate`.	`None`
`scopes`	`Iterable[ScopeRef]`	Scopes to layer, lowest precedence first.	`()`
`policy`	`MergePolicy \| None`	How to combine layers, and what to do about missing ones.	`None`

Returns:

Name	Type	Description
`MergeResult`	`MergeResult`	The merged document and the paths that produced it.

Raises:

Type	Description
`LookupError`	If `policy.on_missing_layer` is `"raise"` and a layer is absent.

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

def get_merged(
    self,
    namespace: str,
    *,
    filename: str | None = None,
    scopes: Iterable[ScopeRef] = (),
    policy: MergePolicy | None = None,
) -> MergeResult:
    """Fetches every layer in the chain and merges them.

    The chain runs ``defaults`` first, then each scope in the order given, and within each
    layer the ``default`` document before any extra ``filename``. This is ficus' documented
    precedence.

    Args:
        namespace: The ficus namespace.
        filename: An extra document to layer on top of ``default`` at every level. ``None``
            uses ``default`` alone. Taken literally, extension included; the ``default``
            document's extension is resolved per layer by :meth:`locate`.
        scopes: Scopes to layer, lowest precedence first.
        policy: How to combine layers, and what to do about missing ones.

    Returns:
        MergeResult: The merged document and the paths that produced it.

    Raises:
        LookupError: If ``policy.on_missing_layer`` is ``"raise"`` and a layer is absent.
    """
    policy = policy or MergePolicy()
    scopes = list(scopes)  # consumed more than once below, so never leave it a generator
    merged: dict[str, Any] = {}
    sources: list[str] = []
    origins: dict[LeafPath, LayerKey] = {}
    chain: list[LayerKey] = []

    for key, layer in self._fetch_chain(namespace, filename, scopes):
        chain.append(key)
        if layer is None:
            if policy.on_missing_layer == "raise":
                raise LookupError(f"No ficus document at {key} in namespace {namespace!r}")
            continue
        merged = deep_merge(merged, layer.data, policy=policy)
        sources.append(layer.source)
        _stamp_origins(layer.data, key, origins)

    if not sources:
        # An unknown namespace returns the same "not found" as a namespace with no documents.
        logger.warning(
            "No ficus document found at any layer for namespace=%r (scopes=%r). If this is "
            "unexpected, check the namespace spelling -- an unknown namespace looks exactly "
            "like an empty one.",
            namespace,
            [s.identifier for s in scopes],
        )

    return MergeResult(data=merged, sources=sources, origins=origins, chain=chain)

write ¶

write(
    namespace: str,
    data: dict[str, Any],
    *,
    filename: str = DEFAULT_FILENAME,
    scope: ScopeRef | None = None,
    create_only: bool = False,
) -> None

Writes one document, to exactly one layer.

POST is tried first and falls back to PATCH on a 409. Ficus' PATCH deep-merges the payload into the existing document server-side, so a partial write needs no read-modify-write here; it cannot remove a key, which takes a delete and a rewrite.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	The ficus namespace.	required
`data`	`dict[str, Any]`	The document, or the part of it being changed.	required
`filename`	`str`	The document to write.	`DEFAULT_FILENAME`
`scope`	`ScopeRef \| None`	The layer to write to. `None` writes the `defaults` layer, which every consumer of the namespace reads.	`None`
`create_only`	`bool`	Whether to fail rather than update an existing document.	`False`

Raises:

Type	Description
`HTTPError`	If the write fails.

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

def write(
    self,
    namespace: str,
    data: dict[str, Any],
    *,
    filename: str = DEFAULT_FILENAME,
    scope: ScopeRef | None = None,
    create_only: bool = False,
) -> None:
    """Writes one document, to exactly one layer.

    ``POST`` is tried first and falls back to ``PATCH`` on a 409. Ficus' ``PATCH`` deep-merges
    the payload into the existing document server-side, so a partial write needs no
    read-modify-write here; it cannot remove a key, which takes a delete and a rewrite.

    Args:
        namespace: The ficus namespace.
        data: The document, or the part of it being changed.
        filename: The document to write.
        scope: The layer to write to. ``None`` writes the ``defaults`` layer, which every
            consumer of the namespace reads.
        create_only: Whether to fail rather than update an existing document.

    Raises:
        requests.HTTPError: If the write fails.
    """
    url = self._write_url(namespace, filename=filename, scope=scope)
    verb = "POST"
    response = self._session.post(url, json=data, timeout=self.timeout)
    if response.status_code == 409 and not create_only:
        verb = "PATCH"
        response = self._session.patch(url, json=data, timeout=self.timeout)
    response.raise_for_status()
    logger.info("Wrote ficus document %s (%s)", url, verb)

plan_write_back `staticmethod` ¶

plan_write_back(
    current: MergeResult,
    new_data: dict[str, Any],
    *,
    target: LayerKey | None = None,
    policy: WritePolicy | None = None,
) -> LayerWrite | None

Works out what a write-back would do, without doing it.

Only the leaves that differ are written, to a single layer. The default target is the top of the merge chain, which no layer can shadow. A lower target is allowed, and policy.on_shadowed_write then governs any leaf whose current value comes from above it.

Parameters:

Name	Type	Description	Default
`current`	`MergeResult`	The config as it resolves now, from :meth:`get_merged`. Its `origins` and `chain` drive the shadow check.	required
`new_data`	`dict[str, Any]`	The config as it should resolve.	required
`target`	`LayerKey \| None`	The layer to write to. `None` (default) means the top of `current.chain`.	`None`
`policy`	`WritePolicy \| None`	How to handle removals and shadowed writes.	`None`

Returns:

Type	Description
`LayerWrite \| None`	LayerWrite \| None: The write to perform, or `None` if nothing changed.

Raises:

Type	Description
`ValueError`	If there is no chain to write to.
`LookupError`	If a key would have to be removed and `policy.on_key_removal` is `"raise"`.
`PermissionError`	If the write would be shadowed and `policy.on_shadowed_write` is `"raise"`.

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

@staticmethod
def plan_write_back(
    current: MergeResult,
    new_data: dict[str, Any],
    *,
    target: LayerKey | None = None,
    policy: WritePolicy | None = None,
) -> LayerWrite | None:
    """Works out what a write-back would do, without doing it.

    Only the leaves that differ are written, to a single layer. The default target is the top
    of the merge chain, which no layer can shadow. A lower target is allowed, and
    ``policy.on_shadowed_write`` then governs any leaf whose current value comes from above it.

    Args:
        current: The config as it resolves now, from :meth:`get_merged`. Its ``origins`` and
            ``chain`` drive the shadow check.
        new_data: The config as it should resolve.
        target: The layer to write to. ``None`` (default) means the top of ``current.chain``.
        policy: How to handle removals and shadowed writes.

    Returns:
        LayerWrite | None: The write to perform, or ``None`` if nothing changed.

    Raises:
        ValueError: If there is no chain to write to.
        LookupError: If a key would have to be removed and ``policy.on_key_removal`` is
            ``"raise"``.
        PermissionError: If the write would be shadowed and ``policy.on_shadowed_write`` is
            ``"raise"``.
    """
    policy = policy or WritePolicy()
    if not current.chain:
        raise ValueError("This config resolved from no chain at all, so there is no layer to write back to.")
    key = target if target is not None else current.chain[-1]

    changed, removed = diff_leaves(current.data, new_data)
    if removed and policy.on_key_removal == "raise":
        raise LookupError(
            f"Writing back would have to remove {[_dotted(p) for p in removed]!r}, but ficus' PATCH "
            f"merges and cannot remove a key. Delete and rewrite {key}, or set "
            f"WritePolicy(on_key_removal='skip') to leave these keys in place."
        )
    if removed:
        logger.warning("Leaving %r in place: a partial write cannot remove keys.", [_dotted(p) for p in removed])

    if not changed:
        logger.debug("Write-back to %s is a no-op: nothing changed.", key)
        return None

    FicusClient._check_not_shadowed(current, changed, key, policy)
    return LayerWrite(key=key, data=nest_leaves(changed), paths=sorted(changed))

write_back ¶

write_back(
    namespace: str,
    current: MergeResult,
    new_data: dict[str, Any],
    *,
    target: LayerKey | None = None,
    policy: WritePolicy | None = None,
) -> LayerWrite | None

Writes the difference between a resolved config and a new one back to a single layer.

See :meth:plan_write_back for how the layer and the payload are chosen; this runs that plan.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	The ficus namespace.	required
`current`	`MergeResult`	The config as it resolves now, from :meth:`get_merged`.	required
`new_data`	`dict[str, Any]`	The config as it should resolve.	required
`target`	`LayerKey \| None`	The layer to write to. `None` (default) means the top of the merge chain.	`None`
`policy`	`WritePolicy \| None`	How to handle removals and shadowed writes.	`None`

Returns:

Type	Description
`LayerWrite \| None`	LayerWrite \| None: What was written, or `None` if nothing changed.

Raises:

Type	Description
`HTTPError`	If the write fails.

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

def write_back(
    self,
    namespace: str,
    current: MergeResult,
    new_data: dict[str, Any],
    *,
    target: LayerKey | None = None,
    policy: WritePolicy | None = None,
) -> LayerWrite | None:
    """Writes the difference between a resolved config and a new one back to a single layer.

    See :meth:`plan_write_back` for how the layer and the payload are chosen; this runs that
    plan.

    Args:
        namespace: The ficus namespace.
        current: The config as it resolves now, from :meth:`get_merged`.
        new_data: The config as it should resolve.
        target: The layer to write to. ``None`` (default) means the top of the merge chain.
        policy: How to handle removals and shadowed writes.

    Returns:
        LayerWrite | None: What was written, or ``None`` if nothing changed.

    Raises:
        requests.HTTPError: If the write fails.
    """
    plan = self.plan_write_back(current, new_data, target=target, policy=policy)
    if plan is None:
        return None
    self.write(namespace, plan.data, filename=plan.key.filename, scope=plan.key.scope)
    logger.info("Wrote %d changed value(s) back to %s", len(plan.paths), plan.key)
    return plan

delete ¶

delete(
    namespace: str,
    *,
    filename: str = DEFAULT_FILENAME,
    scope: ScopeRef | None = None,
) -> bool

Deletes one document.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	The ficus namespace.	required
`filename`	`str`	The document to delete.	`DEFAULT_FILENAME`
`scope`	`ScopeRef \| None`	The layer to delete from. `None` is the `defaults` layer.	`None`

Returns:

Name	Type	Description
`bool`	`bool`	True if a document was deleted, False if there was nothing there.

Raises:

Type	Description
`HTTPError`	If the delete fails for any reason other than absence.

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

def delete(self, namespace: str, *, filename: str = DEFAULT_FILENAME, scope: ScopeRef | None = None) -> bool:
    """Deletes one document.

    Args:
        namespace: The ficus namespace.
        filename: The document to delete.
        scope: The layer to delete from. ``None`` is the ``defaults`` layer.

    Returns:
        bool: True if a document was deleted, False if there was nothing there.

    Raises:
        requests.HTTPError: If the delete fails for any reason other than absence.
    """
    response = self._session.delete(
        self._write_url(namespace, filename=filename, scope=scope), timeout=self.timeout
    )
    if response.status_code == 404 and self._is_absent(response):
        return False
    response.raise_for_status()
    return True

list_files ¶

list_files(
    namespace: str, *, scope: ScopeRef | None = None
) -> list[str]

Lists the documents visible for a namespace, optionally within one scope.

Parameters:

Name	Type	Description	Default
`namespace`	`str`	The ficus namespace.	required
`scope`	`ScopeRef \| None`	Restrict to this scope. `None` lists the `defaults` layer only.	`None`

Returns:

Type	Description
`list[str]`	list[str]: The document paths, or an empty list if the scope holds nothing.

Raises:

Type	Description
`HTTPError`	For any failure other than the scope being absent.

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

def list_files(self, namespace: str, *, scope: ScopeRef | None = None) -> list[str]:
    """Lists the documents visible for a namespace, optionally within one scope.

    Args:
        namespace: The ficus namespace.
        scope: Restrict to this scope. ``None`` lists the ``defaults`` layer only.

    Returns:
        list[str]: The document paths, or an empty list if the scope holds nothing.

    Raises:
        requests.HTTPError: For any failure other than the scope being absent.
    """
    params = {} if scope is None else {scope.param: scope.identifier}
    url = f"{self.base_url}/v1/list_files/{namespace}/configs"
    response = self._session.get(url, params=params, timeout=self.timeout)
    if response.status_code == 404 and self._is_absent(response):
        return []
    response.raise_for_status()
    return response.json()["data"]

FicusSettings ¶

Bases: ServiceSettings

Settings for :class:FicusStore, read from the ficus section of clabe.yml.

settings_customise_sources `classmethod` ¶

settings_customise_sources(
    settings_cls: type[BaseSettings],
    init_settings: PydanticBaseSettingsSource,
    env_settings: PydanticBaseSettingsSource,
    dotenv_settings: PydanticBaseSettingsSource,
    file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]

Customizes the settings sources to include the safe YAML settings source.

Parameters:

Name	Type	Description	Default
`settings_cls`	`type[BaseSettings]`	The settings class	required
`init_settings`	`PydanticBaseSettingsSource`	The initial settings source	required
`env_settings`	`PydanticBaseSettingsSource`	The environment settings source	required
`dotenv_settings`	`PydanticBaseSettingsSource`	The dotenv settings source	required
`file_secret_settings`	`PydanticBaseSettingsSource`	The file secret settings source	required

Returns:

Type	Description
`tuple[PydanticBaseSettingsSource, ...]`	Tuple[PydanticBaseSettingsSource, ...]: A tuple of settings sources

Source code in src/clabe/services.py

@classmethod
def settings_customise_sources(
    cls,
    settings_cls: type[ps.BaseSettings],
    init_settings: ps.PydanticBaseSettingsSource,
    env_settings: ps.PydanticBaseSettingsSource,
    dotenv_settings: ps.PydanticBaseSettingsSource,
    file_secret_settings: ps.PydanticBaseSettingsSource,
) -> tuple[ps.PydanticBaseSettingsSource, ...]:
    """
    Customizes the settings sources to include the safe YAML settings source.

    Args:
        settings_cls: The settings class
        init_settings: The initial settings source
        env_settings: The environment settings source
        dotenv_settings: The dotenv settings source
        file_secret_settings: The file secret settings source

    Returns:
        Tuple[PydanticBaseSettingsSource, ...]: A tuple of settings sources
    """
    yaml_sources = [
        _SafeYamlSettingsSource(settings_cls, yaml_file=p, yaml_config_section=cls.__yml_section__)
        for p in KNOWN_CONFIG_FILES
    ]
    # The in-memory document ranks right after the first (local override) config file
    return (
        init_settings,
        *yaml_sources[:1],
        _InMemorySettingsSource(settings_cls, config_section=cls.__yml_section__),
        *yaml_sources[1:],
        env_settings,
        dotenv_settings,
        file_secret_settings,
    )

FicusStore ¶

FicusStore(
    settings: FicusSettings | None = None,
    *,
    client: FicusClient | None = None,
    serves: str | set[str] | None = None,
    scope: Scope | None = None,
)

Bases: StoreBase

A store over a single ficus namespace.

Serves two shapes of document, told apart by kind name:

A kind in config_kinds ("rig" by default) is layered config: merged across defaults and the configured scopes, from ficus' default document. The merge yields exactly one document, so :meth:resolve never prompts.
Any other kind is a flat record: a standalone document named after the kind, living in one scope, read and written whole. :class:~clabe.modifiers.ByAnimalModifier uses this shape.

Example

store = FicusStore(FicusSettings(namespace="aind-behavior-vr-foraging"))
rig = store.resolve(Kind.from_rig(AindVrForagingRig))
position = store.scoped(subject="789907").resolve(Kind(ManipulatorPosition))

Parameters:

Name	Type	Description	Default
`settings`	`FicusSettings \| None`	Settings for this store. Defaults to `None`, in which case :class:`FicusSettings` is built from the `ficus` YAML section and the environment, and construction fails if `namespace` cannot be resolved that way.	`None`
`client`	`FicusClient \| None`	The ficus client. Defaults to `None`, in which case one is built from `settings`.	`None`
`serves`	`str \| set[str] \| None`	Kind name(s) this store will answer for, or `None` (default) for any kind.	`None`
`scope`	`Scope \| None`	Initial scope. `computer_name` defaults to this machine's and selects the rig layer; narrow it with `scoped(computer_name=...)` to address another machine.	`None`

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

def __init__(
    self,
    settings: FicusSettings | None = None,
    *,
    client: FicusClient | None = None,
    serves: str | set[str] | None = None,
    scope: Scope | None = None,
) -> None:
    """
    Args:
        settings: Settings for this store. Defaults to ``None``, in which case
            :class:`FicusSettings` is built from the ``ficus`` YAML section and the
            environment, and construction fails if ``namespace`` cannot be resolved that way.
        client: The ficus client. Defaults to ``None``, in which case one is built from
            ``settings``.
        serves: Kind name(s) this store will answer for, or ``None`` (default) for any kind.
        scope: Initial scope. ``computer_name`` defaults to this machine's and selects the rig
            layer; narrow it with ``scoped(computer_name=...)`` to address another machine.
    """
    super().__init__(scope={"computer_name": get_computer_name(), **(scope or {})})
    self._settings = settings or FicusSettings()
    self._client = client or FicusClient(
        self._settings.base_url,
        default_extension=self._settings.extension,
        timeout=self._settings.timeout,
    )
    self._serves: set[str] | None = (
        None if serves is None else ({serves} if isinstance(serves, str) else set(serves))
    )

scope `property` ¶

scope: Scope

The scope every call on this store is narrowed by.

client `property` ¶

client: FicusClient

The underlying ficus client.

namespace `property` ¶

namespace: str

The ficus namespace this store reads.

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

Persists a value.

A flat record is written whole, to the subject's scope when one is in scope and the rig's otherwise. Layered config is written as a diff against its current resolution: only the leaves that differ are sent, to the top of the merge chain. See :meth:FicusClient.plan_write_back.

Parameters:

Name	Type	Description	Default
`kind`	`KindLike[T]`	The kind being written.	required
`value`	`T`	The value.	required
`scope`	`Scope \| None`	Extra scope for this call, layered over the store's own.	`None`

Raises:

Type	Description
`LookupError`	If this store does not serve `kind`, if the chain has no scope to write to, or if writing back would have to remove a key.
`PermissionError`	If a layered write would land below a layer that overrides it.
`HTTPError`	If the write fails.

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

def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """Persists a value.

    A flat record is written whole, to the subject's scope when one is in scope and the rig's
    otherwise. Layered config is written as a diff against its current resolution: only the
    leaves that differ are sent, to the top of the merge chain. See
    :meth:`FicusClient.plan_write_back`.

    Args:
        kind: The kind being written.
        value: The value.
        scope: Extra scope for this call, layered over the store's own.

    Raises:
        LookupError: If this store does not serve ``kind``, if the chain has no scope to write
            to, or if writing back would have to remove a key.
        PermissionError: If a layered write would land below a layer that overrides it.
        requests.HTTPError: If the write fails.
    """
    _kind = as_kind(kind)
    self._require_supported(_kind)
    merged = self._merge_scope(scope)
    payload = _kind.adapter.dump_python(value, mode="json")

    if self._is_config(_kind):
        self._write_config(_kind, payload, merged)
        return

    target = self._record_scope(merged)
    # Ficus allows a stem only one extension, so an existing document is the one to update.
    filename, _ = self._client.locate(self.namespace, _kind.name, scope=target)
    self._client.write(self.namespace, payload, filename=filename, scope=target)

LayerKey `dataclass` ¶

LayerKey(scope: ScopeRef | None, filename: str)

Names a layer, whether or not a document exists there yet.

Attributes:

Name	Type	Description
`scope`	`ScopeRef \| None`	The scope the document lives in. `None` is the `defaults` layer.
`filename`	`str`	The document name within that scope.

MergePolicy `dataclass` ¶

MergePolicy(
    null_means: Literal["value", "delete"] = "value",
    lists: Literal["replace", "concat"] = "replace",
    on_type_conflict: Literal[
        "override", "warn", "raise"
    ] = "warn",
    on_missing_layer: Literal["skip", "raise"] = "skip",
)

How layers combine. The defaults reproduce ficus' _deep_update.

Attributes:

Name	Type	Description
`null_means`	`Literal['value', 'delete']`	`"value"` (default) treats `null` as a value, so a layer can blank out a field inherited from below. `"delete"` applies RFC 7386 semantics and removes the key.
`lists`	`Literal['replace', 'concat']`	`"replace"` (default) lets a later list win outright. `"concat"` appends.
`on_type_conflict`	`Literal['override', 'warn', 'raise']`	What to do when a layer replaces a dict with a non-dict or vice versa. `"warn"` (default) logs and takes the override; `"override"` is silent; `"raise"` refuses.
`on_missing_layer`	`Literal['skip', 'raise']`	`"skip"` (default) skips an absent document. `"raise"` requires every requested layer to exist.

WritePolicy `dataclass` ¶

WritePolicy(
    on_key_removal: Literal["raise", "skip"] = "raise",
    on_shadowed_write: Literal[
        "raise", "warn", "allow"
    ] = "raise",
)

How a layered-config write-back behaves.

Attributes:

Name	Type	Description
`on_key_removal`	`Literal['raise', 'skip']`	What to do when a key present in the resolved config is absent from the value being written back. Ficus' `PATCH` merges and cannot remove a key. `"raise"` (default) refuses; `"skip"` leaves the key in place and writes the rest.
`on_shadowed_write`	`Literal['raise', 'warn', 'allow']`	What to do when a changed leaf's current value comes from a layer above the one being written to, so the next read would return the old value. `"raise"` (default) refuses, `"warn"` logs and writes anyway, `"allow"` is silent. Cannot arise when writing to the top of the chain, which is the default target.

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)

stores¶

Candidate dataclass ¶

CompositeStore ¶

scope property ¶

resolve ¶

list ¶

write ¶

scoped ¶

Kind ¶

adapter cached property ¶

from_rig staticmethod ¶

from_task staticmethod ¶

from_session staticmethod ¶

from_trainer_state staticmethod ¶

Store ¶

resolve ¶

list ¶

write ¶

scoped ¶

StoreBase ¶

scope property ¶

scoped ¶

write abstractmethod ¶

list ¶

resolve ¶

DefaultLayout ¶

read ¶

write ¶

Layout ¶

read ¶

write ¶

LocalFileStore ¶

scope property ¶

root property ¶

scoped ¶

list ¶

resolve ¶

write ¶

MemoryStore ¶

scope property ¶

scoped ¶

list ¶

resolve ¶

write ¶

FicusClient ¶

get_layer ¶

locate ¶

get_merged ¶

write ¶

plan_write_back staticmethod ¶

write_back ¶

delete ¶

list_files ¶

FicusSettings ¶

settings_customise_sources classmethod ¶

FicusStore ¶

scope property ¶

client property ¶

namespace property ¶

scoped ¶

list ¶

resolve ¶

write ¶

LayerKey dataclass ¶

MergePolicy dataclass ¶

WritePolicy dataclass ¶

as_kind ¶

Candidate `dataclass` ¶

scope `property` ¶

adapter `cached` `property` ¶

from_rig `staticmethod` ¶

from_task `staticmethod` ¶

from_session `staticmethod` ¶

from_trainer_state `staticmethod` ¶

scope `property` ¶

write `abstractmethod` ¶

scope `property` ¶

root `property` ¶

scope `property` ¶

plan_write_back `staticmethod` ¶

settings_customise_sources `classmethod` ¶

scope `property` ¶

client `property` ¶

namespace `property` ¶

LayerKey `dataclass` ¶

MergePolicy `dataclass` ¶

WritePolicy `dataclass` ¶