#
# Copyright 2018-2025 Elyra Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from __future__ import annotations

from abc import ABC
from abc import abstractmethod
from enum import Enum
from importlib import import_module
import json
import re
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from typing import TYPE_CHECKING
from typing import Union

from elyra.pipeline.pipeline_constants import DISABLE_NODE_CACHING
from elyra.pipeline.pipeline_constants import ENV_VARIABLES
from elyra.pipeline.pipeline_constants import KUBERNETES_POD_ANNOTATIONS
from elyra.pipeline.pipeline_constants import KUBERNETES_POD_LABELS
from elyra.pipeline.pipeline_constants import KUBERNETES_SECRETS
from elyra.pipeline.pipeline_constants import KUBERNETES_SHARED_MEM_SIZE
from elyra.pipeline.pipeline_constants import KUBERNETES_TOLERATIONS
from elyra.pipeline.pipeline_constants import MOUNTED_VOLUMES
from elyra.pipeline.pipeline_constants import PIPELINE_PARAMETERS
from elyra.pipeline.runtime_type import RuntimeProcessorType
from elyra.util.kubernetes import is_valid_annotation_key
from elyra.util.kubernetes import is_valid_annotation_value
from elyra.util.kubernetes import is_valid_kubernetes_key
from elyra.util.kubernetes import is_valid_kubernetes_resource_name
from elyra.util.kubernetes import is_valid_label_key
from elyra.util.kubernetes import is_valid_label_value

if TYPE_CHECKING:  # Prevent a circular reference by importing RuntimePipelineProcessor only during type-checking
    from elyra.pipeline.processor import RuntimePipelineProcessor


class PropertyInputType:
    """
    An object representing a single allowed input type for a PropertyAttribute
    object with optional custom default value, placeholder, and enum.
    """

    _python_defined_types = {
        "str": {"json_type": "string", "default_value": ""},
        "bool": {"json_type": "boolean", "default_value": False, "placeholder": " "},
        "int": {"json_type": "integer"},
        "float": {"json_type": "number"},
        "list": {"json_type": "string", "default_value": "[]"},
        "dict": {"json_type": "string", "default_value": "{}"},
    }

    def __init__(
        self,
        base_type: str,
        default_value: Optional[Any] = None,
        placeholder: Optional[Any] = None,
        enum: Optional[List[Any]] = None,
        **kwargs,
    ):
        # Get type info from passed in type dict, if available; default to standard python types
        allowed_input_types = kwargs.get("runtime_defined_types")
        if not allowed_input_types:
            allowed_input_types = self._python_defined_types

        if base_type not in allowed_input_types:
            raise ValueError(
                f"Invalid property type '{base_type}': valid types are {list(allowed_input_types.keys())}."
            )
        type_defaults = allowed_input_types[base_type]

        # Set types needed for schema
        self.base_type = base_type
        self.json_data_type = type_defaults.get("json_type")

        # Set extra schema details
        self.default_value = default_value if default_value is not None else type_defaults.get("default_value")
        self.placeholder = placeholder if placeholder is not None else type_defaults.get("placeholder")
        self.enum = enum
        self.type_title = type_defaults.get("type_title")


class PropertyAttribute:
    """
    An attribute of an ElyraProperty instance that provides the means to construct the
    schema for a property and contains information for processing property instances.
    """

    def __init__(
        self,
        attribute_id: str,
        description: Optional[str] = None,
        display_name: Optional[str] = None,
        allowed_input_types: Optional[List[PropertyInputType]] = None,
        hidden: Optional[bool] = False,
        required: Optional[bool] = False,
        pattern: Optional[str] = None,
    ):
        """
        :param attribute_id: a shorthand id for this attribute, e.g. "env_var"
        :param description: a description of this attribute
        :param display_name: the display name for this attribute
        :param allowed_input_types: a list of accepted PropertyInputTypes for this property
        :param hidden: whether this attribute should be hidden in the UI, preventing users from entering a value
        :param required: whether a value for this attribute is required
        :param pattern: a regex validation pattern to use for values given for this attribute
        """
        if not attribute_id:
            raise ValueError("Invalid Elyra property attribute: missing attribute_id.")

        self.id = attribute_id
        self.description = description
        self.title = display_name
        self.allowed_input_types = allowed_input_types or []
        self.default_type = self.allowed_input_types[0] if self.allowed_input_types else None
        self.hidden = hidden
        self.required = required
        self.pattern = pattern


