Skip to content

aind_apps.waterlog

WaterlogSettings

Bases: ServiceSettings

Settings for the waterlog service.

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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
@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,
    )

WaterlogApp

WaterlogApp(settings: WaterlogSettings)

Bases: ExecutableApp

App for logging water consumption and related information.

Initialize the WaterlogApp with the given settings.

Source code in src/clabe/aind_apps/waterlog.py
40
41
42
43
44
45
46
def __init__(self, settings: WaterlogSettings):
    """Initialize the WaterlogApp with the given settings."""
    self._executable = str(self._EXECUTABLE)
    self._settings = settings
    self.validate()
    _cmd = [self._executable] + CliApp.serialize(settings)
    self._command = StdCommand(cmd=_cmd)

command property

command: StdCommand

Get the command to execute.

validate

validate() -> None

Validates the settings and checks for the presence of the waterlog executable.

Source code in src/clabe/aind_apps/waterlog.py
48
49
50
51
52
53
54
55
def validate(self) -> None:
    """Validates the settings and checks for the presence of the waterlog executable."""
    if not Path(self._executable).exists():
        if (loc := shutil.which("waterlog.exe")) is None:
            raise FileNotFoundError(
                f"'{self._EXECUTABLE}' command not found. Please ensure it is installed and available."
            )
        self._executable = loc

run

run(
    executor_kwargs: dict[str, Any] | None = None,
) -> CommandResult

Execute the command using a local executor and return the result.

Source code in src/clabe/aind_apps/waterlog.py
62
63
64
65
66
@runnable
def run(self: Self, executor_kwargs: dict[str, Any] | None = None) -> CommandResult:
    """Execute the command using a local executor and return the result."""
    executor = LocalDetachedExecutor(**(executor_kwargs or {}))
    return self.command.execute(executor)