Config

The config module provides configuration handling for Kelvin applications.

Kelvin Application Configuration.

This module provides configuration management for Kelvin applications, including YAML configuration loading and Pydantic-based settings.

Main Components:

KelvinAppConfig: Base configuration class with YAML and environment variable support. load_kelvin_yaml_config: Helper function to load configuration from YAML files.

App Configuration

kelvin.config.appconfig.load_kelvin_yaml_config(paths=None)[source]

Load Kelvin YAML configuration from the first existing file in paths.

Default search order when paths is None:

  • /opt/kelvin/share/config.yaml

  • /opt/kelvin/share/app.yaml

  • config.yaml

  • app.yaml

Extraction rules:

  • If filename is ‘config.yaml’ (local or under /opt/kelvin/share), use the YAML root mapping.

  • Otherwise, try nested:

    app.docker.configuration

    then fall back to:

    defaults.configuration

Return type:

dict[str, Any]

Returns:

dict[str, Any] with configuration values. Empty dict if nothing found or usable.

Parameters:

paths (tuple[Path, ...] | None)

class kelvin.config.appconfig.KelvinYamlConfigSettingsSource(settings_cls)[source]

Bases: PydanticBaseSettingsSource

A Pydantic Settings Source that reads Kelvin-style YAML configuration files.

Parameters:

settings_cls (type[BaseSettings])

get_field_value(field, field_name)[source]

Gets the value, the key for model creation, and a flag to determine whether value is complex.

This is an abstract method that should be overridden in every settings source classes.

Parameters:

  • field (FieldInfo) – The field.

  • field_name (str) – The field name.

Return type:

tuple[Any, str, bool]

Returns:

A tuple that contains the value, key and a flag to determine whether value is complex.

class kelvin.config.appconfig.KelvinAppConfig(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_prefix_target=None, _env_file=PosixPath('.'), _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, _build_sources=None, **values)[source]

Bases: BaseSettings

Kelvin application base configuration.

Load order and override precedence:

  1. init kwargs

  2. environment variables

  3. .env file

  4. YAML configuration (first match in YAML_PATHS or defaults)

  5. file secrets

This means values from environment variables or .env override values from YAML.

YAML discovery:
  • Reads from the first existing file among:

    • /opt/kelvin/share/config.yaml

    • /opt/kelvin/share/app.yaml

    • config.yaml

    • app.yaml

  • If the file name is config.yaml (local or under /opt/kelvin/share), the entire YAML root is used.

  • Otherwise, the loader extracts:

    app.docker.configuration

    or falls back to:

    defaults.configuration

Nested environment variables:

  • Set env_nested_delimiter=”__” so nested fields can be overridden. Example: MQTT__PASSWORD=secret overrides mqtt.password.

Customization:

  • Override YAML_PATHS on your subclass to change search locations.

  • Adjust model_config.env_prefix to fit your application.

Parameters:

  • _case_sensitive (bool | None)

  • _nested_model_default_partial_update (bool | None)

  • _env_prefix (str | None)

  • _env_prefix_target (EnvPrefixTarget | None)

  • _env_file (DotenvType | None)

  • _env_file_encoding (str | None)

  • _env_ignore_empty (bool | None)

  • _env_nested_delimiter (str | None)

  • _env_nested_max_split (int | None)

  • _env_parse_none_str (str | None)

  • _env_parse_enums (bool | None)

  • _cli_prog_name (str | None)

  • _cli_parse_args (bool | list[str] | tuple[str, ...] | None)

  • _cli_settings_source (CliSettingsSource[Any] | None)

  • _cli_parse_none_str (str | None)

  • _cli_hide_none_type (bool | None)

  • _cli_avoid_json (bool | None)

  • _cli_enforce_required (bool | None)

  • _cli_use_class_docs_for_groups (bool | None)

  • _cli_exit_on_error (bool | None)

  • _cli_prefix (str | None)

  • _cli_flag_prefix_char (str | None)

  • _cli_implicit_flags (bool | Literal['dual', 'toggle'] | None)

  • _cli_ignore_unknown_args (bool | None)

  • _cli_kebab_case (bool | Literal['all', 'no_enums'] | None)

  • _cli_shortcuts (Mapping[str, str | list[str]] | None)

  • _secrets_dir (PathType | None)

  • _build_sources (tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None)

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': '', 'env_prefix_target': 'variable', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}

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

