handle forward ref and self-reference in JsonSchemaGenerator, add rule resolve_forward_refs for __origin__, add volatile param for TypeRegistry, fix NotImplementedError in Schema cls

Author: voidZXL

tests/test_spec.py

@@ -1,14 +1,90 @@
+import utype
 from utype.types import *
 from utype.parser.rule import Rule
 
 
+class TargetSchema(utype.Schema):
+    name: str
+    ref: Optional['RefSchema'] = None
+    ref_values: List['RefSchema'] = utype.Field(default_factory=list)
+
+
+class RefSchema(utype.Schema):
+    value: int = None
+
+
+class InfiniteSchema(utype.Schema):
+    name: str
+    self: List['InfiniteSchema'] = utype.Field(default_factory=list)
+
+
 class TestSpec:
     def test_json_schema_parser(self):
         from utype.specs.json_schema.parser import JsonSchemaParser
-        from utype.specs.python.generator import PythonCodeGenerator
         assert JsonSchemaParser({})() == Any
         assert JsonSchemaParser({'anyOf': [{}, {'type': 'null'}]})() in (Rule, Any)
         assert JsonSchemaParser({'type': 'object'})() == dict
         assert JsonSchemaParser({'type': 'array'})() == list
         assert JsonSchemaParser({'type': 'string'})() == str
         assert JsonSchemaParser({'type': 'string', 'format': 'date'})() == date
+
+    def test_schema_generator(self):
+        class TestSchema(utype.Schema):
+            int_val: int
+            str_val: str
+            bytes_val: bytes
+            float_val: float
+            bool_val: bool
+            uuid_val: UUID
+            # test nest types
+            list_val: List[str] = utype.Field(default_factory=list)  # test callable default
+            union_val: Union[int, List[int]] = utype.Field(default_factory=list)
+
+        from utype.specs.json_schema.generator import JsonSchemaGenerator
+        output = JsonSchemaGenerator(TestSchema)()
+        assert output == {'type': 'object',
+                          'properties': {'int_val': {'type': 'integer'},
+                                         'str_val': {'type': 'string'},
+                                         'bytes_val': {'type': 'string', 'format': 'binary'},
+                                         'float_val': {'type': 'number', 'format': 'float'},
+                                         'bool_val': {'type': 'boolean'},
+                                         'uuid_val': {'type': 'string', 'format': 'uuid'},
+                                         'list_val': {'type': 'array', 'items': {'type': 'string'}},
+                                         'union_val': {'anyOf': [{'type': 'integer'},
+                                                                 {'type': 'array', 'items': {'type': 'integer'}}]}},
+                          'required': ['int_val',
+                                       'str_val',
+                                       'bytes_val',
+                                       'float_val',
+                                       'bool_val',
+                                       'uuid_val']}
+
+    def test_forward_ref_generator(self):
+        from utype.specs.json_schema.generator import JsonSchemaGenerator
+        output = JsonSchemaGenerator(TargetSchema)()
+        assert output == {'type': 'object',
+                          'properties': {'name': {'type': 'string'},
+                                         'ref': {'anyOf': [{'type': 'object',
+                                                            'properties': {'value': {'type': 'integer'}}},
+                                                           {'type': 'null'}]},
+                                         'ref_values': {'type': 'array',
+                                                        'items': {'type': 'object',
+                                                                  'properties': {'value': {'type': 'integer'}}}}},
+                          'required': ['name']}
+
+    def test_recursive_generator(self):
+        from utype.specs.json_schema.generator import JsonSchemaGenerator
+        refs = {}
+        ref_output = JsonSchemaGenerator(InfiniteSchema, defs=refs)()
+        assert ref_output == {'$ref': '#/$defs/InfiniteSchema'}
+        assert refs == {InfiniteSchema: {'type': 'object',
+                                         'properties': {'name': {'type': 'string'},
+                                                        'self': {'type': 'array',
+                                                                 'items': {'$ref': '#/$defs/InfiniteSchema'}}},
+                                         'required': ['name']}}
+
+        output = JsonSchemaGenerator(InfiniteSchema)()
+        assert output == {'type': 'object',
+                          'properties': {'name': {'type': 'string'},
+                                         'self': {'type': 'array', 'items': {'$ref': '#/$defs/InfiniteSchema'}}},
+                          'required': ['name']}

utype/parser/rule.py

@@ -177,19 +177,31 @@ def _parse_arg(mcs, arg):
         return Rule.annotate(__origin, *_new_args)
 
     def resolve_forward_refs(cls):
-        if not cls.combinator:
-            return
+        # if not cls.combinator:
+        #     return
+        origin_resolved = False
+        arg_resolved = False
+        origin = getattr(cls, "__origin__", None)
+
+        if origin is not None:
+            origin, origin_resolved = resolve_forward_type(origin)
+
+            if origin_resolved:
+                setattr(cls, "__origin__", cls._parse_arg(origin))
+
         args = []
-        resolved = False
         for i, arg in enumerate(cls.args):
             arg, resolved = resolve_forward_type(arg)
             if resolved:
                 arg = cls._parse_arg(arg)
