Add support for Annotated types in plugins

changelogs/unreleased/8573-add-support-for-annotated-types.yml

@@ -0,0 +1,4 @@
+description: Add support for annotated types to plugins.
+issue-nr: 8573
+change-type: minor
+destination-branches: [master, iso8]

src/inmanta/plugin_typing.py

@@ -0,0 +1,42 @@
+"""
+    Copyright 2025 Inmanta
+
+    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.
+
+    Contact: [email protected]
+"""
+
+import typing
+from dataclasses import dataclass
+
+# TODO: move this module to inmanta.plugins.typing? Probably not because that would import the whole `plugins` namespace?
+
+
+@dataclass(frozen=True)
+class InmantaType:
+    """
+    Declaration of an inmanta type for use with typing.Annotated.
+    When a plugin type is declared as typing.Annotated with an `InmantaType` as annotation, the Python type is completely
+    ignored for type validation and conversion to and from the DSL. Instead the string provided to the `InmantaType` is
+    evaluated as a DSL type, extended with "any".
+    For maximum static type coverage, it is recommended to use these only when absolutely necessary, and to use them as deeply
+    in the type as possible, e.g. prefer `Sequence[Annotated[MyEntity, InmantaType("std::Entity")]]` over
+    `Annotated[Sequence[Entity], InmantaType("std::Entity[]")]`.
+    """
+
+    dsl_type: str
+
+
+# TODO: how to do Entity? "object" is appropriate but raises too many errors for practical use. Any is Any
+Entity: typing.TypeAlias = typing.Annotated[object, InmantaType("std::Entity")]
+string: typing.TypeAlias = typing.Annotated[str, InmantaType("string")]

src/inmanta/plugins.py

@@ -31,8 +31,8 @@
 import typing_inspect
 
 import inmanta.ast.type as inmanta_type
