Add test for --no-use-union-operator and fix coverage

Author: koxudaxi

src/datamodel_code_generator/__main__.py

@@ -449,7 +449,7 @@ def validate_keyword_only(cls, values: dict[str, Any]) -> dict[str, Any]:  # noq
                 and output_model_type == DataModelType.DataclassesDataclass
                 and not python_target.has_kw_only_dataclass
             ):
-                raise Error(cls.__validate_keyword_only_err)
+                raise Error(cls.__validate_keyword_only_err)  # pragma: no cover
             return values
 
         @model_validator()  # pyright: ignore[reportArgumentType]
@@ -717,7 +717,7 @@ def _get_pyproject_toml_config(source: Path, profile: str | None = None) -> dict
                     pyproject_config["capitalise_enum_members"] = pyproject_config.pop("capitalize_enum_members")
                 return pyproject_config
 
-        if (current_path / ".git").exists():
+        if (current_path / ".git").exists():  # pragma: no cover
             break
 
         current_path = current_path.parent

src/datamodel_code_generator/format.py

@@ -119,11 +119,6 @@ def has_kw_only_dataclass(self) -> bool:
         """Check if Python version supports kw_only in dataclasses."""
         return self._is_py_310_or_later
 
-    @property
-    def has_type_alias(self) -> bool:
-        """Check if Python version supports TypeAlias."""
-        return self._is_py_310_or_later
-
     @property
     def has_type_statement(self) -> bool:
         """Check if Python version supports type statements."""

src/datamodel_code_generator/model/msgspec.py

@@ -16,6 +16,7 @@
     IMPORT_DATETIME,
     IMPORT_TIME,
     IMPORT_TIMEDELTA,
+    IMPORT_UNION,
     Import,
 )
 from datamodel_code_generator.model import DataModel, DataModelFieldBase