class ListItemPropertyAttribute(PropertyAttribute):
    """
    An attribute of an ElyraPropertyListItem instance that provides the means to construct the
    schema for a property and contains information for processing property instances.
    """

    def __init__(
        self,
        attribute_id: str,
        description: Optional[str] = None,
        display_name: Optional[str] = None,
        allowed_input_types: Optional[List[PropertyInputType]] = None,
        hidden: Optional[bool] = False,
        required: Optional[bool] = False,
        pattern: Optional[str] = None,
        use_in_key: Optional[bool] = True,
    ):
        """
        :param use_in_key: whether this attribute should be used when constructing a
            key for an instance that will be used to de-duplicate list items
        """
        super().__init__(attribute_id, description, display_name, allowed_input_types, hidden, required, pattern)
        self.use_in_key = use_in_key


class ElyraProperty(ABC):
    """A component property that is defined and processed by Elyra"""

    applies_to_generic: bool  # True if the property applies to generic components
    applies_to_custom: bool  # True if the property applies to custom components

    property_id: str
    property_display_name: str
    property_description: str
    property_attributes: List[PropertyAttribute] = []
    _json_data_type: str = None
    _required: bool = False

    _subclass_property_map: Dict[str, type] = {}

    @classmethod
    def all_subclasses(cls):
        """Get all nested subclasses for a class."""
        return set(cls.__subclasses__()).union([s for c in cls.__subclasses__() for s in c.all_subclasses()])

    @classmethod
    def build_property_map(cls) -> None:
        """Build the map of property subclasses."""
        cls._subclass_property_map = {sc.property_id: sc for sc in cls.all_subclasses() if hasattr(sc, "property_id")}

    @classmethod
    def get_class_for_property(cls, prop_id) -> type | None:
        """Returns the ElyraProperty subclass corresponding to the given property id."""
        if not cls._subclass_property_map:
            cls.build_property_map()
        return cls._subclass_property_map.get(prop_id)

    @classmethod
    def subclass_exists_for_property(cls, prop_id: str) -> bool:
        """
        Returns a boolean indicating whether a corresponding ElyraProperty subclass
        exists for the given property id.
        """
        return cls.get_class_for_property(prop_id) is not None

    @classmethod
    def get_single_instance(cls, value: Optional[Dict[str, Any]] = None) -> ElyraProperty | None:
        """Unpack values from dictionary object and instantiate a class instance."""
        if isinstance(value, ElyraProperty):
            return value  # value is already a single instance, no further action required

        if not isinstance(value, dict):
            value = {}
        params = {attr.id: cls.strip_if_string(value.get(attr.id)) for attr in cls.property_attributes}
        instance = getattr(import_module(cls.__module__), cls.__name__)(**params)
        return None if instance.should_discard() else instance

    @classmethod
    def create_instance(cls, prop_id: str, value: Optional[Any]) -> ElyraProperty | ElyraPropertyList | None:
        """Create an instance of a class with the given property id using the user-entered values."""
        sc = cls.get_class_for_property(prop_id)
        if sc is None:
            return None

        if issubclass(sc, ElyraPropertyListItem):
            if not isinstance(value, list):
                return None
            instances = [sc.get_single_instance(obj) for obj in value]  # create instance for each object
            return ElyraPropertyList(instances).deduplicate()  # convert to ElyraPropertyList and de-dupe
        elif issubclass(sc, ElyraProperty):
            return sc.get_single_instance(value)
        return None

    @classmethod
    def get_classes_for_component_type(
        cls, component_type: str, runtime_type: Optional[RuntimeProcessorType] = None
    ) -> Set[type]:
        """
        Retrieve property subclasses that apply to the given component type
        (e.g., custom or generic) and to the given runtime type.
        """
        from elyra.pipeline.processor import PipelineProcessorManager  # placed here to avoid circular reference

        processor_props = set()
        for processor in PipelineProcessorManager.instance().get_all_processors():
            props = getattr(processor, "supported_properties", set())
            if not runtime_type or runtime_type == RuntimeProcessorType.LOCAL or processor.type == runtime_type:
                processor_props.update(props)

        all_subclasses = set()
        for sc in cls.all_subclasses():
            sc_id = getattr(sc, "property_id", "")
            if sc_id in processor_props and getattr(sc, f"applies_to_{component_type}", False):
                all_subclasses.add(sc)

        return all_subclasses

    @classmethod
    def get_schema(cls) -> Dict[str, Any]:
        """
        Build the JSON schema for an Elyra-owned component property using the attributes
        defined in the human-readable list of PropertyAttribute object for each class.
        """
        class_description = re.sub(" +", " ", cls.property_description.replace("\n", " ")).strip()
        schema = {"title": cls.property_display_name, "description": class_description, "type": cls._json_data_type}
        if cls._json_data_type is not None:  # property is a scalar value  TODO deprecate when able
            return schema

        properties, uihints, required_list = {}, {}, []
        for attr in cls.property_attributes:
            if attr.hidden:
                continue
            if attr.required:
                required_list.append(attr.id)

            properties[attr.id] = {"title": attr.title or attr.id}
            if attr.description:
                properties[attr.id]["description"] = attr.description
            if attr.pattern:
                properties[attr.id]["pattern"] = attr.pattern

            if len(attr.allowed_input_types) == 1:
                allowed_type: PropertyInputType = attr.allowed_input_types[0]

                properties[attr.id]["type"] = allowed_type.json_data_type
                if allowed_type.default_value is not None:
                    properties[attr.id]["default"] = allowed_type.default_value
                if allowed_type.enum:
                    properties[attr.id]["enum"] = allowed_type.enum
                if allowed_type.placeholder is not None:
                    uihints[attr.id] = {"ui:placeholder": allowed_type.placeholder}

            elif len(attr.allowed_input_types) > 1:
                properties[attr.id]["title"] = "Type"
                properties[attr.id]["description"] = "The type of this parameter"

                # Set default type to be the first one defined in the property_attributes list
                properties[attr.id]["default"] = {
                    "type": attr.allowed_input_types[0].base_type,
                    "value": attr.allowed_input_types[0].default_value,
                }

                # Render a oneOf block and loop through allowed types to fill it
                properties[attr.id]["oneOf"] = []
                for allowed_type in sorted(attr.allowed_input_types, key=lambda t: t.base_type):
                    schema_obj = {
                        "type": "object",
                        "title": allowed_type.type_title or allowed_type.base_type,
                        "properties": {
                            "type": {"type": "string", "default": allowed_type.type_title or allowed_type.base_type},
                            "value": {
                                "title": attr.title,
                                "description": attr.description,
                                "type": allowed_type.json_data_type,
                            },
                        },
                        "uihints": {"type": {"ui:widget": "hidden"}},
                    }
                    if allowed_type.default_value is not None:
                        schema_obj["properties"]["value"]["default"] = allowed_type.default_value
                    if allowed_type.enum:
                        schema_obj["properties"]["value"]["enum"] = allowed_type.enum
                    if allowed_type.placeholder is not None:
                        schema_obj["uihints"]["value"] = {"ui:placeholder": allowed_type.placeholder}

                    properties[attr.id]["oneOf"].append(schema_obj)

        if issubclass(cls, ElyraPropertyListItem):
            items = {"type": "object", "properties": properties, "required": required_list}
            schema.update({"type": "array", "default": [], "items": items, "uihints": {"items": uihints}})
        else:
            schema.update({"type": "object", "properties": properties, "required": required_list, "uihints": uihints})

        return schema

    @staticmethod
    def strip_if_string(var: Any) -> Any:
        """Strip surrounding whitespace from variable if it is a string"""
        return var.strip() if isinstance(var, str) else var

    def should_discard(self) -> bool:
        """
        Returns a boolean indicating whether an instance should be silently discarded on
        the basis of its attribute values. A discarded instance will not be validated or
        processed.

        Override this method if there are any constraints that dictate that this instance
        should not be processed.
        """
        return False

    @abstractmethod
    def get_value_for_display(self) -> Dict[str, Any]:
        """
        Get a representation of the instance to display in UI error messages.
        Should be implemented in any subclass that has validation criteria.
        """
        pass

    @abstractmethod
    def get_all_validation_errors(self) -> List[str]:
        """Perform custom validation on an instance."""
        pass

    @abstractmethod
    def add_to_execution_object(self, runtime_processor: RuntimePipelineProcessor, execution_object: Any, **kwargs):
        """
        Add a property instance to the execution object for the given runtime processor.
        Calls the runtime processor's implementation of add_{property_type}, e.g.
        runtime_processor.add_kubernetes_secret(self, execution_object, **kwargs).
        """
        pass