-from inmanta import const, protocol, util
-from inmanta.ast import (  # noqa: F401 Plugin exception is part of the stable api
+from inmanta import const, plugin_typing, protocol, util
+from inmanta.ast import (  # noqa: F401 Plugin e
     LocatableString,
     Location,
     Namespace,
@@ -245,11 +245,18 @@ def __eq__(self, other: object) -> bool:
 }
 
 
-def to_dsl_type(python_type: type[object]) -> inmanta_type.Type:
+def parse_dsl_type(dsl_type: str, location: Range, resolver: Namespace) -> inmanta_type.Type:
+    locatable_type: LocatableString = LocatableString(dsl_type, location, 0, resolver)
+    return inmanta_type.resolve_type(locatable_type, resolver)
+
+
+def to_dsl_type(python_type: type[object], location: Range, resolver: Namespace) -> inmanta_type.Type:
     """
     Convert a python type annotation to an Inmanta DSL type annotation.
 
     :param python_type: The evaluated python type as provided in the Python type annotation.
+    :param location: The location of this evaluation on the model
+    :param resolver: The namespace that can be used to resolve the type annotation of this argument.
     """
     # Any to any
     if python_type is typing.Any:
@@ -268,7 +275,7 @@ def to_dsl_type(python_type: type[object]) -> inmanta_type.Type:
                 # Probably not possible
                 return Null()
             if len(other_types) == 1:
-                return inmanta_type.NullableType(to_dsl_type(other_types[0]))
+                return inmanta_type.NullableType(to_dsl_type(other_types[0], location, resolver))
             # TODO: optional unions
             return inmanta_type.Type()
         else:
@@ -296,7 +303,7 @@ def to_dsl_type(python_type: type[object]) -> inmanta_type.Type:
                 if len(args) == 1:
                     return inmanta_type.TypedDict(inmanta_type.Type())
 
-                return inmanta_type.TypedDict(to_dsl_type(args[1]))
+                return inmanta_type.TypedDict(to_dsl_type(args[1], location, resolver))
             else:
                 raise TypingException(None, f"invalid type {python_type}, dictionary types should be Mapping or dict")
 
@@ -306,27 +313,26 @@ def to_dsl_type(python_type: type[object]) -> inmanta_type.Type:
                 args = typing.get_args(python_type)
                 if not args:
                     return inmanta_type.List()
-                return inmanta_type.TypedList(to_dsl_type(args[0]))
+                return inmanta_type.TypedList(to_dsl_type(args[0], location, resolver))
             else:
                 raise TypingException(None, f"invalid type {python_type}, list types should be Sequence or list")
 
         # Set
         if issubclass(origin, collections.abc.Set):
             raise TypingException(None, f"invalid type {python_type}, set is not supported on the plugin boundary")
 
-    # TODO annotated types
-    # if typing.get_origin(t) is typing.Annotated:
-    #     args: Sequence[object] = typing.get_args(python_type)
-    #     inmanta_types: Sequence[plugin_typing.InmantaType] =
-    #     [arg if isinstance(arg, plugin_typing.InmantaType) for arg in args]
-    #     if inmanta_types:
-    #         if len(inmanta_types) > 1:
-    #             # TODO
-    #             raise Exception()
-    #         # TODO
-    #         return parse_dsl_type(inmanta_types[0].dsl_type)
-    #     # the annotation doesn't concern us => use base type
-    #     return to_dsl_type(args[0])
+    if typing.get_origin(python_type) is typing.Annotated:
+        annotated_args = typing.get_args(python_type)
+        inmanta_types: Sequence[plugin_typing.InmantaType] = [
+            arg for arg in annotated_args if isinstance(arg, plugin_typing.InmantaType)
+        ]
+        if inmanta_types:
+            if len(inmanta_types) > 1:
+                raise TypingException(None, f"invalid type {python_type}, only one InmantaType annotation is supported")
+            return parse_dsl_type(inmanta_types[0].dsl_type, location, resolver)
+        # the annotation doesn't concern us => use base type
+        return to_dsl_type(annotated_args[0], location, resolver)
+
     if python_type in python_to_model:
         return python_to_model[python_type]
 
@@ -386,19 +392,19 @@ def resolve_type(self, plugin: "Plugin", resolver: Namespace) -> inmanta_type.Ty
             self._resolved_type = PLUGIN_TYPES[self.type_expression]
             return self._resolved_type
 
+        plugin_line: Range = Range(plugin.location.file, plugin.location.lnr, 1, plugin.location.lnr + 1, 1)
         if not isinstance(self.type_expression, str):
             if isinstance(self.type_expression, type) or typing.get_origin(self.type_expression) is not None:
-                self._resolved_type = to_dsl_type(self.type_expression)
+                self._resolved_type = to_dsl_type(self.type_expression, plugin_line, resolver)
             else:
                 raise RuntimeException(
                     stmt=None,
                     msg="Bad annotation in plugin %s for %s, expected str or python type but got %s (%s)"
                     % (plugin.get_full_name(), self.VALUE_NAME, type(self.type_expression).__name__, self.type_expression),
                 )
         else:
-            plugin_line: Range = Range(plugin.location.file, plugin.location.lnr, 1, plugin.location.lnr + 1, 1)
-            locatable_type: LocatableString = LocatableString(self.type_expression, plugin_line, 0, resolver)
-            self._resolved_type = inmanta_type.resolve_type(locatable_type, resolver)
+            self._resolved_type = parse_dsl_type(self.type_expression, plugin_line, resolver)
+
         return self._resolved_type
 
     def validate(self, value: object) -> bool:

tests/compiler/test_plugin_types.py

@@ -17,44 +17,71 @@
 """
 
 import collections.abc
-from typing import Any, Mapping, Sequence, Union
+from typing import Annotated, Any, Mapping, Sequence, Union
 
 import pytest
 
 import inmanta.ast.type as inmanta_type
-from inmanta.ast import RuntimeException
+from inmanta import plugin_typing
+from inmanta.ast import Namespace, Range, RuntimeException
 from inmanta.plugins import Null, to_dsl_type
 
+# from inmanta.ast.entity import Entity
+# from inmanta.ast.type import TYPES
+
 
 def test_conversion(caplog):
     """
     Test behaviour of to_dsl_type function.
     """
-    assert inmanta_type.Integer() == to_dsl_type(int)
-    assert inmanta_type.Float() == to_dsl_type(float)
-    assert inmanta_type.NullableType(inmanta_type.Float()) == to_dsl_type(float | None)
-    assert inmanta_type.List() == to_dsl_type(list)
-    assert inmanta_type.TypedList(inmanta_type.String()) == to_dsl_type(list[str])
-    assert inmanta_type.TypedList(inmanta_type.String()) == to_dsl_type(Sequence[str])
-    assert inmanta_type.List() == to_dsl_type(Sequence)
-    assert inmanta_type.List() == to_dsl_type(collections.abc.Sequence)
-    assert inmanta_type.TypedList(inmanta_type.String()) == to_dsl_type(collections.abc.Sequence[str])
-    assert inmanta_type.TypedDict(inmanta_type.Type()) == to_dsl_type(dict)
-    assert inmanta_type.TypedDict(inmanta_type.Type()) == to_dsl_type(Mapping)
-    assert inmanta_type.TypedDict(inmanta_type.String()) == to_dsl_type(dict[str, str])
-    assert inmanta_type.TypedDict(inmanta_type.String()) == to_dsl_type(Mapping[str, str])
-
-    assert inmanta_type.TypedDict(inmanta_type.String()) == to_dsl_type(collections.abc.Mapping[str, str])
-
-    assert Null() == to_dsl_type(Union[None])
-
-    assert isinstance(to_dsl_type(Any), inmanta_type.Type)
+    namespace = Namespace("dummy-namespace")
+    # namespace.set_primitives(inmanta_type.TYPES)
+    # FIXME: Not working    because of std = self.get_ns_from_string("std") std is None
+    namespace.primitives = inmanta_type.TYPES
+
+    location: Range = Range("test", 1, 1, 2, 1)
+
+    assert inmanta_type.String() == to_dsl_type(plugin_typing.string, location, namespace)
+
+    assert inmanta_type.NullableType(inmanta_type.Integer()) == to_dsl_type(
+        Annotated[int | None, "something"], location, namespace
+    )
+
+    assert inmanta_type.TypedDict(inmanta_type.Type()) == to_dsl_type(
+        Annotated[dict[str, int], plugin_typing.InmantaType("dict")], location, namespace
+    )
+
+    # FIXME: Not working
+    # entity: Entity = Entity("my-entity", namespace)
+    # namespace.define_type("my-entity", entity)
+    # assert inmanta_type.TypedDict(inmanta_type.Type()) == to_dsl_type(
+    #     Annotated[Union[int, str] | None, plugin_typing.InmantaType("my-entity")], location, namespace
+    # )
+    assert inmanta_type.Integer() == to_dsl_type(int, location, namespace)
+    assert inmanta_type.Float() == to_dsl_type(float, location, namespace)
+    assert inmanta_type.NullableType(inmanta_type.Float()) == to_dsl_type(float | None, location, namespace)
+    assert inmanta_type.List() == to_dsl_type(list, location, namespace)
+    assert inmanta_type.TypedList(inmanta_type.String()) == to_dsl_type(list[str], location, namespace)
+    assert inmanta_type.TypedList(inmanta_type.String()) == to_dsl_type(Sequence[str], location, namespace)
+    assert inmanta_type.List() == to_dsl_type(Sequence, location, namespace)
+    assert inmanta_type.List() == to_dsl_type(collections.abc.Sequence, location, namespace)
+    assert inmanta_type.TypedList(inmanta_type.String()) == to_dsl_type(collections.abc.Sequence[str], location, namespace)
+    assert inmanta_type.TypedDict(inmanta_type.Type()) == to_dsl_type(dict, location, namespace)
+    assert inmanta_type.TypedDict(inmanta_type.Type()) == to_dsl_type(Mapping, location, namespace)
+    assert inmanta_type.TypedDict(inmanta_type.String()) == to_dsl_type(dict[str, str], location, namespace)
+    assert inmanta_type.TypedDict(inmanta_type.String()) == to_dsl_type(Mapping[str, str], location, namespace)
+
+    assert inmanta_type.TypedDict(inmanta_type.String()) == to_dsl_type(collections.abc.Mapping[str, str], location, namespace)
+
+    assert Null() == to_dsl_type(Union[None], location, namespace)
+
+    assert isinstance(to_dsl_type(Any, location, namespace), inmanta_type.Type)
 
     with pytest.raises(RuntimeException):
-        to_dsl_type(dict[int, int])
+        to_dsl_type(dict[int, int], location, namespace)
 
     with pytest.raises(RuntimeException):
-        to_dsl_type(set[str])
+        to_dsl_type(set[str], location, namespace)
 
     class CustomList[T](list[T]):
         pass
@@ -63,14 +90,14 @@ class CustomDict[K, V](Mapping[K, V]):
         pass
 
     with pytest.raises(RuntimeException):
-        to_dsl_type(CustomList[str])
+        to_dsl_type(CustomList[str], location, namespace)
 
     with pytest.raises(RuntimeException):
-        to_dsl_type(CustomDict[str, str])
+        to_dsl_type(CustomDict[str, str], location, namespace)
 
     # Check that a warning is produced when implicit cast to 'Any'
     caplog.clear()
-    to_dsl_type(complex)
+    to_dsl_type(complex, location, namespace)
     warning_message = (
         "InmantaWarning: Python type <class 'complex'> was implicitly cast to 'Any' because no matching type "
         "was found in the Inmanta DSL. Please refer to the documentation for an overview of supported types at the "

tests/compiler/test_plugins.py

@@ -426,6 +426,7 @@ def test_native_types(snippetcompiler: "SnippetCompilationTest") -> None:
         """
 import plugin_native_types
 
+test_entity = plugin_native_types::TestEntity()
 a = "b"
 a = plugin_native_types::get_from_dict({"a":"b"}, "a")
 
@@ -435,6 +436,10 @@ def test_native_types(snippetcompiler: "SnippetCompilationTest") -> None:
 a = plugin_native_types::many_arguments(["a","c","b"], 1)
 
 none = plugin_native_types::as_none("a")
-        """
+        # Annotated types
+plugin_native_types::annotated_arg_entity(test_entity)     # type value: Annotated[object, InmantaType("TestEntity")]
+plugin_native_types::annotated_return_entity(test_entity)  # type return value: Annotated[object, InmantaType("TestEntity")]
+        """,
+        autostd=True,
     )
     compiler.do_compile()

tests/data/modules/plugin_native_types/model/_init.cf

@@ -0,0 +1,5 @@
+import std
+entity TestEntity:
+end
+
+implement TestEntity using std::none

tests/data/modules/plugin_native_types/plugins/__init__.py

@@ -16,7 +16,9 @@
     Contact: [email protected]
 """
 
+from typing import Any, Annotated
 from inmanta.plugins import plugin
+from inmanta import plugin_typing
 
 
 @plugin
@@ -32,3 +34,16 @@ def many_arguments(il: list[str], idx: int) -> str:
 @plugin
 def as_none(value: str) -> None:
     pass
+
+
+# Annotated values
+
+
+@plugin
+def annotated_arg_entity(value: Annotated[object, plugin_typing.InmantaType("TestEntity")]) -> None:
+    pass
+
+
+@plugin
+def annotated_return_entity(value: Any) -> Annotated[object, plugin_typing.InmantaType("TestEntity")]:
+    return value