+                arg_resolved = True
             args.append(arg)
-        if resolved:
+
+        if arg_resolved:
             # only adjust args if resolved
             setattr(cls, "__args__", tuple(args))
-        return resolved
+
+        return origin_resolved or arg_resolved
 
     def register_forward_refs(
         cls,
@@ -1322,6 +1334,7 @@ def annotate(
         constraints: Dict[str, Any] = None,
         global_vars: Dict[str, Any] = None,
         forward_refs=None,
+        forward_key=None,
         force_clear_refs=False,
         options=None,
         bound=None,
@@ -1378,6 +1391,7 @@ def annotate(
                     global_vars=global_vars,
                     forward_refs=forward_refs,
                     force_clear_refs=force_clear_refs,
+                    forward_key=forward_key,
                     bound=bound
                 )
                 # this annotation can be a ForwardRef
@@ -1508,6 +1522,7 @@ def parse_annotation(
                 forward_refs=forward_refs,
                 global_vars=global_vars,
                 force_clear_refs=force_clear_refs,
+                forward_key=forward_key,
                 bound=bound
             )
         elif annotation:
@@ -1533,6 +1548,7 @@ def parse_annotation(
                 forward_refs=forward_refs,
                 global_vars=global_vars,
                 force_clear_refs=force_clear_refs,
+                forward_key=forward_key,
                 bound=bound
             )
         return None
@@ -1845,34 +1861,42 @@ def _parse_contains(cls, value, context: RuntimeContext):
     @classmethod
     def resolve_forward_refs(cls):
         # an override version of LogicalType.resolve_forward_refs
-        if not cls.__args__:
-            return False
-        args = []
-        arg_transformers = []
-        resolved = False
-        for arg, trans in zip(cls.__args__, cls.__arg_transformers__):
-            if isinstance(arg, LogicalType):
-                # including the Rule class and LogicalType with combinator
-                if arg.resolve_forward_refs():
-                    resolved = True
-            elif isinstance(arg, ForwardRef):
-                if arg.__forward_evaluated__:
-                    arg = arg.__forward_value__
-                    resolved = True
-                    transformer = cls.transformer_cls.resolver_transformer(arg)
-                    if not transformer:
-                        warning_settings.warn(
-                            f"{cls}: arg type: {arg} got no transformer resolved, "
-                            f"will just pass {arg}(data) at runtime",
-                            warning_settings.rule_no_arg_transformer
-                        )
-                    trans = transformer or trans
-            args.append(arg)
-            arg_transformers.append(trans)
-        if resolved:
-            cls.__args__ = tuple(args)
-            cls.__arg_transformers__ = tuple(arg_transformers)
-        return resolved
+        origin_resolved = False
+        args_resolved = False
+
+        if cls.__origin__:
+            origin, origin_resolved = resolve_forward_type(cls.__origin__)
+            if origin_resolved:
+                cls.__origin__ = origin
+
+        if cls.__args__:
+            args = []
+            arg_transformers = []
+
+            for arg, trans in zip(cls.__args__, cls.__arg_transformers__):
+                if isinstance(arg, LogicalType):
+                    # including the Rule class and LogicalType with combinator
+                    if arg.resolve_forward_refs():
+                        args_resolved = True
+                elif isinstance(arg, ForwardRef):
+                    if arg.__forward_evaluated__:
+                        arg = arg.__forward_value__
+                        args_resolved = True
+                        transformer = cls.transformer_cls.resolver_transformer(arg)
+                        if not transformer:
+                            warning_settings.warn(
+                                f"{cls}: arg type: {arg} got no transformer resolved, "
+                                f"will just pass {arg}(data) at runtime",
+                                warning_settings.rule_no_arg_transformer
+                            )
+                        trans = transformer or trans
+                args.append(arg)
+                arg_transformers.append(trans)
+            if args_resolved:
+                cls.__args__ = tuple(args)
+                cls.__arg_transformers__ = tuple(arg_transformers)
+
+        return origin_resolved or args_resolved
 
     @classmethod
     def resolve_args_parser(cls):
@@ -2058,6 +2082,6 @@ def transform_callable(transformer: TypeTransformer, value, t):
         return value
 
 
-@TypeTransformer.registry.register(metaclass=LogicalType)
+@TypeTransformer.registry.register(metaclass=LogicalType, volatile=True)
 def transform_rule(transformer: TypeTransformer, value, t: LogicalType):
     return t(value, context=transformer.context)

utype/schema.py

@@ -95,7 +95,7 @@ def __name__(self):
         return self.__parser__.name
 
     def __class_getitem__(cls, item):
-        raise NotImplemented
+        raise NotImplementedError
 
     def __validate__(self):
         pass
@@ -184,7 +184,7 @@ def __class_getitem__(cls, item):
         #     class _cls(cls):
         #         __options__ = item
         #     return _cls
-        raise NotImplemented
+        raise NotImplementedError
 
     def __validate__(self):
         pass