Native python types

changelogs/unreleased/native_plugin_types.yml

@@ -0,0 +1,5 @@
+description: Added support for python types in plugin annotations
+change-type: minor
+destination-branches: [master, iso7]
+sections:
+  feature: "{{description}}"

mypy-baseline.txt

@@ -548,7 +548,6 @@ src/inmanta/parser/cache.py:0: error: Argument 1 to "ASTUnpickler" has incompati
 src/inmanta/loader.py:0: error: Missing type parameters for generic type Module  [type-arg]
 src/inmanta/loader.py:0: error: Argument 1 to "getsourcefile" has incompatible type "object"; expected Module | type[Any] | MethodType | FunctionType | TracebackType | FrameType | CodeType | Callable[..., Any]  [arg-type]
 src/inmanta/loader.py:0: error: Function is missing a type annotation  [no-untyped-def]
-src/inmanta/loader.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/inmanta/loader.py:0: error: Return type "bytes" of "get_source" incompatible with return type "str | None" in supertype "InspectLoader"  [override]
 src/inmanta/loader.py:0: error: No return value expected  [return-value]
 src/inmanta/loader.py:0: error: Incompatible types in assignment (expression has type "None", variable has type "PluginModuleFinder")  [assignment]
@@ -791,7 +790,11 @@ src/inmanta/protocol/rest/client.py:0: error: Argument "result" to "Result" has
 src/inmanta/plugins.py:0: error: Missing type parameters for generic type "ResultVariable"  [type-arg]
 src/inmanta/plugins.py:0: error: Argument 1 to "add_function" of "PluginMeta" has incompatible type "PluginMeta"; expected "type[Plugin]"  [arg-type]
 src/inmanta/plugins.py:0: error: "type[Plugin]" has no attribute "__fq_plugin_name__"  [attr-defined]
+src/inmanta/plugins.py:0: error: Non-overlapping identity check (left operand type: "type[object]", right operand type: "<typing special form>")  [comparison-overlap]
+src/inmanta/plugins.py:0: error: Argument 1 to "to_dsl_type" has incompatible type "Type"; expected "type[object]"  [arg-type]
+src/inmanta/plugins.py:0: error: Argument 1 to "issubclass" has incompatible type "Any | None"; expected "type"  [arg-type]
 src/inmanta/plugins.py:0: error: Invalid index type "object" for "dict[str | None, Type]"; expected type "str | None"  [index]
+src/inmanta/plugins.py:0: error: Argument 1 to "to_dsl_type" has incompatible type "object"; expected "type[object]"  [arg-type]
 src/inmanta/plugins.py:0: error: Incompatible return value type (got "Type | None", expected "Type")  [return-value]
 src/inmanta/plugins.py:0: error: "type[Plugin]" has no attribute "__function__"  [attr-defined]
 src/inmanta/plugins.py:0: error: "type[Plugin]" has no attribute "__function__"  [attr-defined]

src/inmanta/ast/type.py

@@ -370,6 +370,9 @@ def is_primitive(self) -> bool:
     def get_location(self) -> None:
         return None
 
+    def __eq__(self, other: object) -> bool:
+        return type(self) == type(other)
+
 
 @stable_api
 class List(Type):
@@ -544,6 +547,11 @@ def type_string_internal(self) -> str:
     def get_location(self) -> None:
         return None
 
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, TypedDict):
+            return NotImplemented
+        return self.element_type == other.element_type
+
 
 @stable_api
 class LiteralDict(TypedDict):

src/inmanta/loader.py

@@ -221,7 +221,7 @@ def __lt__(self, other):
             return NotImplemented
         return (self.name, self.hash_value, self.is_byte_code) < (other.name, other.hash_value, other.is_byte_code)
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
         if not isinstance(other, ModuleSource):
             return False
         return (self.name, self.hash_value, self.is_byte_code) == (other.name, other.hash_value, other.is_byte_code)

src/inmanta/plugins.py

@@ -19,11 +19,15 @@
 import asyncio
 import collections.abc
 import inspect
+import numbers
 import os
 import subprocess
+import typing
 import warnings
 from collections import abc
-from typing import TYPE_CHECKING, Any, Callable, Literal, Mapping, Optional, Sequence, Type, TypeVar
+from typing import TYPE_CHECKING, Any, Callable, Dict, Literal, Mapping, Optional, Sequence, Type, TypeVar
+
+import typing_inspect
 
 import inmanta.ast.type as inmanta_type
 from inmanta import const, protocol, util
@@ -35,6 +39,7 @@
     Range,
     RuntimeException,
     TypeNotFoundException,
+    TypingException,
     WithComment,
 )
 from inmanta.ast.type import NamedType
@@ -210,6 +215,9 @@ def type_string(self) -> str:
     def type_string_internal(self) -> str:
         return self.type_string()
 
+    def __eq__(self, other: object) -> bool:
+        return type(self) == type(other)
+
 
 # Define some types which are used in the context of plugins.
 PLUGIN_TYPES = {
@@ -219,6 +227,83 @@ def type_string_internal(self) -> str:
     None: Null(),  # Only NoneValue will pass validation
 }
 
