Further attempts at resolving types because in a Cython environment we still get ForwardRef instances we shouldn't be getting

Author: alessiostalla

pylasu/emf/metamodel_builder.py

@@ -10,7 +10,7 @@
 from pylasu.emf.model import find_eclassifier
 from pylasu.model import Node
 from pylasu.model.model import InternalField
-from pylasu.reflection import getannotations
+from pylasu.reflection import get_type_annotations
 from pylasu.reflection.reflection import get_type_origin, is_enum_type, is_sequence_type, get_type_arguments
 
 
@@ -91,7 +91,7 @@ def setup_base_classes(self, cls):
         return bases
 
     def setup_attributes(self, cls):
-        anns = getannotations(cls)
+        anns = get_type_annotations(cls)
         nmspc = {
             "position": EReference("position", starlasu.Position, containment=True)
         }

pylasu/model/model.py

@@ -1,14 +1,19 @@
+import dataclasses
 import inspect
+import sys
+import typing
 from abc import ABC, abstractmethod, ABCMeta
 from dataclasses import Field, MISSING, dataclass, field
 from typing import Optional, Callable, List, Union
 
 from .naming import ReferenceByName
 from .position import Position, Source
 from .reflection import Multiplicity, PropertyDescription
-from ..reflection import getannotations, get_type_arguments, is_sequence_type
+from ..reflection import get_type_annotations, get_type_arguments, is_sequence_type
 from ..reflection.reflection import get_type_origin
 
+PYLASU_FEATURE = "pylasu_feature"
+
 
 class internal_property(property):
     pass
