Prepare for pydanticv2

Author: ruscoder

fhir_py_types/ast.py

@@ -79,11 +79,13 @@ def format_identifier(
     def uppercamelcase(s: str) -> str:
         return s[:1].upper() + s[1:]
 
-    return (
-        identifier + uppercamelcase(type_.code)
-        if is_polymorphic(definition)
-        else identifier
-    )
+    if is_polymorphic(definition):
+        # TODO: it's fast hack
+        if type_.code[0].islower():
+            return identifier + uppercamelcase(clear_primitive_id(type_.code))
+        return identifier + uppercamelcase(type_.code)
+
+    return identifier
 
 
 def remap_type(
@@ -94,8 +96,8 @@ def remap_type(
             case "Resource":
                 # Different contexts use 'Resource' type to refer to any
                 # resource differentiated by its 'resourceType' (tagged union).
-                # 'AnyResource' is not defined by the spec but rather
-                # generated as a union of all defined resource types.
+                # 'AnyResource' is defined in header as a special type
+                # that dynamically replaced with a right type in run-time
                 type_ = replace(type_, code="AnyResource")
 
     if is_polymorphic(definition):
@@ -106,6 +108,12 @@ def remap_type(
         # with a custom validator that will enforce single required property rule.
         type_ = replace(type_, required=False)
 
+    if is_primitive_type(type_):
+        # Primitive types defined from the small letter (like code)
+        # and it might overlap with model fields
+        # e.g. QuestionnaireItem has attribute code and linkId has type code
+        type_ = replace(type_, code=make_primitive_id(type_.code))
+
     return type_
 
 
@@ -116,9 +124,8 @@ def zip_identifier_type(
 
     for t in [remap_type(definition, t) for t in definition.type]:
         result.append((format_identifier(definition, identifier, t), t))
-        if (
-            definition.kind != StructureDefinitionKind.PRIMITIVE
-            and is_primitive_type(t)
+        if definition.kind != StructureDefinitionKind.PRIMITIVE and is_primitive_type(
+            t
         ):
             result.append(
                 (
@@ -201,11 +208,6 @@ def define_class_object(
             keywords=[],
             type_params=[],
         ),
-        ast.Call(
-            ast.Attribute(value=ast.Name(definition.id), attr="update_forward_refs"),
-            args=[],
-            keywords=[],
-        ),
     ]
 
 
@@ -214,48 +216,22 @@ def define_class(definition: StructureDefinition) -> Iterable[ast.stmt | ast.exp
 
 
 def define_alias(definition: StructureDefinition) -> Iterable[ast.stmt]:
-    return type_annotate(definition, definition.id, AnnotationForm.TypeAlias)
-
-
-def define_tagged_union(
-    name: str, components: Iterable[StructureDefinition], distinct_by: str
-) -> ast.stmt:
-    annotation = functools.reduce(
-        lambda acc, n: ast.BinOp(left=acc, right=n, op=ast.BitOr()),
-        (cast(ast.expr, ast.Name(d.id)) for d in components),
+    # Primitive types are renamed to another name to avoid overlapping with model fields
+    return type_annotate(
+        definition, make_primitive_id(definition.id), AnnotationForm.TypeAlias
     )
 
-    return ast.Assign(
-        targets=[ast.Name(name)],
-        value=ast.Subscript(
-            value=ast.Name("Annotated_"),
-            slice=ast.Tuple(
-                elts=[
-                    annotation,
-                    ast.Call(
-                        ast.Name("Field"),
-                        args=[ast.Constant(...)],
-                        keywords=[
-                            ast.keyword(
-                                arg="discriminator", value=ast.Constant(distinct_by)
-                            ),
-                        ],
-                    ),
-                ]
-            ),
-        ),
-    )
 
+def make_primitive_id(name: str) -> str:
+    if name in ("str", "int", "float", "bool"):
+        return name
+    return f"{name}Type"
 
-def select_tagged_resources(
-    definitions: Iterable[StructureDefinition], key: str
-) -> Iterable[StructureDefinition]:
-    return (
-        definition
-        for definition in definitions
-        if definition.kind == StructureDefinitionKind.RESOURCE
-        and key in definition.elements
-    )
+
+def clear_primitive_id(name: str) -> str:
+    if name.endswith("Type"):
+        return name[:-4]
+    return name
 
 
 def select_nested_definitions(
@@ -301,14 +277,6 @@ def build_ast(
                         f"Unsupported definition {definition.id} of kind {definition.kind}, skipping"
                     )
 
-    resources = list(select_tagged_resources(structure_definitions, key="resourceType"))
-    if resources:
-        typedefinitions.append(
-            define_tagged_union(
-                name="AnyResource", components=resources, distinct_by="resourceType"
-            )
-        )
-
     return sorted(
         typedefinitions,
         # Defer any postprocessing until after the structure tree is defined.

fhir_py_types/header.py.tpl

@@ -1,14 +1,100 @@
-from typing import List as List_, Optional as Optional_, Literal as Literal_, Annotated as Annotated_
+import warnings
+from typing import List as List_, Optional as Optional_, Literal as Literal_
 
-from pydantic import BaseModel as BaseModel_, Field, Extra
+from pydantic import PydanticDeprecatedSince20
+from pydantic import (
+    BaseModel as BaseModel_,
+    Field,
+    Extra,
+    field_validator,
+    ValidationError,
+)
+from pydantic_core import PydanticCustomError
+
+
+class AnyResource(BaseModel_):
+    class Config:
+        extra = Extra.allow
+
+    resourceType: str
 
 
 class BaseModel(BaseModel_):
     class Config:
         extra = Extra.forbid
         validate_assignment = True
-        allow_population_by_field_name = True
+        populate_by_name = True
+        defer_build = True
+
+    def model_dump(self, *args, **kwargs):
+        by_alias = kwargs.pop("by_alias", True)
+        return super().model_dump(*args, **kwargs, by_alias=by_alias)
 
     def dict(self, *args, **kwargs):
-        by_alias = kwargs.pop('by_alias', True)
-        return super().dict(*args, **kwargs, by_alias=by_alias)
+        warnings.warn(
+            "The `dict` method is deprecated; use `model_dump` instead.",
+            category=PydanticDeprecatedSince20,
+        )
+        by_alias = kwargs.pop("by_alias", True)
+        return super().model_dump(*args, **kwargs, by_alias=by_alias)
+
+    @field_validator("*")
+    @classmethod
+    def validate(cls, value):
+        if isinstance(value, list):
+            return [
+                _init_any_resource(v, index=index) for index, v in enumerate(value)
+            ]
+        return _init_any_resource(value)
+
+
+def _init_any_resource(value, index=None):
+    if isinstance(value, AnyResource):
+        try:
+            klass = globals().get(value.resourceType)
+        except PydanticCustomError as exc:
+            raise ValidationError.from_exception_data(
+                "ImportError",
+                [
+                    {
+                        "loc": [index, "resourceType"],
+                        "type": "value_error",
+                        "msg": f"{value.resourceType} resource is not found",
+                        "input": [value],
+                        "ctx": {
+                            "error": f"{value.resourceType} resource is not found"
+                        },
+                    }
+                ],
+            ) from exc
+
+        if (
+            not issubclass(klass, BaseModel)
+            or "resourceType" not in klass.__fields__
+        ):
+            raise ValidationError.from_exception_data(
+                "ImportError",
+                [
+                    {
+                        "loc": [index, "resourceType"],
+                        "type": "value_error",
+                        "msg": f"{value.resourceType} is not a resource",
+                        "input": [value],
+                        "ctx": {
+                            "error": f"{value.resourceType} is not a resource"
+                        },
+                    }
+                ],
+            )
+
+        try:
+            return klass(**value.model_dump())
+        except ValidationError as exc:
+            raise ValidationError.from_exception_data(
+                exc.title,
+                [{**error, "loc": [index, *error["loc"]]} for error in exc.errors()]
+                if index is not None
+                else exc.errors(),
+            ) from exc
+
+    return value