YAML_PATHS: ClassVar[tuple[Path, ...]] = (PosixPath('/opt/kelvin/share/config.yaml'), PosixPath('/opt/kelvin/share/app.yaml'), PosixPath('config.yaml'), PosixPath('app.yaml'))
classmethod settings_customise_sources(settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings)[source]

Define the sources and their order for loading the settings values.

Parameters:

  • settings_cls (type[BaseSettings]) – The Settings class.

  • init_settings (PydanticBaseSettingsSource) – The InitSettingsSource instance.

  • env_settings (PydanticBaseSettingsSource) – The EnvSettingsSource instance.

  • dotenv_settings (PydanticBaseSettingsSource) – The DotEnvSettingsSource instance.

  • file_secret_settings (PydanticBaseSettingsSource) – The SecretsSettingsSource instance.

Return type:

tuple[PydanticBaseSettingsSource, ...]

Returns:

A tuple containing the sources and their order for loading the settings values.

App YAML

exception kelvin.config.appyaml.ConfigurationError[source]

Bases: Exception

class kelvin.config.appyaml.MetricInfo(**data)[source]

Bases: BaseModel

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

asset_names: list[str] | None
class kelvin.config.appyaml.Metric(**data)[source]

Bases: BaseModel

Parameters:

  • name (str)

  • data_type (str)

  • control_change (bool)

  • extra_data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

name: str
data_type: str
control_change: bool
class kelvin.config.appyaml.ParameterDefinition(**data)[source]

Bases: BaseModel

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

name: str
data_type: str
default: dict[str, Any] | None
class kelvin.config.appyaml.MetricInput(**data)[source]

Bases: Metric

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

sources: list[MetricInfo] | None
class kelvin.config.appyaml.MetricOutput(**data)[source]

Bases: Metric

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

targets: list[MetricInfo] | None
class kelvin.config.appyaml.AssetsEntry(**data)[source]

Bases: BaseModel

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

name: str
parameters: dict[str, Any]
properties: dict[str, Any]
class kelvin.config.appyaml.MetricMap(**data)[source]

Bases: BaseModel

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

name: str
asset_name: str
data_type: str
access: str
configuration: dict[str, Any]
class kelvin.config.appyaml.AppKelvin(**data)[source]

Bases: BaseModel

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

assets: list[AssetsEntry]
inputs: list[Metric]
outputs: list[Metric]
parameters: list[ParameterDefinition]
configuration: dict[str, Any]
class kelvin.config.appyaml.AppBridge(**data)[source]

Bases: BaseModel

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

metrics_map: list[MetricMap]
configuration: dict[str, Any]
class kelvin.config.appyaml.AppDocker(**data)[source]

Bases: ConfigBaseModel

Parameters:

extra_data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.appyaml.EnvironmentConfig(**data)[source]

Bases: ConfigBaseModel

Parameters:

  • workload_name (str)

  • node_name (str)

workload_name: str
node_name: str
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.appyaml.AppConfig(**data)[source]

Bases: ConfigBaseModel

Parameters:
type: Literal[AppTypes.bridge, AppTypes.kelvin_app, AppTypes.docker]
kelvin: AppKelvin | None
bridge: AppBridge | None
docker: AppDocker | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.appyaml.AppYamlInfo(**data)[source]

Bases: ConfigBaseModel

Parameters:
name: str
title: str
description: str
version: str
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.appyaml.SystemConfig(**data)[source]

Bases: ConfigBaseModel

Parameters:

  • privileged (bool)

  • extra_data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