class DisableNodeCaching(ElyraProperty):
    """An ElyraProperty representing node cache preference"""

    applies_to_generic = False
    applies_to_custom = True

    property_id = DISABLE_NODE_CACHING
    property_display_name = "Disable node caching"
    property_description = "Disable caching to force node re-execution in the target runtime environment."
    _json_data_type = "string"

    def __init__(self, selection: Union[str, bool], **kwargs):
        self.selection = None
        if selection in ["True", "true", True]:
            self.selection = True
        elif selection in ["False", "false", False]:
            self.selection = False

    @classmethod
    def get_single_instance(cls, value: Optional[Any] = None) -> ElyraProperty | None:
        if isinstance(value, ElyraProperty):
            return value  # value is already a single instance, no further action required
        instance = DisableNodeCaching(selection=value)
        return None if instance.should_discard() else instance

    @classmethod
    def get_schema(cls) -> Dict[str, Any]:
        """Build the JSON schema for this property"""
        schema = super().get_schema()
        schema["enum"] = ["True", "False"]
        schema["uihints"] = {"ui:placeholder": "Use runtime default"}
        return schema

    def get_value_for_display(self) -> Dict[str, Any]:
        return self.selection

    def get_all_validation_errors(self) -> List[str]:
        return []

    def add_to_execution_object(self, runtime_processor: RuntimePipelineProcessor, execution_object: Any, **kwargs):
        """Add DisableNodeCaching info to the execution object for the given runtime processor"""
        runtime_processor.add_disable_node_caching(instance=self, execution_object=execution_object, **kwargs)

    def should_discard(self) -> bool:
        """Ignore this DisableNodeCaching instance if neither True nor False was selected."""
        return self.selection is None


