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
- class kelvin.config.appconfig.KelvinYamlConfigSettingsSource(settings_cls)[source]¶
Bases:
PydanticBaseSettingsSourceA Pydantic Settings Source that reads Kelvin-style YAML configuration files.
- Parameters:
settings_cls (Type[BaseSettings])
- 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:
BaseSettingsKelvin application base configuration.
- Load order and override precedence:
init kwargs
environment variables
.env file
YAML configuration (first match in YAML_PATHS or defaults)
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_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)
_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:
- Returns:
A tuple containing the sources and their order for loading the settings values.
App YAML¶
- class kelvin.config.appyaml.MetricInfo(**data)[source]¶
Bases:
BaseModel
- class kelvin.config.appyaml.Metric(**data)[source]¶
Bases:
BaseModel
- class kelvin.config.appyaml.ParameterDefinition(**data)[source]¶
Bases:
BaseModel
- 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
- class kelvin.config.appyaml.MetricMap(**data)[source]¶
Bases:
BaseModel- Parameters:
- class kelvin.config.appyaml.AppKelvin(**data)[source]¶
Bases:
BaseModel- Parameters:
assets (List[AssetsEntry])
parameters (List[ParameterDefinition])
configuration (Dict)
extra_data (Any)
- model_config: ClassVar[ConfigDict] = {'extra': 'allow'}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- assets: List[AssetsEntry]¶
- parameters: List[ParameterDefinition]¶
- class kelvin.config.appyaml.AppBridge(**data)[source]¶
Bases:
BaseModel
- class kelvin.config.appyaml.AppDocker(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
extra_data (Any)
- class kelvin.config.appyaml.EnvironmentConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.appyaml.AppConfig(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
- class kelvin.config.appyaml.AppYamlInfo(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.appyaml.SystemConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.appyaml.AppYaml(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
spec_version (str)
environment (EnvironmentConfig | None)
info (AppYamlInfo)
app (AppConfig)
system (SystemConfig | None)
- environment: EnvironmentConfig | None¶
- info: AppYamlInfo¶
- system: SystemConfig | None¶
Common¶
- class kelvin.config.common.ConfigBaseModel(**data)[source]¶
Bases:
BaseModel- 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_defaults=False, exclude_none=True, serialize_as_any=False, **kwargs)[source]¶
- !!! abstract “Usage Documentation”
[model_dump](../concepts/serialization.md#modelmodel_dump)
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
- Parameters:
mode (
Union[Literal['json','python'],str]) – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.include – A set of fields to include in the output.
exclude – A set of fields to exclude from the output.
context – Additional context to pass to the serializer.
by_alias (
bool) – Whether to use the field’s alias in the dictionary key if defined.exclude_unset (
bool) – Whether to exclude fields that have not been explicitly set.exclude_defaults (
bool) – Whether to exclude fields that are set to their default value.exclude_none (
bool) – Whether to exclude fields that have a value of None.round_trip – If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
fallback – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.
serialize_as_any (
bool) – Whether to serialize fields with duck-typing serialization behavior.kwargs (Any)
- Return type:
- Returns:
A dictionary representation of the model.
- class kelvin.config.common.AppTypes(*values)[source]¶
-
- importer = 'importer'¶
- exporter = 'exporter'¶
- app = 'app'¶
- docker = 'docker'¶
- kelvin_app = 'kelvin'¶
- bridge = 'bridge'¶
- legacy_docker = 'legacy_docker'¶
- class kelvin.config.common.PrimitiveTypes(*values)[source]¶
-
- 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])?$)]¶
- class kelvin.config.common.CustomActionsIO(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
inputs (List[CustomActionDef])
outputs (List[CustomActionDef])
- inputs: List[CustomActionDef]¶
- outputs: List[CustomActionDef]¶
- 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])?$)]¶
- version: [0-9A-Za-z-]+)*))?(?:\+[0-9A-Za-z-]+)?$)]¶
Exporter¶
- class kelvin.config.exporter.RuntimeUpdateConfig(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
configuration (bool)
- class kelvin.config.exporter.ExporterFlags(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
enable_runtime_update (RuntimeUpdateConfig)
resources_required (bool | None)
- enable_runtime_update: RuntimeUpdateConfig¶
- class kelvin.config.exporter.SchemasConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.exporter.ExporterIO(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.exporter.DeploymentDefaults(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.exporter.ExporterConfig(**data)[source]¶
Bases:
AppBaseConfig- 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)
- flags: ExporterFlags¶
- exporter_io: List[ExporterIO]¶
- ui_schemas: SchemasConfig¶
- defaults: DeploymentDefaults¶
- custom_actions: CustomActionsIO¶
- kelvin.config.exporter.convert_exporter_to_manifest(config, read_schemas=True, workdir=PosixPath('.'))[source]¶
- Return type:
- Parameters:
config (ExporterConfig)
read_schemas (bool)
workdir (Path)
External¶
- class kelvin.config.external.RuntimeUpdateConfig(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
configuration (bool)
- class kelvin.config.external.ExternalFlags(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
enable_runtime_update (RuntimeUpdateConfig)
- enable_runtime_update: RuntimeUpdateConfig¶
- class kelvin.config.external.SchemasConfig(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
configuration (str | None)
- class kelvin.config.external.DeploymentDefaults(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.external.ExternalConfig(**data)[source]¶
Bases:
AppBaseConfigRepresents 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)
- flags: ExternalFlags¶
- ui_schemas: SchemasConfig¶
- defaults: DeploymentDefaults¶
- kelvin.config.external.convert_external_to_manifest(config, read_schemas=True, workdir=PosixPath('.'))[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:
Importer¶
- class kelvin.config.importer.RuntimeUpdateConfig(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
configuration (bool)
- class kelvin.config.importer.ImporterFlags(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
enable_runtime_update (RuntimeUpdateConfig)
resources_required (bool | None)
- enable_runtime_update: RuntimeUpdateConfig¶
- class kelvin.config.importer.SchemasConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.importer.ImporterIO(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.importer.DeploymentDefaults(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.importer.ImporterConfig(**data)[source]¶
Bases:
AppBaseConfig- 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)
- flags: ImporterFlags¶
- importer_io: List[ImporterIO]¶
- ui_schemas: SchemasConfig¶
- defaults: DeploymentDefaults¶
- custom_actions: CustomActionsIO¶
- kelvin.config.importer.convert_importer_to_manifest(config, read_schemas=True, workdir=PosixPath('.'))[source]¶
- Return type:
- Parameters:
config (ImporterConfig)
read_schemas (bool)
workdir (Path)
Manifest¶
- class kelvin.config.manifest.RuntimeUpdateFlags(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.manifest.DeploymentFlags(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
allowed_resources (List[KRN])
- 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¶
- class kelvin.config.manifest.IOWay(*values)[source]¶
-
- 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.IODefinition(**data)[source]¶
Bases:
ConfigBaseModel- storage: IOStorage¶
- class kelvin.config.manifest.DynamicIoOwnership(*values)[source]¶
-
- both = 'both'¶
- owned = 'owned'¶
- remote = 'remote'¶
- class kelvin.config.manifest.DynamicIoType(*values)[source]¶
-
- both = 'both'¶
- data = 'data'¶
- control = 'control'¶
- class kelvin.config.manifest.DynamicIODataTypes(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
name (str)
- class kelvin.config.manifest.DynamicIODefinition(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
type_name (str)
ownership (DynamicIoOwnership)
type (DynamicIoType)
- ownership: DynamicIoOwnership¶
- type: DynamicIoType¶
- class kelvin.config.manifest.CustomActionWay(*values)[source]¶
-
- 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¶
- class kelvin.config.manifest.ParamDefinition(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
- class kelvin.config.manifest.IOSchema(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.manifest.SchemasDefinition(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.manifest.DeploymentType(*values)[source]¶
-
- standard = 'standard'¶
- staged_instant_apply = 'staged+instant-apply'¶
- staged_only = 'staged-only'¶
- class kelvin.config.manifest.ClusterDefinition(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
name (str)
- class kelvin.config.manifest.DeploymentTargetDefaults(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
type (str | None)
cluster (ClusterDefinition | None)
- cluster: ClusterDefinition | None¶
- class kelvin.config.manifest.DeploymentDefaults(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
max_resources (int | None)
deployment_type (DeploymentType | None)
target (DeploymentTargetDefaults | None)
- deployment_type: DeploymentType | None¶
- target: DeploymentTargetDefaults | None¶
- class kelvin.config.manifest.IODatastreamMapping(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.manifest.AppDefaults(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
io_datastream_mapping (List[IODatastreamMapping] | None)
- io_datastream_mapping: List[IODatastreamMapping] | None¶
- class kelvin.config.manifest.DefaultsDefinition(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
deployment (DeploymentDefaults | None)
app (AppDefaults | None)
system (dict | None)
- deployment: DeploymentDefaults | None¶
- app: AppDefaults | None¶
- class kelvin.config.manifest.ManifCustomDataQuality(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.manifest.AppManifest(**data)[source]¶
Bases:
AppBaseConfig- 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)
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])
- io: List[IODefinition]¶
- dynamic_io: List[DynamicIODefinition]¶
- parameters: List[ParamDefinition]¶
- schemas: SchemasDefinition | None¶
- defaults: DefaultsDefinition | None¶
- custom_actions: List[ManifCustomAction]¶
- data_quality: List[ManifCustomDataQuality]¶
Parser¶
- 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)
- config: ExporterConfig | ImporterConfig | SmartAppConfig | ExternalConfig | AppYaml¶
- 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:
Smart App¶
- class kelvin.config.smart_app.RuntimeUpdateConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.smart_app.SmartAppFlags(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
enable_runtime_update (RuntimeUpdateConfig)
- enable_runtime_update: RuntimeUpdateConfig¶
- class kelvin.config.smart_app.IOConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.smart_app.DataIo(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.smart_app.DataQualityConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.smart_app.DataQualityIO(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
inputs (List[DataQualityConfig])
outputs (List[DataQualityConfig])
- inputs: List[DataQualityConfig]¶
- outputs: List[DataQualityConfig]¶
- class kelvin.config.smart_app.SmartAppParams(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
- class kelvin.config.smart_app.SchemasConfig(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.smart_app.DatastreamMapping(**data)[source]¶
Bases:
ConfigBaseModel
- class kelvin.config.smart_app.DeploymentDefaults(**data)[source]¶
Bases:
ConfigBaseModel- Parameters:
- datastream_mapping: List[DatastreamMapping]¶
- class kelvin.config.smart_app.SmartAppConfig(**data)[source]¶
Bases:
AppBaseConfig- 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.app])
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 (SmartAppFlags)
data_streams (DataIo)
control_changes (DataIo)
data_quality (DataQualityIO)
parameters (List[SmartAppParams])
ui_schemas (SchemasConfig)
defaults (DeploymentDefaults)
custom_actions (CustomActionsIO)
- flags: SmartAppFlags¶
- data_quality: DataQualityIO¶
- parameters: List[SmartAppParams]¶
- ui_schemas: SchemasConfig¶
- defaults: DeploymentDefaults¶
- custom_actions: CustomActionsIO¶
- kelvin.config.smart_app.convert_smart_app_to_manifest(config, read_schemas=True, workdir=PosixPath('.'))[source]¶
- Return type:
- Parameters:
config (SmartAppConfig)
read_schemas (bool)
workdir (Path)