+python_to_model = {
+    str: inmanta_type.String(),
+    float: inmanta_type.Float(),
+    numbers.Number: inmanta_type.Number(),
+    int: inmanta_type.Integer(),
+    bool: inmanta_type.Bool(),
+    dict: inmanta_type.LiteralDict(),
+    list: inmanta_type.LiteralList(),
+}
+
+
+def to_dsl_type(python_type: type[object]) -> inmanta_type.Type:
+    """
+    :param python_type: The evaluated python type as provided in the Python type annotation.
+    """
+    # Any to any
+    if python_type is typing.Any:
+        return inmanta_type.Type()
+
+    # None to None
+    if python_type is type(None):
+        return Null()
+
+    # Unions and optionals
+    if typing_inspect.is_union_type(python_type):
+        # Optional type
+        if any(typing_inspect.is_optional_type(tt) for tt in typing.get_args(python_type)):
+            other_types = [tt for tt in typing.get_args(python_type) if not typing_inspect.is_optional_type(tt)]
+            if len(other_types) == 0:
+                # Probably not possible
+                return Null()
+            if len(other_types) == 1:
+                return inmanta_type.NullableType(to_dsl_type(other_types[0]))
+            # TODO: optional unions
+            return inmanta_type.Type()
+        else:
+            # TODO: unions
+            return inmanta_type.Type()
+        # bases: Sequence[inmanta.ast.type.Type] = [to_dsl_type(arg) for arg in typing.get_args(python_type)]
+        # return inmanta.ast.type.Union(bases)
+
+    # Lists and dicts
+    if typing_inspect.is_generic_type(python_type):
+        # List
+        origin = typing.get_origin(python_type)
+        if origin is Sequence:
+            # Can this fail?
+            base: inmanta_type.Type = typing.get_args(python_type)[0]
+            return inmanta_type.TypedList(to_dsl_type(base))
+
+        # dict
+        if issubclass(origin, collections.abc.Mapping):
+            args = typing_inspect.get_args(python_type)
+            assert len(args) == 2
+            if not issubclass(args[0], str):
+                raise TypingException(
+                    None, f"invalid type {python_type}, the keys of any dict should be 'str', got {args[0]} instead"
+                )
+            return inmanta_type.TypedDict(to_dsl_type(args[1]))
+
+    # 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 python_type in python_to_model:
+        return python_to_model[python_type]
+
+    return inmanta_type.Type()
+
 
 class PluginValue:
     """
@@ -267,15 +352,18 @@ def resolve_type(self, plugin: "Plugin", resolver: Namespace) -> inmanta_type.Ty
             return self._resolved_type
 
         if not isinstance(self.type_expression, str):
-            raise RuntimeException(
-                stmt=None,
-                msg="Bad annotation in plugin %s for %s, expected str but got %s (%s)"
-                % (plugin.get_full_name(), self.VALUE_NAME, type(self.type_expression).__name__, self.type_expression),
-            )
-
-        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)
+            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)
+            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)
         return self._resolved_type
 
     def validate(self, value: object) -> bool:

tests/compiler/test_plugin_types.py

@@ -0,0 +1,25 @@
+import collections.abc
+from typing import Any, Mapping, Union
+
+import pytest
+
+import inmanta.ast.type as inmanta_type
+from inmanta.ast import RuntimeException
+from inmanta.plugins import Null, to_dsl_type
+
+
+def test_conversion():
+    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.LiteralList() == to_dsl_type(list)
+    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.AnyType)
+
+    with pytest.raises(RuntimeException):
+        to_dsl_type(dict[int, int])

tests/compiler/test_plugins.py

@@ -416,3 +416,25 @@ def test_context_and_defaults(snippetcompiler: "SnippetCompilationTest") -> None
         """
     )
     compiler.do_compile()
+
+
+def test_native_types(snippetcompiler: "SnippetCompilationTest") -> None:
+    """
+    test the use of python types
+    """
+    snippetcompiler.setup_for_snippet(
+        """
+import plugin_native_types
+
+a = "b"
+a = plugin_native_types::get_from_dict({"a":"b"}, "a")
+
+none = null
+none = plugin_native_types::get_from_dict({"a":"b"}, "B")
+
+a = plugin_native_types::many_arguments(["a","c","b"], 1)
+
+none = plugin_native_types::as_none("a")
+        """
+    )
+    compiler.do_compile()

tests/data/modules/plugin_native_types/module.yml

@@ -0,0 +1,3 @@
+name: plugin_native_types
+license: Apache 2.0
+version: 1.0.0

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

@@ -0,0 +1,34 @@
+"""
+    Copyright 2023 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]
+"""
+
+from inmanta.plugins import plugin
+
+
+@plugin
+def get_from_dict(value: dict[str, str], key: str) -> str | None:
+    return value.get(key)
+
+
+@plugin
+def many_arguments(il: list[str], idx: int) -> str:
+    return sorted(il)[idx]
+
+
+@plugin
+def as_none(value: str) -> None:
+    pass