class CustomSharedMemorySize(ElyraProperty):
    """An ElyraProperty representing shared memory size for a node."""

    applies_to_generic = True  # custom shared mem size applies to generic components
    applies_to_custom = True  # custom shared mem size applies to custom components

    property_id = KUBERNETES_SHARED_MEM_SIZE
    property_display_name = "Shared Memory Size"
    property_description = """Configure a custom shared memory size in
    gigabytes (10^9 bytes) for the pod that executes a node. A custom
    value is assigned if the size property value is a number greater than zero."""
    property_attributes = [
        PropertyAttribute(
            attribute_id="size",
            display_name="Memory Size (GB)",
            allowed_input_types=[PropertyInputType(base_type="int", placeholder=0)],
            hidden=False,
            required=False,
        ),
        PropertyAttribute(
            attribute_id="units",
            display_name="Units",
            allowed_input_types=[PropertyInputType(base_type="str")],
            hidden=True,
            required=False,
        ),
    ]

    default_units = "G"

    def __init__(self, size: str, units: str, **kwargs):
        self.size = size
        self.units = units or CustomSharedMemorySize.default_units

    @classmethod
    def get_schema(cls) -> Dict[str, Any]:
        """Build the JSON schema for an Elyra-owned component property"""
        schema = super().get_schema()
        schema["properties"]["size"]["minimum"] = 0
        return schema

    def to_dict(self) -> Dict[str, Any]:
        """Convert instance to a dict with relevant class attributes."""
        dict_repr = {attr.id: getattr(self, attr.id, None) for attr in self.property_attributes}
        return dict_repr

    def get_value_for_display(self) -> Dict[str, Any]:
        """Get a representation of the instance to display in UI error messages."""
        return self.to_dict()

    def get_all_validation_errors(self) -> List[str]:
        """Validate this instance. If the size attribute is set and not zero it
        must be a positive floating point number. The units attribute must be 'G'."""
        validation_errors = []
        # verify custom size
        try:
            if self.size:
                size = float(self.size)
                if size < 0:
                    raise ValueError()
        except ValueError:
            validation_errors.append(f"Shared memory size '{self.size}' must be a positive number.")
        # verify units
        if self.units not in ["G"]:
            validation_errors.append(f"Shared memory size units '{self.units}' must be 'G'.")
        return validation_errors

    def add_to_execution_object(self, runtime_processor: RuntimePipelineProcessor, execution_object: Any, **kwargs):
        """Add CustomSharedMemorySize instance to the execution object for the given runtime processor"""
        runtime_processor.add_custom_shared_memory_size(instance=self, execution_object=execution_object, **kwargs)

    def should_discard(self) -> bool:
        """Ignore this CustomSharedMemorySize instance if no custom size is specified."""
        return not self.size


