Configuration

The otter.config package contains the configuration models, parsers and utilities.

config.model module

Models for the config.

otter.config.model.LOG_LEVELS

The log levels.

alias of Literal[‘TRACE’, ‘DEBUG’, ‘INFO’, ‘SUCCESS’, ‘WARNING’, ‘ERROR’, ‘CRITICAL’]

class otter.config.model.BaseConfig(*, runner_name: str | None = None, step: str | None = None, config_path: Path | None = None, work_path: Path | None = None, release_uri: Annotated[str, AfterValidator(func=_validate_uri)] | None = None, pool_size: Annotated[int, AfterValidator(func=_validate_pool_size)] | None = None, log_level: Literal['TRACE', 'DEBUG', 'INFO', 'SUCCESS', 'WARNING', 'ERROR', 'CRITICAL'] | None = None)

Bases: BaseModel

Base config model.

runner_name: str | None

step: str | None

config_path: Path | None

work_path: Path | None

release_uri: Annotated[str, AfterValidator(_validate_uri)] | None

pool_size: Annotated[int, AfterValidator(_validate_pool_size)] | None

log_level: LOG_LEVELS | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class otter.config.model.Defaultconfig(*, runner_name: str = 'otter', config_path: Path = PosixPath('config.yaml'), work_path: Path = PosixPath('output'), release_uri: str | None = None, pool_size: int = 5, log_level: Literal['TRACE', 'DEBUG', 'INFO', 'SUCCESS', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO')

Bases: BaseModel

Default config model.

runner_name: str

config_path: Path

work_path: Path

release_uri: str | None

pool_size: int

log_level: LOG_LEVELS

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class otter.config.model.YamlConfig(*, work_path: Path | None = None, release_uri: Annotated[str, AfterValidator(func=_validate_uri)] | None = None, pool_size: Annotated[int, AfterValidator(func=_validate_pool_size)] | None = None, log_level: Literal['TRACE', 'DEBUG', 'INFO', 'SUCCESS', 'WARNING', 'ERROR', 'CRITICAL'] | None = None, steps: list[str] = [])

Bases: BaseModel

YAML config model.

work_path: Path | None

release_uri: Annotated[str, AfterValidator(_validate_uri)] | None

pool_size: Annotated[int, AfterValidator(_validate_pool_size)] | None

log_level: LOG_LEVELS | None

steps: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class otter.config.model.Config(*, runner_name: str = 'otter', step: str, steps: list[str] = [], config_path: Path = PosixPath('config.yaml'), work_path: Path = PosixPath('output'), release_uri: Annotated[str, AfterValidator(func=_validate_uri)] | None = None, pool_size: Annotated[int, AfterValidator(func=_validate_pool_size)] = 5, log_level: Literal['TRACE', 'DEBUG', 'INFO', 'SUCCESS', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO')

Bases: BaseModel

Config model.

This model is used to define the config for the application.

It is constructed by merging the config from the defaults, env vars, CLI, and YAML config file. The config is loaded in order of precedence: 1. Command line arguments 2. Environment variables 3. YAML configuration file 4. Default settings

The model is frozen after validation to prevent further modification.

model_config = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runner_name: str

The name of the runner/application.

step: str

The step to run. This is a required field.

steps: list[str]

The list of steps defined in the configuration file.

config_path: Path

The path to the configuration file.

work_path: Path

The local working path. This is where resources will be downloaded and the manifest and logs will be written to before upload to the GCS bucket.

release_uri: Annotated[str, AfterValidator(_validate_uri)] | None

The release URI. If present, this is where resources, logs and manifest will be uploaded to.

pool_size: Annotated[int, AfterValidator(_validate_pool_size)]

The maximum number of workers in the pool where tasks will run.

log_level: LOG_LEVELS

See LOG_LEVELS.

config.cli module

CLI argument parser.

otter.config.cli.config_to_env(var: str, runner_name: str) → str

Convert a config variable name to its env var.

otter.config.cli.parse_cli(runner_name: str) → BaseConfig

Parse the command line arguments.

Parameters:

runner_name (str) – The name of the runner.

Returns:

The parsed command line arguments.

Return type:

BaseConfig

config.env module

Environment variable parser.

otter.config.env.env_to_config(env_var: str, env_prefix: str) → str

Convert an env var to its config variable name.

otter.config.env.parse_env(runner_name: str) → BaseConfig

Parses the environment variables and returns an BaseConfig object.

Parameters:

runner_name (str) – The name of the runner.

Returns:

The parsed environment variables.

Return type:

BaseConfig

config.yaml module

YAML config parser.

otter.config.yaml.parse_yaml(path: Path) → dict[str, Any]

Parse a yaml file.

Loads a yaml file, parses its content, and returns it as a dictionary. There is a cache, so the file is only read once.

Warning

If the file cannot be read or the yaml content cannot be parsed, the program will log an error and exit.

Parameters:

path (Path) – The path to the yaml file.

Returns:

The parsed yaml content.

Return type:

dict

Module contents

Configuration package.

otter.config.load_config(runner_name: str) → Config

Load the config.

See otter.config.model.Config for the Config object.

Parameters:

runner_name (str) – The name of the runner.

Returns:

The config object.

Return type:

Config