from __future__ import annotations
from enum import Enum
from pathlib import Path
from typing import Any, Literal, Optional
from pydantic import Field
from kelvin.config.common import (
AppBaseConfig,
ConfigBaseModel,
ConfigError,
CustomActionsIO,
CustomActionTypeStr,
DataQualityConfig,
DataQualityIO,
DataQualityType,
RecommendationTypeStr,
VersionStr,
read_schema_file,
resolve_schema_path,
)
from kelvin.krn import KRN
from kelvin.message import ParameterType
from kelvin.message.msg_type import PrimitiveTypes
from kelvin.message.runtime_manifest import IOStorage
[docs]
class RuntimeUpdateFlags(ConfigBaseModel):
configuration: bool = False
resource_parameters: bool = True
resource_properties: bool = True
[docs]
class DeploymentFlags(ConfigBaseModel):
allowed_resources: list[KRN] = Field(default_factory=list)
[docs]
class Flags(ConfigBaseModel):
spec_version: VersionStr
enable_runtime_update: RuntimeUpdateFlags = RuntimeUpdateFlags()
deployment: DeploymentFlags = DeploymentFlags()
resources_required: Optional[bool] = None
[docs]
class IOWay(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"
[docs]
class DQWay(str, Enum):
output = "output"
input = "input"
[docs]
class DQTarget(str, Enum):
"""Manifest data quality target. Mirrors the server's short enum form."""
asset = "asset"
ad = "ad"
[docs]
class IODefinition(ConfigBaseModel):
name: str
data_type: str
unit: Optional[str]
way: IOWay = IOWay.output
storage: IOStorage = IOStorage.node_and_cloud
[docs]
class DynamicIoOwnership(str, Enum):
both = "both"
owned = "owned"
remote = "remote"
[docs]
class DynamicIoType(str, Enum):
both = "both"
data = "data"
control = "control"
[docs]
class DynamicIODataTypes(ConfigBaseModel):
name: str
[docs]
class DynamicIODefinition(ConfigBaseModel):
type_name: str
ownership: DynamicIoOwnership = DynamicIoOwnership.both
type: DynamicIoType = DynamicIoType.both
data_types: list[str] = Field(default_factory=list)
[docs]
class CustomActionWay(str, Enum):
input_ca = "input-ca"
output_ca = "output-ca"
[docs]
class ManifCustomAction(ConfigBaseModel):
type: CustomActionTypeStr
way: CustomActionWay
schema_: Optional[dict[str, Any]] = Field(default=None, alias="schema")
[docs]
def build_custom_actions(
custom_actions: CustomActionsIO,
ui_schemas: dict[str, str],
workdir: Path,
read_schemas: bool,
) -> list[ManifCustomAction]:
"""Convert input/output custom actions to manifest format, embedding UI schemas.
``ui_schemas`` maps a custom action type to a JSON schema file path. A schema
applies to every action matching that type, regardless of way. Actions without
a schema (or when ``read_schemas`` is False) get ``schema=None`` (key omitted).
Schema keys are always validated against declared action types; files are only
read from disk when ``read_schemas`` is True.
Raises:
ConfigError: if a ``ui_schemas`` key does not match any declared action type.
"""
declared_types = {action.type for action in custom_actions.inputs}
declared_types |= {action.type for action in custom_actions.outputs}
schemas: dict[str, dict[str, Any]] = {}
for action_type, schema_path in ui_schemas.items():
if action_type not in declared_types:
raise ConfigError(f"custom action schema references unknown action type {action_type!r}")
if read_schemas:
resolved = resolve_schema_path(workdir, schema_path)
schemas[action_type] = read_schema_file(resolved)
input_actions = [
ManifCustomAction(type=action.type, way=CustomActionWay.input_ca, schema=schemas.get(action.type))
for action in custom_actions.inputs
]
output_actions = [
ManifCustomAction(type=action.type, way=CustomActionWay.output_ca, schema=schemas.get(action.type))
for action in custom_actions.outputs
]
return input_actions + output_actions
[docs]
class RecWay(str, Enum):
output = "output"
[docs]
class ManifRecommendation(ConfigBaseModel):
type: RecommendationTypeStr
way: RecWay = RecWay.output
[docs]
class DataTagWay(str, Enum):
input = "input"
output = "output"
[docs]
class ManifDataTag(ConfigBaseModel):
tag_name: str
way: DataTagWay
ParameterDataTypes = Literal[PrimitiveTypes.number, PrimitiveTypes.string, PrimitiveTypes.boolean]
[docs]
class ParamDefinition(ConfigBaseModel):
name: str
title: Optional[str] = None
data_type: Optional[ParameterDataTypes] = None
default: Optional[ParameterType] = None
[docs]
class IOSchema(ConfigBaseModel):
type_name: str
io_schema: dict[str, Any] = Field(default_factory=dict, alias="schema")
[docs]
class SchemasDefinition(ConfigBaseModel):
parameters: dict[str, Any] = Field(default_factory=dict)
configuration: dict[str, Any] = Field(default_factory=dict)
io_configurations: list[IOSchema] = Field(default_factory=list)
[docs]
class DeploymentType(str, Enum):
standard = "standard"
staged_instant_apply = "staged+instant-apply"
staged_only = "staged-only"
[docs]
class ClusterDefinition(ConfigBaseModel):
name: str
[docs]
class DeploymentTargetDefaults(ConfigBaseModel):
type: Optional[str] = None
cluster: Optional[ClusterDefinition] = None
[docs]
class DeploymentDefaults(ConfigBaseModel):
max_resources: Optional[int] = None
deployment_type: Optional[DeploymentType] = None
target: Optional[DeploymentTargetDefaults] = None
[docs]
class IODatastreamMapping(ConfigBaseModel):
io: str
datastream: str
[docs]
class AppDefaults(ConfigBaseModel):
configuration: dict[str, Any] = Field(default_factory=dict)
io_datastream_mapping: Optional[list[IODatastreamMapping]] = None
[docs]
class DefaultsDefinition(ConfigBaseModel):
deployment: Optional[DeploymentDefaults] = None
app: Optional[AppDefaults] = None
system: Optional[dict[str, Any]] = None
api_permissions: Optional[list[str]] = None
[docs]
class ManifCustomDataQuality(ConfigBaseModel):
name: str
data_type: str
way: DQWay
type: DQTarget = DQTarget.asset
datastreams: list[str] = Field(default_factory=list)
[docs]
def build_data_quality(data_quality: DataQualityIO) -> list[ManifCustomDataQuality]:
"""Convert input/output data quality definitions to manifest format.
Translates the user-facing ``type`` (``asset`` / ``asset-datastream``) to the
manifest's short form (``asset`` / ``ad``). When ``type`` is omitted, it is
inferred for backwards compatibility: ``ad`` if data streams are declared,
otherwise ``asset`` (matching the server's behaviour).
Raises:
ConfigError: if data streams are declared with ``type: asset``.
"""
def _resolve_target(dq: DataQualityConfig) -> DQTarget:
if dq.type is None:
return DQTarget.ad if dq.data_streams else DQTarget.asset
if dq.type == DataQualityType.asset_datastream:
return DQTarget.ad
# type == asset
if dq.data_streams:
raise ConfigError(f"data quality {dq.name} declares data streams but has type 'asset'")
return DQTarget.asset
def _convert(dq: DataQualityConfig, way: DQWay) -> ManifCustomDataQuality:
return ManifCustomDataQuality(
name=dq.name,
data_type=dq.data_type,
way=way,
type=_resolve_target(dq),
datastreams=dq.data_streams,
)
return [_convert(dq, DQWay.input) for dq in data_quality.inputs] + [
_convert(dq, DQWay.output) for dq in data_quality.outputs
]
[docs]
class AppManifest(AppBaseConfig):
flags: Optional[Flags] = None
io: list[IODefinition] = Field(default_factory=list)
dynamic_io: list[DynamicIODefinition] = Field(default_factory=list)
parameters: list[ParamDefinition] = Field(default_factory=list)
schemas: Optional[SchemasDefinition] = None
defaults: Optional[DefaultsDefinition] = None
custom_actions: list[ManifCustomAction] = Field(default_factory=list)
recommendations: list[ManifRecommendation] = Field(default_factory=list)
data_quality: list[ManifCustomDataQuality] = Field(default_factory=list)
data_tags: list[ManifDataTag] = Field(default_factory=list)