class ElyraPropertyListItem(ElyraProperty):
    """
    An Elyra-owned property that is meant to be a member of an ElyraOwnedPropertyList.
    """

    property_attributes: List[ListItemPropertyAttribute] = []

    def to_dict(self) -> Dict[str, Any]:
        """Convert instance to a dict with relevant class attributes."""
        dict_repr = {attr.id: getattr(self, attr.id, None) for attr in self.property_attributes}
        return dict_repr

    def get_key_for_dict_entry(self) -> str:
        """
        Given the attribute names in the 'key' property, construct a key
        based on the attribute values of the instance.
        """
        prop_key = ""
        keys = [attr.id for attr in self.property_attributes if attr.use_in_key]
        for key_attr in keys:
            key_part = getattr(self, key_attr)
            if key_part:
                prop_key += f"{key_part}:" if key_attr != keys[-1] else key_part
        return prop_key

    def get_value_for_dict_entry(self) -> Union[str, Dict[str, Any]]:
        """Returns the value to be used when constructing a dict from a list of ElyraPropertyListItem."""
        return self.to_dict()

    def get_value_for_display(self) -> Dict[str, Any]:
        """Get a representation of the instance to display in UI error messages."""
        dict_repr = self.to_dict()
        for attr in self.property_attributes:
            if attr.hidden:
                dict_repr.pop(attr.id)
        return dict_repr


