stores.ficus

A store backend over ficus, a layered configuration service.

A config is assembled from a defaults document plus one document per scope (a computer, a subject), deep-merged in precedence order. Two shapes of document are served:

• Config -- layered, merged from ficus' conventional default document at each scope. See :meth:FicusClient.get_merged and :class:MergePolicy.
• Records -- flat, standalone documents named after their kind (manipulator_position.json), belonging to exactly one scope, read and written whole. Applied to a model by :class:~clabe.modifiers.ByAnimalModifier.

Layers are fetched individually with merge=false and merged client-side; ficus' own merge=true returns 404 for the whole request when any requested scope has no document. The merge reproduces ficus' _deep_update: dicts recurse, everything else is replaced, and null is a value rather than a deletion.

Writing a config back sends only the leaves that differ from the current resolution, to a single layer. See :meth:FicusClient.plan_write_back.

ScopeRef dataclass

ScopeRef(segment: Segment, param: str, identifier: str)

One ficus scope, narrowed to a single identifier.

A scope has a collection name and a parameter name. Ficus' routes use them differently::

POST|PATCH|DELETE  /v1/computers/{hostname}/namespaces/{namespace}/config/{filename}
POST|PATCH|DELETE  /v1/subjects/{subject_id}/namespaces/{namespace}/config/{filename}
GET                /v1/namespaces/{namespace}/config?hostname=&subject_id=&filename=&merge=

Writes address a scope by path, under the collection. There is no scoped GET route -- a GET on a write path returns 405 -- so reads address it by query parameter on the unscoped route. An unrecognised query parameter is ignored, and such a read returns the defaults layer.

Attributes:

Name	Type	Description
segment	Segment	The collection a write addresses this scope under, e.g. "computers".
param	str	The identifier's parameter name, e.g. "hostname" -- the path variable when writing, the query parameter when reading.
identifier	str	The machine name or subject id.

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.

Layer dataclass

Layer(
    data: dict[str, Any],
    source: str,
    key: LayerKey | None = None,
)

One document fetched from ficus, with the znode path it came from.

Attributes:

Name	Type	Description
data	dict[str, Any]	The document's contents.
source	str	The path ficus reports for it, e.g. /scratch/computers/DT201256/aind-behavior-vr-foraging/default.json.
key	LayerKey | None	The layer as addressed, rather than as reported. None when not recorded.

MergeResult dataclass

MergeResult(
    data: dict[str, Any],
    sources: list[str],
    origins: dict[LeafPath, LayerKey] = dict(),
    chain: list[LayerKey] = list(),
)

A merged config together with the documents that produced it.

Attributes:

Name	Type	Description
data	dict[str, Any]	The merged document.
sources	list[str]	The contributing paths, in the order they were applied (lowest precedence first). Empty when no layer existed at all.
origins	dict[LeafPath, LayerKey]	For each leaf in :attr:data, the layer whose value won it.
chain	list[LayerKey]	Every layer consulted, in precedence order, including those that held no document.

origin

origin(path: LeafPath) -> LayerKey | None

Returns the layer a leaf's value came from.

Parameters:

Name	Type	Description	Default
path	LeafPath	The leaf to look up.	required

Returns:

Type	Description
LayerKey | None	LayerKey | None: The winning layer, or None if nothing recorded one.

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

def origin(self, path: LeafPath) -> LayerKey | None:
    """Returns the layer a leaf's value came from.

    Args:
        path: The leaf to look up.

    Returns:
        LayerKey | None: The winning layer, or ``None`` if nothing recorded one.
    """
    return self.origins.get(path)

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.

LayerWrite dataclass

LayerWrite(
    key: LayerKey,
    data: dict[str, Any],
    paths: list[LeafPath],
)

One document a write-back will PATCH.

Attributes:

Name	Type	Description
key	LayerKey	The layer being written.
data	dict[str, Any]	The partial document to send: the changed leaves, nested back into shape.
paths	list[LeafPath]	The leaves this write carries.

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

client property

client: FicusClient

The underlying ficus client.

namespace property

namespace: str

The ficus namespace this store reads.

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

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)

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

deep_merge

deep_merge(
    base: dict[str, Any],
    override: dict[str, Any],
    *,
    policy: MergePolicy | None = None,
    _path: str = "",
) -> dict[str, Any]

Merges override onto base, mirroring ficus' own _deep_update.

Dicts present on both sides recurse; everything else is replaced outright, lists included. null on the overriding side is a value, not a deletion, unless policy says otherwise.

Parameters:

Name	Type	Description	Default
base	dict[str, Any]	The lower-precedence document.	required
override	dict[str, Any]	The higher-precedence document, layered on top.	required
policy	MergePolicy | None	How to combine. on_missing_layer is not read here; it governs fetching, and is applied by :meth:FicusClient.get_merged.	None
_path	str	Internal. Dotted path to the current level, used only in messages.	''

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: A new merged document. Neither argument is modified.

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