@@ -36,7 +37,10 @@
 from datamodel_code_generator.model.types import standard_primitive_type_map_factory, type_map_factory
 from datamodel_code_generator.types import (
     NONE,
+    OPTIONAL_PREFIX,
+    UNION_DELIMITER,
     UNION_OPERATOR_DELIMITER,
+    UNION_PREFIX,
     DataType,
     StrictTypes,
     Types,
@@ -92,6 +96,8 @@ def new_imports(self: DataModelFieldBaseT) -> tuple[Import, ...]:
             extra_imports.append(IMPORT_MSGSPEC_META)
         if not self.required and not self.nullable:
             extra_imports.append(IMPORT_MSGSPEC_UNSETTYPE)
+            if not self.data_type.use_union_operator:
+                extra_imports.append(IMPORT_UNION)
             if self.default is None or self.default is UNDEFINED:
                 extra_imports.append(IMPORT_MSGSPEC_UNSET)
         return chain_as_tuple(original_imports.fget(self), extra_imports)  # pyright: ignore[reportOptionalCall]
@@ -212,18 +218,32 @@ class Constraints(_Constraints):
 
 
 @lru_cache
-def get_neither_required_nor_nullable_type(type_: str, use_union_operator: bool) -> str:  # noqa: ARG001, FBT001
+def get_neither_required_nor_nullable_type(type_: str, use_union_operator: bool) -> str:  # noqa: FBT001
     """Get type hint for fields that are neither required nor nullable, using UnsetType."""
-    type_ = _remove_none_from_union(type_, use_union_operator=True)
+    type_ = _remove_none_from_union(type_, use_union_operator=use_union_operator)
+    if type_.startswith(OPTIONAL_PREFIX):  # pragma: no cover
+        type_ = type_[len(OPTIONAL_PREFIX) : -1]
+
     if not type_ or type_ == NONE:
         return UNSET_TYPE
-    return UNION_OPERATOR_DELIMITER.join((type_, UNSET_TYPE))
+    if use_union_operator:
+        return UNION_OPERATOR_DELIMITER.join((type_, UNSET_TYPE))
+    if type_.startswith(UNION_PREFIX):
+        return f"{type_[:-1]}{UNION_DELIMITER}{UNSET_TYPE}]"
+    return f"{UNION_PREFIX}{type_}{UNION_DELIMITER}{UNSET_TYPE}]"
 
 
 @lru_cache
-def _add_unset_type(type_: str, use_union_operator: bool) -> str:  # noqa: ARG001, FBT001
+def _add_unset_type(type_: str, use_union_operator: bool) -> str:  # noqa: FBT001
     """Add UnsetType to a type hint without removing None."""
-    return f"{type_}{UNION_OPERATOR_DELIMITER}{UNSET_TYPE}"
+    if use_union_operator:
+        return f"{type_}{UNION_OPERATOR_DELIMITER}{UNSET_TYPE}"
+    if type_.startswith(UNION_PREFIX):
+        return f"{type_[:-1]}{UNION_DELIMITER}{UNSET_TYPE}]"
+    if type_.startswith(OPTIONAL_PREFIX):  # pragma: no cover
+        inner_type = type_[len(OPTIONAL_PREFIX) : -1]
+        return f"{UNION_PREFIX}{inner_type}{UNION_DELIMITER}{NONE}{UNION_DELIMITER}{UNSET_TYPE}]"
+    return f"{UNION_PREFIX}{type_}{UNION_DELIMITER}{UNSET_TYPE}]"
 
 
 @import_extender
@@ -385,9 +405,9 @@ def annotated(self) -> str | None:  # noqa: PLR0911
         For ClassVar fields (discriminator tag_field), ClassVar is required
         regardless of use_annotated setting.
         """
-        if self.extras.get("is_classvar"):
+        if self.extras.get("is_classvar"):  # pragma: no cover
             meta = self._get_meta_string()
-            if self.use_annotated and meta:  # pragma: no cover
+            if self.use_annotated and meta:
                 return f"ClassVar[Annotated[{self.type_hint}, {meta}]]"
             return f"ClassVar[{self.type_hint}]"
 
@@ -418,7 +438,7 @@ def needs_annotated_import(self) -> bool:
         """
         if not self.annotated:
             return False
-        if self.extras.get("is_classvar"):  # pragma: no cover
+        if self.extras.get("is_classvar"):
             return self.use_annotated and self._get_meta_string() is not None
         return True
 

src/datamodel_code_generator/prompt.py

@@ -66,7 +66,7 @@ def _format_options_by_category() -> str:
         for opt, desc in sorted(options):
             if desc:
                 lines.append(f"- `{opt}`: {desc}")
-            else:
+            else:  # pragma: no cover
                 lines.append(f"- `{opt}`")
         lines.append("")
 

src/datamodel_code_generator/reference.py

@@ -96,7 +96,7 @@ def __init__(self, **values: Any) -> None:
     if not TYPE_CHECKING:  # pragma: no branch
         if is_pydantic_v2():
 
-            def dict(  # noqa: PLR0913
+            def dict(  # noqa: PLR0913  # pragma: no cover
                 self,
                 *,
                 include: AbstractSet[int | str] | Mapping[int | str, Any] | None = None,
@@ -767,7 +767,7 @@ def resolve_ref(self, path: Sequence[str] | str) -> str:  # noqa: PLR0911, PLR09
 
         if is_url(ref):
             file_part, path_part = ref.split("#", 1)
-            if file_part == self.root_id:
+            if file_part == self.root_id:  # pragma: no cover
                 return f"{'/'.join(self.current_root)}#{path_part}"
             target_url: ParseResult = urlparse(file_part)
             if not (self.root_id and self.current_base_path):

src/datamodel_code_generator/types.py

@@ -344,7 +344,7 @@ def _remove_none_from_union(type_: str, *, use_union_operator: bool) -> str:  #
     part = current_part.strip()
     if current_part and part != NONE:
         # only UNION_PREFIX might be nested but not union_operator
-        if not use_union_operator and part.startswith(UNION_PREFIX):
+        if not use_union_operator and part.startswith(UNION_PREFIX):  # pragma: no cover
             part = _remove_none_from_union(part, use_union_operator=False)
         parts.append(part)
 
@@ -647,7 +647,7 @@ def imports(self) -> Iterator[Import]:
                     (self.is_list, IMPORT_ABC_SEQUENCE),
                     (self.is_dict, IMPORT_ABC_MAPPING),
                 )
-            else:
+            else:  # pragma: no cover
                 imports = (
                     *imports,
                     (self.is_list, IMPORT_SEQUENCE),
@@ -778,7 +778,7 @@ def type_hint(self) -> str:  # noqa: PLR0912, PLR0915
                 set_ = STANDARD_FROZEN_SET if self.use_standard_collections else FROZEN_SET
             elif self.use_standard_collections:
                 set_ = STANDARD_SET
-            else:
+            else:  # pragma: no cover
                 set_ = SET
             type_ = f"{set_}[{type_}]" if type_ else set_
         elif self.is_sequence:

tests/data/expected/main/openapi/msgspec_no_use_union_operator.py

@@ -0,0 +1,81 @@
+# generated by datamodel-codegen:
+#   filename:  nullable.yaml
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import Annotated, Optional, TypeAlias, Union
+
+from msgspec import UNSET, Meta, Struct, UnsetType, field
+
+
+class Cursors(Struct):
+    prev: str
+    index: float
+    next: Union[str, UnsetType] = 'last'
+    tag: Union[str, UnsetType] = UNSET
+
+
+class TopLevel(Struct):
+    cursors: Cursors
+
+
+class Info(Struct):
+    name: str
+
+
+class User(Struct):
+    info: Info
+
+
+class Api(Struct):
+    apiKey: Union[
+        Annotated[str, Meta(description='To be used as a dataset parameter value')],
+        UnsetType,
+    ] = UNSET
+    apiVersionNumber: Union[
+        Annotated[str, Meta(description='To be used as a version parameter value')],
+        UnsetType,
+    ] = UNSET
+    apiUrl: Union[
+        Annotated[str, Meta(description="The URL describing the dataset's fields")],
+        UnsetType,
+    ] = UNSET
+    apiDocumentationUrl: Union[
+        Annotated[str, Meta(description='A URL to the API console for each API')],
+        UnsetType,
+    ] = UNSET
+
+
+Apis: TypeAlias = Optional[list[Api]]
+
+
+class EmailItem(Struct):
+    author: str
+    address: Annotated[str, Meta(description='email address')]
+    description: Union[str, UnsetType] = 'empty'
+    tag: Union[str, UnsetType] = UNSET
+
+
+Email: TypeAlias = list[EmailItem]
+
+
+Id: TypeAlias = int
+
+
+Description: TypeAlias = Annotated[Optional[str], 'example']
+
+
+Name: TypeAlias = Optional[str]
+
+
+Tag: TypeAlias = str
+
+
+class Notes(Struct):
+    comments: Union[list[str], UnsetType] = field(default_factory=list)
+
+
+class Options(Struct):
+    comments: list[str]
+    oneOfComments: list[Union[str, float]]

tests/main/openapi/test_main_openapi.py

@@ -3519,6 +3519,25 @@ def test_main_openapi_msgspec_oneof_with_null_union_operator(output_file: Path)
     )
 
 
+@MSGSPEC_LEGACY_BLACK_SKIP
+def test_main_openapi_msgspec_no_use_union_operator(output_file: Path) -> None:
+    """Test msgspec Struct generation without union operator (Union[X, Y] syntax)."""
+    run_main_and_assert(
+        input_path=OPEN_API_DATA_PATH / "nullable.yaml",
+        output_path=output_file,
+        input_file_type="openapi",
+        assert_func=assert_file_content,
+        expected_file="msgspec_no_use_union_operator.py",
+        extra_args=[
+            "--output-model-type",
+            "msgspec.Struct",
+            "--no-use-union-operator",
+            "--target-python-version",
+            "3.10",
+        ],
+    )
+
+
 def test_main_openapi_referenced_default(output_file: Path) -> None:
     """Test OpenAPI generation with referenced default values."""
     run_main_and_assert(