class EnvironmentVariable(ElyraPropertyListItem):
    """An ElyraProperty representing a single Environment Variable"""

    applies_to_generic = True
    applies_to_custom = False

    property_id = ENV_VARIABLES
    property_display_name = "Environment Variables"
    property_description = "Environment variables to be set on the execution environment."
    property_attributes = [
        ListItemPropertyAttribute(
            attribute_id="env_var",
            display_name="Environment Variable",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="ENV_VAR")],
            hidden=False,
            required=True,
            use_in_key=True,
        ),
        ListItemPropertyAttribute(
            attribute_id="value",
            display_name="Value",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="value")],
            hidden=False,
            required=False,
            use_in_key=False,
        ),
    ]

    def __init__(self, env_var, value, **kwargs):
        self.env_var = env_var
        self.value = value

    @classmethod
    def get_schema(cls) -> Dict[str, Any]:
        """Build the JSON schema for an Elyra-owned component property"""
        schema = super().get_schema()
        schema["uihints"].update({"canRefresh": True})
        return schema

    def get_value_for_dict_entry(self) -> Union[str, Dict[str, Any]]:
        """
        Returns the value to be used when constructing a dict from a list of ElyraPropertyListItem.
        A EnvironmentVariable dict entry will be of the form {self.env_var: self.value}
        """
        return self.value

    def should_discard(self) -> bool:
        """If a value is not specified, this EnvironmentVariable instance should be silently ignored."""
        return not self.value

    def get_all_validation_errors(self) -> List[str]:
        """Perform custom validation on an instance."""
        validation_errors = []
        if not self.env_var:
            validation_errors.append("Required environment variable was not specified.")
        elif " " in self.env_var:
            validation_errors.append(f"Environment variable '{self.env_var}' includes invalid space character(s).")

        return validation_errors

    def add_to_execution_object(self, runtime_processor: RuntimePipelineProcessor, execution_object: Any, **kwargs):
        """Add EnvironmentVariable instance to the execution object for the given runtime processor"""
        runtime_processor.add_env_var(instance=self, execution_object=execution_object, **kwargs)


class KubernetesSecret(ElyraPropertyListItem):
    """An ElyraProperty representing a single Kubernetes secret"""

    applies_to_generic = True
    applies_to_custom = False

    property_id = KUBERNETES_SECRETS
    property_display_name = "Kubernetes Secrets"
    property_description = """Kubernetes secrets to make available as environment
    variables to this node. The secret name and key given must be present in the
    Kubernetes namespace where the node is executed or this node will not run."""
    property_attributes = [
        ListItemPropertyAttribute(
            attribute_id="env_var",
            display_name="Environment Variable",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="ENV_VAR")],
            hidden=False,
            required=True,
            use_in_key=True,
        ),
        ListItemPropertyAttribute(
            attribute_id="name",
            display_name="Secret Name",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="secret-name")],
            hidden=False,
            required=True,
            use_in_key=False,
        ),
        ListItemPropertyAttribute(
            attribute_id="key",
            display_name="Secret Key",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="secret-key")],
            hidden=False,
            required=True,
            use_in_key=False,
        ),
    ]

    def __init__(self, env_var, name, key, **kwargs):
        self.env_var = env_var
        self.name = name
        self.key = key

    def get_all_validation_errors(self) -> List[str]:
        """Perform custom validation on an instance."""
        validation_errors = []
        if not self.env_var:
            validation_errors.append("Required environment variable was not specified.")
        if not self.name:
            validation_errors.append("Required secret name was not specified.")
        elif not is_valid_kubernetes_resource_name(self.name):
            validation_errors.append(
                f"Secret name '{self.name}' is not a valid Kubernetes resource name.",
            )
        if not self.key:
            validation_errors.append("Required secret key was not specified.")
        elif not is_valid_kubernetes_key(self.key):
            validation_errors.append(
                f"Key '{self.key}' is not a valid Kubernetes secret key.",
            )

        return validation_errors

    def add_to_execution_object(self, runtime_processor: RuntimePipelineProcessor, execution_object: Any, **kwargs):
        """Add KubernetesSecret instance to the execution object for the given runtime processor"""
        runtime_processor.add_kubernetes_secret(instance=self, execution_object=execution_object, **kwargs)


class VolumeMount(ElyraPropertyListItem):
    """An ElyraProperty representing a single PVC"""

    applies_to_generic = True
    applies_to_custom = True

    property_id = MOUNTED_VOLUMES
    property_display_name = "Data Volumes"
    property_description = """Volumes to be mounted in this node. The specified Persistent Volume Claims
    must exist in the Kubernetes namespace where the node is executed or this node will not run."""
    property_attributes = [
        ListItemPropertyAttribute(
            attribute_id="path",
            display_name="Mount Path",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="/mount/path")],
            hidden=False,
            required=True,
            use_in_key=True,
        ),
        ListItemPropertyAttribute(
            attribute_id="pvc_name",
            display_name="Persistent Volume Claim Name",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="pvc-name")],
            hidden=False,
            required=True,
            use_in_key=False,
        ),
        ListItemPropertyAttribute(
            attribute_id="sub_path",
            display_name="Sub Path",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="relative/path/within/volume")],
            hidden=False,
            required=False,
            use_in_key=False,
        ),
        ListItemPropertyAttribute(
            attribute_id="read_only",
            display_name="Mount volume read-only",
            allowed_input_types=[PropertyInputType(base_type="bool", default_value=False, placeholder=None)],
            hidden=False,
            required=False,
            use_in_key=False,
        ),
    ]

    def __init__(self, path: str, pvc_name: str, sub_path: str, read_only: bool, **kwargs):
        self.path = path
        self.pvc_name = pvc_name
        self.sub_path = sub_path
        self.read_only = read_only

    def get_all_validation_errors(self) -> List[str]:
        """Identify configuration issues for this instance"""
        validation_errors = []
        if not self.path:
            validation_errors.append("Required mount path was not specified.")
        if not self.pvc_name:
            validation_errors.append("Required persistent volume claim name was not specified.")
        elif not is_valid_kubernetes_resource_name(self.pvc_name):
            validation_errors.append(f"PVC name '{self.pvc_name}' is not a valid Kubernetes resource name.")
        if self.sub_path and self.sub_path.startswith("/"):
            validation_errors.append(f"Sub-path '{self.sub_path}' must be a relative path.")

        return validation_errors

    def add_to_execution_object(self, runtime_processor: RuntimePipelineProcessor, execution_object: Any, **kwargs):
        """Add VolumeMount instance to the execution object for the given runtime processor"""
        self.path = f"/{self.path.strip('/')}"  # normalize path
        if self.read_only is None:
            self.read_only = False
        runtime_processor.add_mounted_volume(instance=self, execution_object=execution_object, **kwargs)

    def should_discard(self) -> bool:
        """
        Returns a boolean indicating whether this VolumeMount instance should be silently discarded on
        the basis of its mount path, PVC name, and sub-path attribute values. If these attributes
        don't contain values this instance will not be validated or processed.
        """
        # ignore the read_only attribute because it always contains a value
        return not (self.path or self.pvc_name or self.sub_path)


class KubernetesAnnotation(ElyraPropertyListItem):
    """An ElyraProperty representing a single Kubernetes annotation"""

    applies_to_generic = True
    applies_to_custom = True

    property_id = KUBERNETES_POD_ANNOTATIONS
    property_display_name = "Kubernetes Pod Annotations"
    property_description = """Metadata to be added to this node. The metadata is exposed
    as annotation in the Kubernetes pod that executes this node."""
    property_attributes = [
        ListItemPropertyAttribute(
            attribute_id="key",
            display_name="Key",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="annotation_key")],
            hidden=False,
            required=True,
            use_in_key=True,
        ),
        ListItemPropertyAttribute(
            attribute_id="value",
            display_name="Value",
            allowed_input_types=[PropertyInputType(base_type="str", placeholder="annotation_value")],
            hidden=False,
            required=False,
            use_in_key=False,
        ),
    ]