@@ -40,6 +45,24 @@ def internal_field(
         return InternalField(default, default_factory, init, repr, hash, compare, metadata)
 
 
+def node_property(default=MISSING):
+    description = PropertyDescription(
+        "", None,
+        multiplicity=Multiplicity.OPTIONAL if default is None else Multiplicity.SINGULAR)
+    return field(default=default, metadata={PYLASU_FEATURE: description})
+
+
+def node_containment(multiplicity: Multiplicity = Multiplicity.SINGULAR):
+    description = PropertyDescription("", None, is_containment=True, multiplicity=multiplicity)
+
+    if multiplicity == Multiplicity.SINGULAR:
+        return field(metadata={PYLASU_FEATURE: description})
+    elif multiplicity == Multiplicity.OPTIONAL:
+        return field(default=None, metadata={PYLASU_FEATURE: description})
+    elif multiplicity == Multiplicity.MANY:
+        return field(default_factory=list, metadata={PYLASU_FEATURE: description})
+
+
 class Origin(ABC):
     @internal_property
     @abstractmethod
@@ -105,15 +128,58 @@ def get_only_type_arg(decl_type):
         return None
 
 
-def process_annotated_property(name, decl_type, known_property_names):
-    multiplicity = Multiplicity.SINGULAR
-    is_reference = False
+def process_annotated_property(cl: type, name: str, decl_type):
+    try:
+        fields = dataclasses.fields(cl)
+    except TypeError:
+        fields = tuple()
+    for field in fields:
+        if field.name == name and PYLASU_FEATURE in field.metadata:
+            feature = field.metadata[PYLASU_FEATURE]
+            feature.name = name
+            if isinstance(decl_type, type):
+                feature.type = decl_type
+            elif type(field.type) is str:
+                feature.type = try_to_resolve_string_type(field.type, name, cl)
+            return feature
+    return compute_feature_from_annotation(cl, name, decl_type)
+
+
+def compute_feature_from_annotation(cl, name, decl_type):
+    feature = PropertyDescription(name, None, False, False, Multiplicity.SINGULAR)
+    decl_type = try_to_resolve_type(decl_type, feature)
+    if not isinstance(decl_type, type):
+        fwref = None
+        if hasattr(typing, "ForwardRef"):
+            fwref = typing.ForwardRef
+        if fwref and isinstance(decl_type, fwref):
+            raise Exception(f"Feature {name}'s type is unresolved forward reference {decl_type}, "
+                            f"please use node_containment or node_property")
+        elif type(decl_type) is str:
+            decl_type = try_to_resolve_string_type(decl_type, name, cl)
+        if not isinstance(decl_type, type):
+            raise Exception(f"Unsupported feature {name} of type {decl_type}")
+    feature.type = decl_type
+    feature.is_containment = provides_nodes(decl_type) and not feature.is_reference
+    return feature
+
+
+def try_to_resolve_string_type(decl_type, name, cl):
+    try:
+        ns = getattr(sys.modules.get(cl.__module__, None), '__dict__', globals())
+        decl_type = ns[decl_type]
+    except KeyError:
+        raise Exception(f"Unsupported feature {name} of unknown type {decl_type}")
+    return decl_type
+
+
+def try_to_resolve_type(decl_type, feature):
     if get_type_origin(decl_type) is ReferenceByName:
         decl_type = get_only_type_arg(decl_type) or decl_type
-        is_reference = True
+        feature.is_reference = True
     if is_sequence_type(decl_type):
         decl_type = get_only_type_arg(decl_type) or decl_type
-        multiplicity = Multiplicity.MANY
+        feature.multiplicity = Multiplicity.MANY
     if get_type_origin(decl_type) is Union:
         type_args = get_type_arguments(decl_type)
         if len(type_args) == 1:
@@ -124,16 +190,12 @@ def process_annotated_property(name, decl_type, known_property_names):
             elif type_args[1] is type(None):
                 decl_type = type_args[0]
             else:
-                raise Exception(f"Unsupported feature {name} of type {decl_type}")
-            if multiplicity == Multiplicity.SINGULAR:
-                multiplicity = Multiplicity.OPTIONAL
+                raise Exception(f"Unsupported feature {feature.name} of union type {decl_type}")
+            if feature.multiplicity == Multiplicity.SINGULAR:
+                feature.multiplicity = Multiplicity.OPTIONAL
         else:
-            raise Exception(f"Unsupported feature {name} of type {decl_type}")
-    if not isinstance(decl_type, type):
-        raise Exception(f"Unsupported feature {name} of type {decl_type}")
-    is_containment = provides_nodes(decl_type) and not is_reference
-    known_property_names.add(name)
-    return PropertyDescription(name, decl_type, is_containment, is_reference, multiplicity)
+            raise Exception(f"Unsupported feature {feature.name} of union type {decl_type}")
+    return decl_type
 
 
 class Concept(ABCMeta):
@@ -155,16 +217,21 @@ def node_properties(cls):
             yield from cls._direct_node_properties(cl, names)
 
     def _direct_node_properties(cls, cl, known_property_names):
-        anns = getannotations(cl)
+        if not isinstance(cls, Concept):
+            return
+        anns = get_type_annotations(cl)
         if not anns:
             return
         for name in anns:
             if name not in known_property_names and cls.is_node_property(name):
-                yield process_annotated_property(name, anns[name], known_property_names)
+                feature = process_annotated_property(cl, name, anns[name])
+                known_property_names.add(name)
+                yield feature
         for name in dir(cl):
             if name not in known_property_names and cls.is_node_property(name):
+                feature = PropertyDescription(name, None, False, False)
                 known_property_names.add(name)
-                yield PropertyDescription(name, None, False, False)
+                yield feature
 
     def is_node_property(cls, name):
         return not name.startswith('_') and name not in cls.__internal_properties__

pylasu/model/reflection.py

@@ -13,8 +13,8 @@ class Multiplicity(enum.Enum):
 class PropertyDescription:
     name: str
     type: Optional[type]
-    is_containment: bool
-    is_reference: bool
+    is_containment: bool = False
+    is_reference: bool = False
     multiplicity: Multiplicity = Multiplicity.SINGULAR
     value: object = None
 

pylasu/reflection/__init__.py

@@ -1 +1 @@
-from .reflection import getannotations, get_type_arguments, is_sequence_type
+from .reflection import get_type_annotations, get_type_arguments, is_sequence_type

pylasu/reflection/reflection.py

@@ -3,20 +3,24 @@
 from typing import Callable
 
 
-def getannotations(cls):
-    try:
+def get_type_annotations(cls: type):
+    if hasattr(typing, "get_type_hints"):
         # https://peps.python.org/pep-0563/
-        return typing.get_type_hints(cls, globalns=None, localns=None)
-    except AttributeError:
+        return typing.get_type_hints(cls)
+    else:
         try:
             # On Python 3.10+
             import inspect
-            return inspect.getannotations(cls)
-        except AttributeError:
-            if isinstance(cls, type):
-                return cls.__dict__.get('__annotations__', None)
-            else:
-                return getattr(cls, '__annotations__', None)
+            if hasattr(inspect, "get_annotations"):
+                return inspect.get_annotations(cls)
+            elif hasattr(inspect, "getannotations"):
+                return inspect.getannotations(cls)
+        except ModuleNotFoundError:
+            pass
+        if isinstance(cls, type):
+            return cls.__dict__.get('__annotations__', {})
+        else:
+            return getattr(cls, '__annotations__', {})
 
 
 def get_type_origin(tp):

tests/model/test_model.py

@@ -3,6 +3,7 @@
 from typing import List, Optional, Union
 
 from pylasu.model import Node, Position, Point, internal_field
+from pylasu.model.model import node_property, node_containment
 from pylasu.model.reflection import Multiplicity, PropertyDescription
 from pylasu.model.naming import ReferenceByName, Named, Scope, Symbol
 from pylasu.support import extension_method
@@ -32,6 +33,8 @@ class ExtendedNode(SomeNode):
     multiple2: List[SomeNode] = dataclasses.field(default_factory=list)
     multiple_fwd: List["ForwardReferencedNode"] = dataclasses.field(default_factory=list)
     internal2: Node = internal_field(default=None)
+    explicit_property: str = node_property("42")
+    explicit_containment: "ExtendedNode" = node_containment(Multiplicity.MANY)
 
 
 @dataclasses.dataclass
@@ -234,7 +237,7 @@ def frob_node_2(_: Node):
             pass
 
         pds = [pd for pd in sorted(ExtendedNode.node_properties, key=lambda x: x.name)]
-        self.assertEqual(13, len(pds), f"{pds} should be 7")
+        self.assertEqual(15, len(pds), f"{pds}")
         self.assertEqual("bar", pds[0].name)
         self.assertFalse(pds[0].is_containment)
         self.assertEqual("cont_fwd", pds[1].name)
@@ -245,16 +248,23 @@ def frob_node_2(_: Node):
         self.assertEqual(ForwardReferencedNode, pds[2].type)
         self.assertEqual("containment", pds[3].name)
         self.assertTrue(pds[3].is_containment)
-        self.assertEqual("foo", pds[4].name)
-        self.assertEqual("multiple", pds[5].name)
-        self.assertTrue(pds[5].is_containment)
-        self.assertEqual(Multiplicity.MANY, pds[5].multiplicity)
-        self.assertEqual("multiple2", pds[6].name)
-        self.assertTrue(pds[6].is_containment)
-        self.assertEqual(Multiplicity.MANY, pds[6].multiplicity)
-        self.assertEqual("multiple_fwd", pds[7].name)
+        self.assertEqual("explicit_containment", pds[4].name)
+        self.assertTrue(pds[4].is_containment)
+        self.assertEqual(ExtendedNode, pds[4].type)
+        self.assertEqual(Multiplicity.MANY, pds[4].multiplicity)
+        self.assertEqual("explicit_property", pds[5].name)
+        self.assertEqual(str, pds[5].type)
+        self.assertEqual(Multiplicity.SINGULAR, pds[5].multiplicity)
+        self.assertEqual("foo", pds[6].name)
+        self.assertEqual("multiple", pds[7].name)
         self.assertTrue(pds[7].is_containment)
-        self.assertEqual(ForwardReferencedNode, pds[7].type)
         self.assertEqual(Multiplicity.MANY, pds[7].multiplicity)
+        self.assertEqual("multiple2", pds[8].name)
+        self.assertTrue(pds[8].is_containment)
+        self.assertEqual(Multiplicity.MANY, pds[8].multiplicity)
+        self.assertEqual("multiple_fwd", pds[9].name)
+        self.assertTrue(pds[9].is_containment)
+        self.assertEqual(ForwardReferencedNode, pds[9].type)
+        self.assertEqual(Multiplicity.MANY, pds[9].multiplicity)
 
         self.assertRaises(Exception, lambda: [x for x in InvalidNode.node_properties])