Source code for ska_sdp_piper.piper.piper_base_model

from copy import deepcopy

from pydantic import BaseModel, ConfigDict

from .piper_undefined import PIPER_UNDEFINED, PiperUndefined

__all__ = ("PIPER_MODEL_CONFIG", "PiperBaseModel")


PIPER_MODEL_CONFIG = ConfigDict(
    strict=False,
    extra="forbid",
    # To prevent serialisation issues in YAML config
    # we don't allow arbitrary types
    arbitrary_types_allowed=False,
    validate_assignment=True,
    validate_default=True,
)




class PiperBaseModel(BaseModel):
    """
    A base model class that enforces strict validation rules on all
    subclasses. This model extends Pydantic's BaseModel with custom
    initialization logic to ensure that all subclass fields have default
    values and properly handle undefined field markers.

    Validation Checks:

    - Ensures all fields in subclasses have default values assigned
    - Fields without defaults must be explicitly annotated with
      PIPER_UNDEFINED marker
    - Prevents fields from having both a default value AND
      PIPER_UNDEFINED annotation
    - Raises TypeError if any required fields (without defaults) are
      defined without PIPER_UNDEFINED annotation
    - Automatically wraps field annotations with serialize/validate
      methods when PIPER_UNDEFINED marker is detected in metadata

    All subclasses must conform to these constraints to be properly
    instantiated.
    """

    model_config = PIPER_MODEL_CONFIG

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)

        annotations = getattr(cls, "__annotations__", {})
        updated_annotations = deepcopy(annotations)

        for p_name, p_annot in annotations.items():
            metadata = getattr(p_annot, "__metadata__", [])

            if PIPER_UNDEFINED in metadata:
                assert p_name not in cls.__dict__, (
                    f"Argument '{p_name}' of model '{cls.__name__}' "
                    "has a default value set, and it "
                    f"is also annotated with {PIPER_UNDEFINED}. This is "
                    "not allowed as it may lead to undesired behavior."
                )
                updated_annotations[p_name] = PiperUndefined.annotate(p_annot)

        setattr(cls, "__annotations__", updated_annotations)

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs):
        super().__pydantic_init_subclass__(**kwargs)

        required_fields = [
            name
            for name, field in cls.model_fields.items()
            if field.is_required()
        ]
        if required_fields:
            raise TypeError(
                f"Model '{cls.__name__}' "
                f"has following fields without defaults: {required_fields}. "
                "All fields must have a default value. If a field cannot have "
                f"default value, annotate with {PIPER_UNDEFINED}. "
                "Please refer piper documentation."
            )