privileged: bool
class kelvin.config.appyaml.AppYaml(**data)[source]

Bases: ConfigBaseModel

Parameters:
spec_version: str
environment: EnvironmentConfig | None
info: AppYamlInfo
app: AppConfig
system: SystemConfig | None
to_manifest(workdir, read_schemas)[source]
Return type:

AppManifest

Parameters:
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

Common

exception kelvin.config.common.ConfigError[source]

Bases: Exception

class kelvin.config.common.ConfigBaseModel(**data)[source]

Bases: BaseModel

Base model for configuration classes with sensible serialization defaults.

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

model_dump(*, mode='json', by_alias=True, exclude_unset=True, exclude_none=True, **kwargs)[source]

Serialize model with config-friendly defaults.

Changes from BaseModel defaults: - mode: “json” (was “python”) - by_alias: True (was False) - exclude_unset: True (was False) - exclude_none: True (was False)

Return type:

dict[str, Any]

Parameters:

  • mode (Literal['json', 'python'] | str)

  • by_alias (bool | None)

  • exclude_unset (bool)

  • exclude_none (bool)

  • kwargs (Any)

class kelvin.config.common.AppTypes(*values)[source]

Bases: str, Enum

importer = 'importer'
exporter = 'exporter'
app = 'app'
docker = 'docker'
kelvin_app = 'kelvin'
bridge = 'bridge'
legacy_docker = 'legacy_docker'
class kelvin.config.common.PrimitiveTypes(*values)[source]

Bases: str, Enum

number = 'number'
string = 'string'
boolean = 'boolean'
object = 'object'
class kelvin.config.common.CustomActionDef(**data)[source]

Bases: ConfigBaseModel

Parameters:

type (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-zA-Z0-9]([-_ .a-zA-Z0-9]*[a-zA-Z0-9])?$)])

type: a-zA-Z0-9]*[a-zA-Z0-9])?$)]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.common.CustomActionsIO(**data)[source]

Bases: ConfigBaseModel

Parameters:
inputs: list[CustomActionDef]
outputs: list[CustomActionDef]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.common.AppBaseConfig(**data)[source]

Bases: ConfigBaseModel

Parameters:

  • name (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-z0-9]([-a-z0-9]*[a-z0-9])?$)])

  • title (str)

  • description (str)

  • type (AppTypes)

  • version (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^([0-9]+)\.([0-9]+)\.([0-9]+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)])

  • category (str | None)

name: StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-z0-9]([-a-z0-9]*[a-z0-9])?$)]
title: str
description: str
type: AppTypes
version: [0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)]
category: str | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

kelvin.config.common.read_schema_file(file_path)[source]
Return type:

dict[str, Any]

Parameters:

file_path (Path)

kelvin.config.common.resolve_schema_path(workdir, schema_path)[source]

Resolve a schema path under workdir, preventing path traversal.

Return type:

Path

Parameters:

Exporter

class kelvin.config.exporter.RuntimeUpdateConfig(**data)[source]

Bases: ConfigBaseModel

Controls which parts of the exporter can be updated at runtime.

Parameters:

configuration (bool)

configuration: bool
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.exporter.ExporterFlags(**data)[source]

Bases: ConfigBaseModel

Feature flags for exporter behavior.

Parameters:
enable_runtime_update: RuntimeUpdateConfig
resources_required: bool | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.exporter.SchemasConfig(**data)[source]

Bases: ConfigBaseModel

Paths to JSON schema files for configuration validation.

Parameters:
configuration: str | None
io_configuration: dict[str, str]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.exporter.ExporterIO(**data)[source]

Bases: ConfigBaseModel

Defines a dynamic IO channel for the exporter.

Parameters:
name: str
data_types: list[str]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.exporter.DeploymentDefaults(**data)[source]

Bases: ConfigBaseModel

Default values applied when deploying the exporter.

Parameters:
system: dict[str, Any]
configuration: dict[str, Any]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.exporter.ExporterConfig(**data)[source]

