stores.confierge

ConfiergeSettings

Bases: ServiceSettings

Settings for :class:ConfiergeStore, read from the confierge 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,
    )

ConfiergeStore

ConfiergeStore(
    settings: ConfiergeSettings,
    *,
    scope: Scope | None = None,
    scope_mapping: ScopeMapping | None = None,
    client: Confierge | None = None,
)

Bases: StoreBase

A store over ficus, through the confierge client.

This is a thin adapter: it translates the store API's scope into ficus' scopes (see :data:DEFAULT_SCOPE_MAPPING) and leaves merging, validation of scope names and the offline cache to ficus and confierge. Ficus owns the layering, so a read yields exactly one document and never prompts.

A store addresses one ficus namespace, whatever the kind. To serve several kinds, give each its own store and route them with :class:~clabe.stores.CompositeStore.

Example

store = ConfiergeStore(ConfiergeSettings(namespace="aind-behavior-vr-foraging"))
rig = store.scoped(subject="789907").resolve(Kind.from_rig(MyRig))

Parameters:

Name	Type	Description	Default
settings	ConfiergeSettings	Which namespace and server to use, and how writes behave.	required
scope	Scope | None	Initial scope. computer_name defaults to this machine's, as in :class:~clabe.stores.LocalFileStore.	None
scope_mapping	ScopeMapping | None	Translates scope keys into ficus scope names. Defaults to :data:DEFAULT_SCOPE_MAPPING.	None
client	Confierge | None	A ready confierge client. Built from settings.base_url when omitted.	None

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

def __init__(
    self,
    settings: ConfiergeSettings,
    *,
    scope: Scope | None = None,
    scope_mapping: ScopeMapping | None = None,
    client: Confierge | None = None,
) -> None:
    """
    Args:
        settings: Which namespace and server to use, and how writes behave.
        scope: Initial scope. ``computer_name`` defaults to this machine's, as in
            :class:`~clabe.stores.LocalFileStore`.
        scope_mapping: Translates scope keys into ficus scope names. Defaults to
            :data:`DEFAULT_SCOPE_MAPPING`.
        client: A ready confierge client. Built from ``settings.base_url`` when omitted.
    """
    super().__init__(scope={"computer_name": get_computer_name(), **(scope or {})})
    self._settings = settings
    self._scope_mapping = dict(DEFAULT_SCOPE_MAPPING if scope_mapping is None else scope_mapping)
    self._client = client if client is not None else Confierge(base_url=settings.base_url)

scope property

scope: Scope

The scope every call on this store is narrowed by.

write

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

Posts a record to ficus, which decides which layer receives it from the scope.

Raises:

Type	Description
TypeError	If the record does not serialize to a JSON object.

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

def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """
    Posts a record to ficus, which decides which layer receives it from the scope.

    Raises:
        TypeError: If the record does not serialize to a JSON object.
    """
    _kind = as_kind(kind)
    payload = _kind.adapter.dump_python(value, mode="json")
    if not isinstance(payload, dict):
        raise TypeError(f"{_kind!r} must serialize to a JSON object to be written, got {type(payload).__name__}.")
    self._client.post_config_file(
        namespace=self._settings.namespace,
        config_data=payload,
        mode=self._settings.mode,
        scopes=self._ficus_scopes(self._merge_scope(scope)),
        overwrite_defaults=self._settings.overwrite_defaults,
        create_if_missing=self._settings.create_if_missing,
        append_new_fields_to_last_scope=self._settings.append_new_fields_to_last_scope,
    )

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