stores.dataverse

DataverseSuggestion

Bases: BaseModel

Internal representation of a suggestion entry in Dataverse.

validate_trainer_state classmethod

validate_trainer_state(value)

Validate and convert the trainer_state field from a JSON string to a TrainerState object.

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

@field_validator("trainer_state", mode="before")
@classmethod
def validate_trainer_state(cls, value):
    """
    Validate and convert the trainer_state field from a JSON string to a TrainerState object.
    """
    if value is None:
        return value
    if isinstance(value, str):
        return TrainerState.model_validate_json(value)
    return value

from_request_output classmethod

from_request_output(
    subject: str, request_output: dict
) -> DataverseSuggestion

Create a _Suggestion instance from a dictionary of data.

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

@classmethod
def from_request_output(cls, subject: str, request_output: dict) -> "DataverseSuggestion":
    """
    Create a _Suggestion instance from a dictionary of data.
    """
    trainer_state = request_output.get("aibs_trainer_state", None)
    trainer_state = TrainerState.model_validate_json(cls._strip_html(trainer_state)) if trainer_state else None
    return cls(
        subject_id=subject,
        trainer_state=trainer_state,
        task_name=request_output.get("aibs_task_name", None),
        stage_name=request_output.get("aibs_stage_name", None),
        modified_on=request_output.get("modifiedon", None),
        created_on=request_output.get("createdon", None),
    )

from_trainer_state classmethod

from_trainer_state(
    subject: str, trainer_state: TrainerState
) -> DataverseSuggestion

Create a _Suggestion instance from a TrainerState object.

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

@classmethod
def from_trainer_state(cls, subject: str, trainer_state: TrainerState) -> "DataverseSuggestion":
    """
    Create a _Suggestion instance from a TrainerState object.
    """
    if trainer_state is None:
        raise ValueError("trainer_state cannot be None")
    if trainer_state.stage is None:
        raise ValueError("trainer_state.stage cannot be None")
    return cls(
        subject_id=subject,
        trainer_state=trainer_state,
        task_name=trainer_state.stage.task.name,
        stage_name=trainer_state.stage.name,
    )

DataverseStore

DataverseStore(
    *,
    client: _DataverseRestClient | None = None,
    history: int = _DEFAULT_HISTORY,
    scope: Scope | None = None,
)

Bases: StoreBase

A store over the Dataverse suggestion tables, serving trainer state only.

Reads and appends to aibs_fact_mouse_proposed_behavior_sessionses, keyed through aibs_dim_mices. Requires subject and task_name in scope.

Example

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

Parameters:

Name	Type	Description	Default
client	_DataverseRestClient | None	The REST client. Defaults to one built from KeePass credentials.	None
history	int	How many recent suggestions to offer. At the default of 1 the latest suggestion is used without prompting.	_DEFAULT_HISTORY
scope	Scope | None	Initial scope.	None

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

def __init__(
    self,
    *,
    client: _DataverseRestClient | None = None,
    history: int = _DEFAULT_HISTORY,
    scope: Scope | None = None,
) -> None:
    """
    Args:
        client: The REST client. Defaults to one built from KeePass credentials.
        history: How many recent suggestions to offer. At the default of 1 the
            latest suggestion is used without prompting.
        scope: Initial scope.
    """
    super().__init__(scope=scope)
    self._client = client or _DataverseRestClient(_DataverseRestClientSettings.from_keepass())
    self._history = history

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

Appends a new suggestion for the subject in scope.

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

def write(self, kind: KindLike[T], value: T, *, scope: Scope | None = None) -> None:
    """Appends a new suggestion for the subject in scope."""
    self._require_supported(as_kind(kind))
    subject, _ = self._required_scope(self._merge_scope(scope))
    if not isinstance(value, TrainerState):
        raise TypeError(f"Expected a TrainerState, got {type(value).__name__}.")
    logger.info("Pushing new suggestion to Dataverse for subject %s", subject)
    _append_suggestion(self._client, subject, value)

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