Bases: AppBaseConfig

Complete configuration for an exporter application.

Parameters:

  • name (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-z0-9]([-a-z0-9]*[a-z0-9])?$)])

  • title (str)

  • description (str)

  • type (Literal[AppTypes.exporter])

  • version (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^([0-9]+)\.([0-9]+)\.([0-9]+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)])

  • category (str | None)

  • spec_version (str)

  • flags (ExporterFlags)

  • exporter_io (list[ExporterIO])

  • ui_schemas (SchemasConfig)

  • defaults (DeploymentDefaults)

  • custom_actions (CustomActionsIO)

  • api_permissions (list[str] | None)

type: Literal[AppTypes.exporter]
spec_version: str
flags: ExporterFlags
exporter_io: list[ExporterIO]
ui_schemas: SchemasConfig
defaults: DeploymentDefaults
custom_actions: CustomActionsIO
api_permissions: list[str] | None
to_manifest(workdir, read_schemas)[source]
Return type:

AppManifest

Parameters:
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

kelvin.config.exporter.convert_exporter_to_manifest(config, workdir, read_schemas)[source]

Convert an ExporterConfig to its AppManifest representation.

Return type:

AppManifest

Parameters:

External

class kelvin.config.external.RuntimeUpdateConfig(**data)[source]

Bases: ConfigBaseModel

Parameters:

configuration (bool)

configuration: bool
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.external.ExternalFlags(**data)[source]

Bases: ConfigBaseModel

Parameters:

enable_runtime_update (RuntimeUpdateConfig)

enable_runtime_update: RuntimeUpdateConfig
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.external.SchemasConfig(**data)[source]

Bases: ConfigBaseModel

Parameters:

configuration (str | None)

configuration: str | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.external.DeploymentDefaults(**data)[source]

Bases: ConfigBaseModel

Parameters:
system: dict[str, Any]
configuration: dict[str, Any]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.external.ExternalConfig(**data)[source]

Bases: AppBaseConfig

Represents the configuration for an external (docker) app.

Parameters:

  • name (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-z0-9]([-a-z0-9]*[a-z0-9])?$)])

  • title (str)

  • description (str)

  • type (Literal[AppTypes.docker])

  • version (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^([0-9]+)\.([0-9]+)\.([0-9]+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)])

  • category (str | None)

  • spec_version (str)

  • flags (ExternalFlags)

  • ui_schemas (SchemasConfig)

  • defaults (DeploymentDefaults)

  • api_permissions (list[str] | None)

type: Literal[AppTypes.docker]
spec_version: str
flags: ExternalFlags
ui_schemas: SchemasConfig
defaults: DeploymentDefaults
api_permissions: list[str] | None
to_manifest(workdir, read_schemas)[source]
Return type:

AppManifest

Parameters:
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

kelvin.config.external.convert_external_to_manifest(config, workdir, read_schemas)[source]

Converts an ExternalConfig object to an AppManifest object.

Parameters:

  • config (ExternalConfig) – The external app configuration.

  • read_schemas (bool) – Whether to read schema files.

  • workdir (Path) – The working directory to use for reading schema files.

Returns:

The generated app manifest.

Return type:

AppManifest

Importer

class kelvin.config.importer.RuntimeUpdateConfig(**data)[source]

Bases: ConfigBaseModel

Controls which parts of the importer can be updated at runtime.

Parameters:

configuration (bool)

configuration: bool
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.importer.ImporterFlags(**data)[source]

Bases: ConfigBaseModel

Feature flags for importer behavior.

Parameters:
enable_runtime_update: RuntimeUpdateConfig
resources_required: bool | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.importer.SchemasConfig(**data)[source]

Bases: ConfigBaseModel

Paths to JSON schema files for configuration validation.

Parameters:
configuration: str | None
io_configuration: dict[str, str]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.importer.ImporterIO(**data)[source]

Bases: ConfigBaseModel

Defines a dynamic IO channel for the importer.