def deep_merge(
    base: dict[str, Any],
    override: dict[str, Any],
    *,
    policy: MergePolicy | None = None,
    _path: str = "",
) -> dict[str, Any]:
    """Merges ``override`` onto ``base``, mirroring ficus' own ``_deep_update``.

    Dicts present on both sides recurse; everything else is replaced outright, lists included.
    ``null`` on the overriding side is a value, not a deletion, unless ``policy`` says otherwise.

    Args:
        base: The lower-precedence document.
        override: The higher-precedence document, layered on top.
        policy: How to combine. ``on_missing_layer`` is not read here; it governs fetching, and is
            applied by :meth:`FicusClient.get_merged`.
        _path: Internal. Dotted path to the current level, used only in messages.

    Returns:
        dict[str, Any]: A new merged document. Neither argument is modified.
    """
    policy = policy or MergePolicy()
    result = dict(base)
    for key, value in override.items():
        here = f"{_path}.{key}" if _path else key

        if value is None and policy.null_means == "delete":
            result.pop(key, None)
            continue

        current = result.get(key, MISSING)

        if isinstance(current, dict) and isinstance(value, dict):
            result[key] = deep_merge(current, value, policy=policy, _path=here)
            continue

        if isinstance(current, list) and isinstance(value, list) and policy.lists == "concat":
            result[key] = [*current, *value]
            continue

        # A dict replacing a scalar, or the reverse. Filling in a null is not a conflict.
        if (
            current is not MISSING
            and current is not None
            and value is not None
            and isinstance(current, dict) != isinstance(value, dict)
        ):
            _note_type_conflict(here, current, value, policy)

        result[key] = value
    return result

iter_leaves

iter_leaves(
    data: dict[str, Any], _prefix: LeafPath = ()
) -> Iterator[tuple[LeafPath, Any]]

Walks a document, yielding every leaf and the path of keys that reaches it.

A leaf is anything :func:deep_merge replaces wholesale: scalars, None, lists and empty dicts. Non-empty dicts are recursed into.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The document to walk.	required
_prefix	LeafPath	Internal. The path to the current level.	()

Yields:

Type	Description
tuple[LeafPath, Any]	tuple[LeafPath, Any]: Each leaf's path and value.

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

def iter_leaves(data: dict[str, Any], _prefix: LeafPath = ()) -> Iterator[tuple[LeafPath, Any]]:
    """Walks a document, yielding every leaf and the path of keys that reaches it.

    A leaf is anything :func:`deep_merge` replaces wholesale: scalars, ``None``, lists and empty
    dicts. Non-empty dicts are recursed into.

    Args:
        data: The document to walk.
        _prefix: Internal. The path to the current level.

    Yields:
        tuple[LeafPath, Any]: Each leaf's path and value.
    """
    for key, value in data.items():
        here = (*_prefix, key)
        if isinstance(value, dict) and value:
            yield from iter_leaves(value, here)
        else:
            yield here, value

diff_leaves

diff_leaves(
    old: dict[str, Any], new: dict[str, Any]
) -> tuple[dict[LeafPath, Any], list[LeafPath]]

Finds the leaves that differ between two documents.

Parameters:

Name	Type	Description	Default
old	dict[str, Any]	The document as it currently resolves.	required
new	dict[str, Any]	The document as it should resolve.	required

Returns:

Type	Description
tuple[dict[LeafPath, Any], list[LeafPath]]	tuple[dict[LeafPath, Any], list[LeafPath]]: The changed leaves with their new values, and the paths present only in old.

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

def diff_leaves(old: dict[str, Any], new: dict[str, Any]) -> tuple[dict[LeafPath, Any], list[LeafPath]]:
    """Finds the leaves that differ between two documents.

    Args:
        old: The document as it currently resolves.
        new: The document as it should resolve.

    Returns:
        tuple[dict[LeafPath, Any], list[LeafPath]]: The changed leaves with their new values, and
            the paths present only in ``old``.
    """
    old_leaves = dict(iter_leaves(old))
    new_leaves = dict(iter_leaves(new))

    changed = {path: value for path, value in new_leaves.items() if path not in old_leaves or old_leaves[path] != value}

    # A leaf that turned into a subtree, or the reverse, appears as both a change and a removal.
    # The change already replaces everything underneath it, so it is not also a removal.
    def _related_to_a_change(path: LeafPath) -> bool:
        return any(path[: len(c)] == c or c[: len(path)] == path for c in changed)

    removed = [path for path in old_leaves if path not in new_leaves and not _related_to_a_change(path)]
    return changed, removed

nest_leaves

nest_leaves(leaves: dict[LeafPath, Any]) -> dict[str, Any]

Rebuilds a nested document from flat leaf paths.

Parameters:

Name	Type	Description	Default
leaves	dict[LeafPath, Any]	Leaf paths and their values.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: The nested document.

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

def nest_leaves(leaves: dict[LeafPath, Any]) -> dict[str, Any]:
    """Rebuilds a nested document from flat leaf paths.

    Args:
        leaves: Leaf paths and their values.

    Returns:
        dict[str, Any]: The nested document.
    """
    out: dict[str, Any] = {}
    for path, value in leaves.items():
        node = out
        for part in path[:-1]:
            node = node.setdefault(part, {})
        node[path[-1]] = value
    return out