from __future__ import annotations
import json
from enum import Enum
from pathlib import Path
from typing import Any, ClassVar, Literal, Optional, cast
from pydantic import AfterValidator, BaseModel, ConfigDict, Field, StringConstraints
from typing_extensions import Annotated, override
NameDNS = Annotated[str, StringConstraints(pattern="^[a-z0-9]([-a-z0-9]*[a-z0-9])?$")]
NameV2 = Annotated[str, StringConstraints(pattern="^[a-zA-Z0-9]([-_ .a-zA-Z0-9]*[a-zA-Z0-9])?$")]
VersionStr = Annotated[
str,
StringConstraints(
pattern="^([0-9]+)\\.([0-9]+)\\.([0-9]+)(?:-([0-9A-Za-z-]+(?:\\.[0-9A-Za-z-]+)*))?(?:\\+[0-9A-Za-z-]+)?$"
),
]
DescriptionStr = Annotated[str, StringConstraints(max_length=256)]
CustomActionTypeStr = NameV2
RecommendationTypeStr = NameV2
[docs]
def validate_system_defaults(system: dict[str, Any]) -> dict[str, Any]:
"""Check what the app.yaml JSON schema cannot express about ``defaults.system``.
Volume names must be unique; the platform rejects duplicates, but only after
the image has been built and pushed. Anything else (missing or non-string
names, a non-list ``volumes``) is left to schema validation to report.
"""
volumes = cast(object, system.get("volumes"))
if isinstance(volumes, list):
seen: set[str] = set()
duplicate_names: set[str] = set()
for volume in cast(list[object], volumes):
name = cast(dict[str, object], volume).get("name") if isinstance(volume, dict) else None
if isinstance(name, str):
(duplicate_names if name in seen else seen).add(name)
duplicates = sorted(duplicate_names)
if duplicates:
raise ValueError(f"duplicate volume name(s): {', '.join(duplicates)}")
return system
#: ``defaults.system`` as written in app.yaml, with the checks the schema cannot make.
SystemDefaults = Annotated[dict[str, Any], AfterValidator(validate_system_defaults)]
[docs]
class ConfigError(Exception):
pass
[docs]
class ConfigBaseModel(BaseModel):
"""Base model for configuration classes with sensible serialization defaults."""
model_config: ClassVar[ConfigDict] = ConfigDict(populate_by_name=True)
[docs]
@override
def model_dump(
self,
*,
mode: Literal["json", "python"] | str = "json",
by_alias: Optional[bool] = True,
exclude_unset: bool = True,
exclude_none: bool = True,
**kwargs: Any,
) -> dict[str, Any]:
"""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 super().model_dump(
mode=mode,
by_alias=by_alias,
exclude_unset=exclude_unset,
exclude_none=exclude_none,
**kwargs,
)
[docs]
class AppTypes(str, Enum):
importer = "importer"
exporter = "exporter"
app = "app"
docker = "docker"
# legacy - deprecated
kelvin_app = "kelvin"
bridge = "bridge"
legacy_docker = "legacy_docker"
[docs]
class PrimitiveTypes(str, Enum):
number = "number"
string = "string"
boolean = "boolean"
object = "object"
[docs]
class DataQualityType(str, Enum):
"""User-facing data quality target, as declared in app.yaml.
Translated to the manifest's short form (``asset`` / ``ad``) during conversion.
"""
asset = "asset"
asset_datastream = "asset-datastream"
[docs]
class DataQualityConfig(ConfigBaseModel):
"""Defines a data quality metric, its target type and associated data streams."""
name: str
data_type: str
type: Optional[DataQualityType] = None
data_streams: list[str] = Field(default_factory=list)
[docs]
class DataQualityIO(ConfigBaseModel):
"""Container for input and output data quality configurations."""
inputs: list[DataQualityConfig] = Field(default_factory=list)
outputs: list[DataQualityConfig] = Field(default_factory=list)
[docs]
class CustomActionDef(ConfigBaseModel):
type: CustomActionTypeStr
[docs]
class CustomActionsIO(ConfigBaseModel):
inputs: list[CustomActionDef] = Field(default_factory=list)
outputs: list[CustomActionDef] = Field(default_factory=list)
[docs]
class AppBaseConfig(ConfigBaseModel):
name: NameDNS
title: str
description: DescriptionStr
type: AppTypes
version: VersionStr
category: Optional[str] = None
[docs]
def read_schema_file(file_path: Path) -> dict[str, Any]:
if not file_path.exists():
raise ConfigError(f"schema file {file_path} does not exist")
try:
with file_path.open(encoding="utf-8") as file:
content = file.read()
except OSError as exc:
raise ConfigError(f"failed to read schema file {file_path}: {exc}") from exc
if not content.strip():
return {}
try:
return json.loads(content)
except json.JSONDecodeError as exc:
raise ConfigError(f"schema file {file_path} contains invalid JSON") from exc
[docs]
def resolve_schema_path(workdir: Path, schema_path: str) -> Path:
"""Resolve a schema path under workdir, preventing path traversal."""
resolved_workdir = workdir.resolve()
candidate = Path(schema_path)
resolved_path = candidate.resolve() if candidate.is_absolute() else (resolved_workdir / candidate).resolve()
try:
_ = resolved_path.relative_to(resolved_workdir)
except ValueError as exc:
raise ConfigError(f"schema path {schema_path} escapes workdir {workdir}") from exc
if not resolved_path.exists():
raise ConfigError(f"schema file {schema_path} does not exist")
return resolved_path