Update custom or_ to support union of a param model and its forward ref

Add 'self' to the illegal param name list

Author: yugokato

src/openapi_test_client/libraries/api/api_functions/utils/endpoint_model.py

@@ -75,8 +75,8 @@ def create_endpoint_model(endpoint_func: EndpointFunc, api_spec: dict[str, Any]
     # Some OpenAPI specs define a parameter name using characters we can't use as a python variable name.
     # We will use the cleaned name as the model field and annotate it as `Annotated[field_type, Alias(<original_val>)]`
     # When calling an endpoint function, the actual name will be automatically resolved in the payload/query parameters
-    param_model_util.alias_illegal_model_field_names(path_param_fields)
-    param_model_util.alias_illegal_model_field_names(body_or_query_param_fields)
+    param_model_util.alias_illegal_model_field_names(model_name, path_param_fields)
+    param_model_util.alias_illegal_model_field_names(model_name, body_or_query_param_fields)
 
     fields = path_param_fields + body_or_query_param_fields
     return cast(

src/openapi_test_client/libraries/api/api_functions/utils/param_model.py

@@ -69,6 +69,20 @@ def get_param_model(annotated_type: Any) -> type[ParamModel] | ForwardRef | list
             return base_type
 
 
+def get_param_model_name(param_model: type[ParamModel] | ForwardRef) -> str:
+    """Get the model name
+
+    :param param_model: Param model. This can be a forward ref
+    """
+    if not is_param_model(param_model):
+        raise ValueError(f"{param_model} is not a param model")
+
+    if isinstance(param_model, ForwardRef):
+        return param_model.__forward_arg__
+    else:
+        return param_model.__name__
+
+
 @lru_cache
 def generate_model_name(field_name: str, field_type: str | Any) -> str:
     """Generate model name from the given field
@@ -130,7 +144,7 @@ def create_model_from_param_def(
             )
             for inner_param_name, inner_param_obj in param_def.get("properties", {}).items()
         ]
-        alias_illegal_model_field_names(fields)
+        alias_illegal_model_field_names(model_name, fields)
         return cast(
             type[ParamModel],
             make_dataclass(
@@ -299,9 +313,10 @@ def visit(model_name: str) -> None:
     return sorted(models, key=lambda x: sorted_models_names.index(x.__name__))
 
 
-def alias_illegal_model_field_names(model_fields: list[DataclassModelField]) -> str:
+def alias_illegal_model_field_names(model_name: str, model_fields: list[DataclassModelField]) -> None:
     """Clean illegal model field name and annotate the field type with Alias class
 
+    :param model_name: Model name
     :param model_fields: fields value to be passed to make_dataclass()
     """
 
@@ -315,7 +330,7 @@ def make_alias(name: str, param_type: Any) -> str:
         else:
             name = clean_obj_name(name)
             # NOTE: The escaping of kwargs is already is handled in endpoint model
-            reserved_param_names = [*get_supported_request_parameters(), "validate"]
+            reserved_param_names = ["self", "validate", *get_supported_request_parameters()]
             if name in get_reserved_model_names() + reserved_param_names:
                 # The field name conflicts with one of reserved names
                 name += "_"

src/openapi_test_client/libraries/api/api_functions/utils/param_type.py

@@ -463,6 +463,11 @@ def merge_args_per_origin(args: Sequence[Any]) -> tuple[Any, ...]:
             for args_ in args_per_origin.values()
         )
 
+    if isinstance(tp1, str):
+        tp1 = ForwardRef(tp1)
+    if isinstance(tp2, str):
+        tp2 = ForwardRef(tp2)
+
     origin = get_origin(tp1)
     origin2 = get_origin(tp2)
     if origin or origin2:
@@ -528,7 +533,11 @@ def or_(x: Any, y: Any) -> Any:
     >>> reduce(or_, [Model1 | None, Model2])
     __main__.MyModel | None
     """
-    if param_model_util.is_param_model(x) and param_model_util.is_param_model(y) and x.__name__ == y.__name__:
+    if (
+        param_model_util.is_param_model(x)
+        and param_model_util.is_param_model(y)
+        and param_model_util.get_param_model_name(x) == param_model_util.get_param_model_name(y)
+    ):
         return x
     else:
         is_x_union = is_union_type(x)

tests/unit/test_param_type.py

@@ -468,6 +468,7 @@ def test_merge_annotation_types(tp1: Any, tp2: Any, expected_type: Any) -> None:
         (MyParamModel, MyParamModel2 | None, MyParamModel | None),
         (MyParamModel2 | None, MyParamModel3, MyParamModel2 | None),
         (MyParamModel2, MyParamModel3 | None, MyParamModel2 | None),
+        (MyParamModel, ForwardRef(MyParamModel.__name__), MyParamModel),
     ],
 )
 def test_custom_or_(tp1: Any, tp2: Any, expected_type: Any) -> None: