fixed ytping

Author: Andrei Neagu

packages/models-library/src/models_library/osparc_variable_identifier.py

@@ -7,6 +7,7 @@
 from pydantic import BaseModel, Discriminator, PositiveInt, Tag
 
 from .utils.string_substitution import OSPARC_IDENTIFIER_PREFIX
+from .utils.types import get_types_from_annotated_union
 
 T = TypeVar("T")
 
@@ -89,7 +90,7 @@ def example_func(par: OsparcVariableIdentifier | int) -> None:
     Raises:
         TypeError: if the the OsparcVariableIdentifier was unresolved
     """
-    if isinstance(var, OsparcVariableIdentifier):
+    if isinstance(var, get_types_from_annotated_union(OsparcVariableIdentifier)):
         raise UnresolvedOsparcVariableIdentifierError(value=var)
     return var
 
@@ -116,7 +117,7 @@ def replace_osparc_variable_identifier(  # noqa: C901
     ```
     """
 
-    if isinstance(obj, OsparcVariableIdentifier):
+    if isinstance(obj, get_types_from_annotated_union(OsparcVariableIdentifier)):
         if obj.name in osparc_variables:
             return deepcopy(osparc_variables[obj.name])  # type: ignore
         if obj.default_value is not None:
@@ -154,7 +155,7 @@ def raise_if_unresolved_osparc_variable_identifier_found(obj: Any) -> None:
         UnresolvedOsparcVariableIdentifierError: if not all instances of
         `OsparcVariableIdentifier` were replaced
     """
-    if isinstance(obj, OsparcVariableIdentifier):
+    if isinstance(obj, get_types_from_annotated_union(OsparcVariableIdentifier)):
         raise_if_unresolved(obj)
     elif isinstance(obj, dict):
         for key, value in obj.items():

packages/models-library/src/models_library/service_settings_nat_rule.py

@@ -1,10 +1,21 @@
 from collections.abc import Generator
 from typing import Final
 
-from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationInfo, field_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    TypeAdapter,
+    ValidationInfo,
+    field_validator,
+)
 
 from .basic_types import PortInt
-from .osparc_variable_identifier import OsparcVariableIdentifier, raise_if_unresolved
+from .osparc_variable_identifier import (
+    OsparcVariableIdentifier,
+    raise_if_unresolved,
+)
+from .utils.types import get_types_from_annotated_union
 
 # Cloudflare DNS server address
 DEFAULT_DNS_SERVER_ADDRESS: Final[str] = "1.1.1.1"  # NOSONAR
@@ -20,13 +31,15 @@ class _PortRange(BaseModel):
     @field_validator("upper")
     @classmethod
     def lower_less_than_upper(cls, v, info: ValidationInfo) -> PortInt:
-        if isinstance(v, OsparcVariableIdentifier):
+        if isinstance(v, get_types_from_annotated_union(OsparcVariableIdentifier)):
             return v  # type: ignore # bypass validation if unresolved
 
         upper = v
         lower: PortInt | OsparcVariableIdentifier | None = info.data.get("lower")
 
-        if lower and isinstance(lower, OsparcVariableIdentifier):
+        if lower and isinstance(
+            lower, get_types_from_annotated_union(OsparcVariableIdentifier)
+        ):
             return v  # type: ignore # bypass validation if unresolved
 
         if lower is None or lower >= upper:

packages/models-library/src/models_library/utils/types.py

@@ -0,0 +1,32 @@
+from typing import Annotated, Any, Union, get_args, get_origin
+
+
+def get_types_from_annotated_union(annotated_alias: Any) -> tuple[type, ...]:
+    """
+    Introspects a complex Annotated alias to extract the base types from its inner Union.
+    """
+    if get_origin(annotated_alias) is not Annotated:
+        msg = "Expected an Annotated type."
+        raise TypeError(msg)
+
+    # Get the contents of Annotated, e.g., (Union[...], Discriminator(...))
+    annotated_args = get_args(annotated_alias)
+    union_type = annotated_args[0]
+
+    # The Union can be from typing.Union or the | operator
+    if get_origin(union_type) is not Union:
+        msg = "Expected a Union inside the Annotated type."
+        raise TypeError(msg)
+
+    # Get the members of the Union, e.g., (Annotated[TypeA, ...], Annotated[TypeB, ...])
+    union_members = get_args(union_type)
+
+    extracted_types = []
+    for member in union_members:
+        # Each member is also Annotated, so we extract its base type
+        if get_origin(member) is Annotated:
+            extracted_types.append(get_args(member)[0])
+        else:
+            extracted_types.append(member)  # Handle non-annotated members in the union
+
+    return tuple(extracted_types)

packages/models-library/tests/test_service_settings_nat_rule.py

@@ -9,6 +9,7 @@
     replace_osparc_variable_identifier,
 )
 from models_library.service_settings_nat_rule import NATRule
+from models_library.utils.types import get_types_from_annotated_union
 from pydantic import TypeAdapter
 
 SUPPORTED_TEMPLATES: set[str] = {
@@ -111,13 +112,13 @@ def test_______(replace_with_value: Any):
     a_var = TypeAdapter(OsparcVariableIdentifier).validate_python(
         "$OSPARC_VARIABLE_some_var"
     )
-    assert isinstance(a_var, OsparcVariableIdentifier)
+    assert isinstance(a_var, get_types_from_annotated_union(OsparcVariableIdentifier))
 
     replaced_var = replace_osparc_variable_identifier(
         a_var, {"OSPARC_VARIABLE_some_var": replace_with_value}
     )
     # NOTE: after replacement the original reference still points
-    assert isinstance(a_var, OsparcVariableIdentifier)
+    assert isinstance(a_var, get_types_from_annotated_union(OsparcVariableIdentifier))
     assert replaced_var == replace_with_value
 
 
@@ -154,15 +155,15 @@ def test_replace_an_instance_of_osparc_variable_identifier(
         formatted_template = var_template
 
     a_var = TypeAdapter(OsparcVariableIdentifier).validate_python(formatted_template)
-    assert isinstance(a_var, OsparcVariableIdentifier)
+    assert isinstance(a_var, get_types_from_annotated_union(OsparcVariableIdentifier))
 
     replace_with_identifier_default = identifier_has_default and replace_with_default
     replacement_content = (
         {} if replace_with_identifier_default else {a_var.name: replace_with_value}
     )
     replaced_var = replace_osparc_variable_identifier(a_var, replacement_content)
     # NOTE: after replacement the original reference still points
-    assert isinstance(a_var, OsparcVariableIdentifier)
+    assert isinstance(a_var, get_types_from_annotated_union(OsparcVariableIdentifier))
     if replace_with_identifier_default:
         assert replaced_var == default_value
     else:

packages/service-integration/tests/data/runtime.yml

@@ -11,10 +11,6 @@ settings:
     type: string
     value:
       - node.platform.os == linux
-  # # https://docs.docker.com/compose/compose-file/compose-file-v3/#environment
-  # - name: environment
-  #   type: string
-  #     -
 paths-mapping:
   inputs_path: "/config/workspace/inputs"
   outputs_path: "/config/workspace/outputs"