Source code for kelvin.config.common

from __future__ import annotations

import json
from enum import Enum
from pathlib import Path
from typing import Any, ClassVar, Literal, Optional

from pydantic import BaseModel, ConfigDict, Field, StringConstraints
from typing_extensions import Annotated, override

NameDNS = Annotated[str, StringConstraints(pattern="^[a-z0-9]([-a-z0-9]*[a-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-]+)?$"
    ),
]
CustomActionTypeStr = Annotated[str, StringConstraints(pattern="^[a-zA-Z0-9]([-_ .a-zA-Z0-9]*[a-zA-Z0-9])?$")]




class ConfigError(Exception):
    pass





class ConfigBaseModel(BaseModel):
    """Base model for configuration classes with sensible serialization defaults."""

    model_config: ClassVar[ConfigDict] = ConfigDict(populate_by_name=True)



    @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,
        )






class AppTypes(str, Enum):
    importer = "importer"
    exporter = "exporter"
    app = "app"
    docker = "docker"

    # legacy - deprecated
    kelvin_app = "kelvin"
    bridge = "bridge"
    legacy_docker = "legacy_docker"





class PrimitiveTypes(str, Enum):
    number = "number"
    string = "string"
    boolean = "boolean"
    object = "object"





class CustomActionDef(ConfigBaseModel):
    type: CustomActionTypeStr





class CustomActionsIO(ConfigBaseModel):
    inputs: list[CustomActionDef] = Field(default_factory=list)
    outputs: list[CustomActionDef] = Field(default_factory=list)





class AppBaseConfig(ConfigBaseModel):
    name: NameDNS
    title: str
    description: str
    type: AppTypes
    version: VersionStr
    category: Optional[str] = None





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





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