Parameters:
name: str
data_types: list[str]
control: bool
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.importer.DeploymentDefaults(**data)[source]

Bases: ConfigBaseModel

Default values applied when deploying the importer.

Parameters:
system: dict[str, Any]
configuration: dict[str, Any]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.importer.ImporterConfig(**data)[source]

Bases: AppBaseConfig

Complete configuration for an importer application.

Parameters:

  • name (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-z0-9]([-a-z0-9]*[a-z0-9])?$)])

  • title (str)

  • description (str)

  • type (Literal[AppTypes.importer])

  • version (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^([0-9]+)\.([0-9]+)\.([0-9]+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)])

  • category (str | None)

  • spec_version (str)

  • flags (ImporterFlags)

  • importer_io (list[ImporterIO])

  • ui_schemas (SchemasConfig)

  • defaults (DeploymentDefaults)

  • custom_actions (CustomActionsIO)

  • api_permissions (list[str] | None)

type: Literal[AppTypes.importer]
spec_version: str
flags: ImporterFlags
importer_io: list[ImporterIO]
ui_schemas: SchemasConfig
defaults: DeploymentDefaults
custom_actions: CustomActionsIO
api_permissions: list[str] | None
to_manifest(workdir, read_schemas)[source]
Return type:

AppManifest

Parameters:
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

kelvin.config.importer.convert_importer_to_manifest(config, workdir, read_schemas)[source]

Convert an ImporterConfig to its AppManifest representation.

Return type:

AppManifest

Parameters:

Manifest

class kelvin.config.manifest.RuntimeUpdateFlags(**data)[source]

Bases: ConfigBaseModel

Parameters:

  • configuration (bool)

  • resource_parameters (bool)

  • resource_properties (bool)

configuration: bool
resource_parameters: bool
resource_properties: bool
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DeploymentFlags(**data)[source]

Bases: ConfigBaseModel

Parameters:

allowed_resources (list[KRN])

allowed_resources: list[KRN]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.Flags(**data)[source]

Bases: ConfigBaseModel

Parameters:

  • spec_version (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^([0-9]+)\.([0-9]+)\.([0-9]+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)])

  • enable_runtime_update (RuntimeUpdateFlags)

  • deployment (DeploymentFlags)

  • resources_required (bool | None)

spec_version: [0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)]
enable_runtime_update: RuntimeUpdateFlags
deployment: DeploymentFlags
resources_required: bool | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.IOWay(*values)[source]

Bases: str, Enum

output = 'output'
input_cc = 'input-cc'
input_cc_output = 'input-cc+output'
input = 'input'
output_cc = 'output-cc'
input_output_cc = 'input+output-cc'
class kelvin.config.manifest.DQWay(*values)[source]

Bases: str, Enum

output = 'output'
input = 'input'
class kelvin.config.manifest.IODefinition(**data)[source]

Bases: ConfigBaseModel

Parameters:

  • name (str)

  • data_type (str)

  • unit (str | None)

  • way (IOWay)

  • storage (IOStorage)

name: str
data_type: str
unit: str | None
way: IOWay
storage: IOStorage
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DynamicIoOwnership(*values)[source]

Bases: str, Enum

both = 'both'
owned = 'owned'
remote = 'remote'
class kelvin.config.manifest.DynamicIoType(*values)[source]

Bases: str, Enum

both = 'both'
data = 'data'
control = 'control'
class kelvin.config.manifest.DynamicIODataTypes(**data)[source]

Bases: ConfigBaseModel

Parameters:

name (str)

name: str
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DynamicIODefinition(**data)[source]

Bases: ConfigBaseModel

Parameters:
type_name: str
ownership: DynamicIoOwnership
type: DynamicIoType
data_types: list[str]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.CustomActionWay(*values)[source]

Bases: str, Enum

input_ca = 'input-ca'
output_ca = 'output-ca'
class kelvin.config.manifest.ManifCustomAction(**data)[source]

Bases: ConfigBaseModel

Parameters:

  • type (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^[a-zA-Z0-9]([-_ .a-zA-Z0-9]*[a-zA-Z0-9])?$)])

  • way (CustomActionWay)

type: a-zA-Z0-9]*[a-zA-Z0-9])?$)]
way: CustomActionWay
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DataTagWay(*values)[source]

Bases: str, Enum

input = 'input'
output = 'output'
class kelvin.config.manifest.ManifDataTag(**data)[source]

Bases: ConfigBaseModel

Parameters:
tag_name: str
way: DataTagWay
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.ParamDefinition(**data)[source]

Bases: ConfigBaseModel

Parameters:
name: str
title: str | None
data_type: Literal[PrimitiveTypes.number, PrimitiveTypes.string, PrimitiveTypes.boolean] | None
default: Annotated[bool, Strict(strict=True)] | Annotated[int, Strict(strict=True)] | Annotated[float, Strict(strict=True)] | Annotated[str, Strict(strict=True)] | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.IOSchema(**data)[source]

Bases: ConfigBaseModel

Parameters:
type_name: str
io_schema: dict[str, Any]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.SchemasDefinition(**data)[source]

Bases: ConfigBaseModel

Parameters:
parameters: dict[str, Any]
configuration: dict[str, Any]
io_configurations: list[IOSchema]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DeploymentType(*values)[source]

Bases: str, Enum

standard = 'standard'
staged_instant_apply = 'staged+instant-apply'
staged_only = 'staged-only'
class kelvin.config.manifest.ClusterDefinition(**data)[source]

Bases: ConfigBaseModel

Parameters:

name (str)

name: str
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DeploymentTargetDefaults(**data)[source]

Bases: ConfigBaseModel

Parameters:
type: str | None
cluster: ClusterDefinition | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DeploymentDefaults(**data)[source]

Bases: ConfigBaseModel

Parameters:
max_resources: int | None
deployment_type: DeploymentType | None
target: DeploymentTargetDefaults | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.IODatastreamMapping(**data)[source]

Bases: ConfigBaseModel

Parameters:
io: str
datastream: str
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.AppDefaults(**data)[source]

Bases: ConfigBaseModel

Parameters:
configuration: dict[str, Any]
io_datastream_mapping: list[IODatastreamMapping] | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.DefaultsDefinition(**data)[source]

Bases: ConfigBaseModel

Parameters:
deployment: DeploymentDefaults | None
app: AppDefaults | None
system: dict[str, Any] | None
api_permissions: list[str] | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.ManifCustomDataQuality(**data)[source]

Bases: ConfigBaseModel

Parameters:
name: str
data_type: str
way: DQWay
datastreams: list[str]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.manifest.AppManifest(**data)[source]

Bases: AppBaseConfig

Parameters:
flags: Flags | None
io: list[IODefinition]
dynamic_io: list[DynamicIODefinition]
parameters: list[ParamDefinition]
schemas: SchemasDefinition | None
defaults: DefaultsDefinition | None
custom_actions: list[ManifCustomAction]
data_quality: list[ManifCustomDataQuality]
data_tags: list[ManifDataTag]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

Parser

kelvin.config.parser.missing_config_error(file_path)[source]
Return type:

str

Parameters:

file_path (str)

kelvin.config.parser.invalid_yaml_error(file_path)[source]
Return type:

str

Parameters:

file_path (str)

class kelvin.config.parser.AppConfigObj(name, version, type, config)[source]

Bases: object

Parameters:
name: str
version: str
type: AppTypes
config: ExporterConfig | ImporterConfig | SmartAppConfig | ExternalConfig | AppYaml
is_legacy()[source]
Return type:

bool

to_app_manifest(workdir, read_schemas=True)[source]
Return type:

AppManifest

Parameters:
kelvin.config.parser.parse_config_file(file_path)[source]

Parses a YAML configuration file and returns an AppConfigObj.

Parameters:

file_path (str) – The path to the configuration file.

Raises:

ConfigError – If the file does not exist, path is invalid, or contains invalid YAML.

Returns:

The parsed configuration object.

Return type:

AppConfigObj

kelvin.config.parser.parse_config(config)[source]
Return type:

AppConfigObj

Parameters:

config (dict[str, Any])

Smart App

class kelvin.config.smart_app.RuntimeUpdateConfig(**data)[source]

Bases: ConfigBaseModel

Controls which parts of the smart app can be updated at runtime.

Parameters:

  • configuration (bool)

  • parameters (bool)

  • resource_properties (bool)

configuration: bool
parameters: bool
resource_properties: bool
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.SmartAppFlags(**data)[source]

Bases: ConfigBaseModel

Feature flags for smart app behavior.

Parameters:

enable_runtime_update (RuntimeUpdateConfig)

enable_runtime_update: RuntimeUpdateConfig
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.IOConfig(**data)[source]

Bases: ConfigBaseModel

Defines a single IO channel with name, data type, and optional unit.

Parameters:

  • name (str)

  • data_type (str)

  • unit (str | None)

name: str
data_type: str
unit: str | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.DataIo(**data)[source]

Bases: ConfigBaseModel

Container for input and output IO configurations.

Parameters:
inputs: list[IOConfig]
outputs: list[IOConfig]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.DataQualityConfig(**data)[source]

Bases: ConfigBaseModel

Defines a data quality metric and its associated data streams.

Parameters:
name: str
data_type: str
data_streams: list[str]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.DataQualityIO(**data)[source]

Bases: ConfigBaseModel

Container for input and output data quality configurations.

Parameters:
inputs: list[DataQualityConfig]
outputs: list[DataQualityConfig]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.SmartAppParams(**data)[source]

Bases: ConfigBaseModel

Defines a smart app parameter with name and primitive data type.

Parameters:

  • name (str)

  • data_type (Literal[PrimitiveTypes.number, PrimitiveTypes.string, PrimitiveTypes.boolean])

name: str
data_type: Literal[PrimitiveTypes.number, PrimitiveTypes.string, PrimitiveTypes.boolean]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.SchemasConfig(**data)[source]

Bases: ConfigBaseModel

Paths to JSON schema files for configuration and parameters validation.

Parameters:

  • configuration (str | None)

  • parameters (str | None)

configuration: str | None
parameters: str | None
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.DatastreamMapping(**data)[source]

Bases: ConfigBaseModel

Maps an app IO to a datastream.

Parameters:
app: str
datastream: str
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.DeploymentDefaults(**data)[source]

Bases: ConfigBaseModel

Default values applied when deploying the smart app.

Parameters:
system: dict[str, Any]
datastream_mapping: list[DatastreamMapping]
configuration: dict[str, Any]
parameters: dict[str, Annotated[bool, Strict(strict=True)] | Annotated[int, Strict(strict=True)] | Annotated[float, Strict(strict=True)] | Annotated[str, Strict(strict=True)]]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.DataTagsIO(**data)[source]

Bases: ConfigBaseModel

Parameters:
inputs: list[str]
outputs: list[str]
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

class kelvin.config.smart_app.SmartAppConfig(**data)[source]

Bases: AppBaseConfig

Complete configuration for a smart app application.

Parameters:
type: Literal[AppTypes.app]
spec_version: str
flags: SmartAppFlags
data_streams: DataIo
control_changes: DataIo
data_quality: DataQualityIO
parameters: list[SmartAppParams]
ui_schemas: SchemasConfig
defaults: DeploymentDefaults
custom_actions: CustomActionsIO
api_permissions: list[str] | None
data_tags: DataTagsIO
to_manifest(workdir, read_schemas)[source]
Return type:

AppManifest

Parameters:
model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

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

kelvin.config.smart_app.convert_smart_app_to_manifest(config, workdir, read_schemas=True)[source]

Convert a SmartAppConfig to its AppManifest representation.

Return type:

AppManifest

Parameters: