Fix msgspec discriminated union ClassVar generation and nullable type handling (#2620)

* Add support for --ignore-pyproject and --profile options in CLI

* Fix discriminator handling for ClassVar and add nullable checks in msgspec

* Add support for discriminator ClassVar with Meta constraints in msgspec

Author: koxudaxi

src/datamodel_code_generator/model/base.py

@@ -217,7 +217,7 @@ def imports(self) -> tuple[Import, ...]:
 
         if has_optional:
             imports.append((IMPORT_OPTIONAL,))
-        if self.use_annotated and self.annotated:
+        if self.use_annotated and self.needs_annotated_import:
             imports.append((IMPORT_ANNOTATED,))
         return chain_as_tuple(*imports)
 
@@ -269,6 +269,16 @@ def annotated(self) -> str | None:
         """Get the Annotated type hint content, if any."""
         return None
 
+    @property
+    def needs_annotated_import(self) -> bool:
+        """Check if this field requires the Annotated import."""
+        return bool(self.annotated)
+
+    @property
+    def needs_meta_import(self) -> bool:  # pragma: no cover
+        """Check if this field requires the Meta import (msgspec only)."""
+        return False
+
     @property
     def has_default_factory(self) -> bool:
         """Check if this field has a default_factory."""

src/datamodel_code_generator/model/msgspec.py

@@ -89,7 +89,7 @@ def new_imports(self: DataModelFieldBaseT) -> tuple[Import, ...]:
             extra_imports.append(IMPORT_MSGSPEC_FIELD)
         if self.field and "lambda: convert" in self.field:
             extra_imports.append(IMPORT_MSGSPEC_CONVERT)
-        if self.annotated:
+        if isinstance(self, DataModelField) and self.needs_meta_import:
             extra_imports.append(IMPORT_MSGSPEC_META)
         if self.extras.get("is_classvar"):
             extra_imports.append(IMPORT_CLASSVAR)
@@ -281,7 +281,7 @@ def __str__(self) -> str:
     def type_hint(self) -> str:
         """Return the type hint, using UnsetType for non-required non-nullable fields."""
         type_hint = super().type_hint
-        if self._not_required and not self.nullable:
+        if self._not_required and not self.nullable and not self.data_type.is_optional:
             return get_neither_required_nor_nullable_type(type_hint, self.data_type.use_union_operator)
         return type_hint
 
@@ -294,12 +294,8 @@ def fall_back_to_nullable(self) -> bool:
         """Return whether to fall back to nullable type instead of UnsetType."""
         return not self._not_required
 
-    @property
-    def annotated(self) -> str | None:
-        """Get Annotated type hint with Meta constraints."""
-        if not self.use_annotated:  # pragma: no cover
-            return None
-
+    def _get_meta_string(self) -> str | None:
+        """Compute Meta(...) string if there are any meta constraints."""
         data: dict[str, Any] = {k: v for k, v in self.extras.items() if k in self._META_FIELD_KEYS}
         has_type_constraints = self.data_type.kwargs is not None and len(self.data_type.kwargs) > 0
         if (
@@ -317,21 +313,50 @@ def annotated(self) -> str | None:
             }
 
         meta_arguments = sorted(f"{k}={v!r}" for k, v in data.items() if v is not None)
-        if not meta_arguments:
+        return f"Meta({', '.join(meta_arguments)})" if meta_arguments else None
+
+    @property
+    def annotated(self) -> str | None:
+        """Get Annotated type hint with Meta constraints."""
+        if not self.use_annotated:  # pragma: no cover
             return None
 
-        meta = f"Meta({', '.join(meta_arguments)})"
+        meta = self._get_meta_string()
+
+        if not meta and not self.extras.get("is_classvar"):
+            return None
 
         if not self.required and not self.extras.get("is_classvar"):
             type_hint = self.data_type.type_hint
             annotated_type = f"Annotated[{type_hint}, {meta}]"
             return get_neither_required_nor_nullable_type(annotated_type, self.data_type.use_union_operator)
 
-        annotated_type = f"Annotated[{self.type_hint}, {meta}]"
+        # Handle ClassVar case (for discriminator fields in msgspec)
         if self.extras.get("is_classvar"):
-            annotated_type = f"ClassVar[{annotated_type}]"
+            if meta:
+                annotated_type = f"Annotated[{self.type_hint}, {meta}]"
+                return f"ClassVar[{annotated_type}]"
+            return f"ClassVar[{self.type_hint}]"
+
+        return f"Annotated[{self.type_hint}, {meta}]"
 
-        return annotated_type
+    @property
+    def needs_annotated_import(self) -> bool:
+        """Check if this field requires the Annotated import.
+
+        ClassVar fields without Meta constraints don't need Annotated.
+        """
+        if not self.annotated:
+            return False
+        # ClassVar without Meta doesn't use Annotated
+        if self.extras.get("is_classvar"):
+            return self._get_meta_string() is not None
+        return True
+
+    @property
+    def needs_meta_import(self) -> bool:
+        """Check if this field requires the Meta import."""
+        return self._get_meta_string() is not None
 
     def _get_default_as_struct_model(self) -> str | None:
         """Convert default value to Struct model using msgspec convert."""

src/datamodel_code_generator/parser/base.py

@@ -991,7 +991,7 @@ def _create_discriminator_data_type(
             data_type = self.data_type(literals=type_names)
         return data_type
 
-    def __apply_discriminator_type(  # noqa: PLR0912, PLR0915
+    def __apply_discriminator_type(  # noqa: PLR0912, PLR0914, PLR0915
         self,
         models: list[DataModel],
         imports: Imports,
@@ -1097,7 +1097,15 @@ def check_paths(
                         if field_name not in {discriminator_field.original_name, discriminator_field.name}:
                             continue
                         literals = discriminator_field.data_type.literals
-                        if len(literals) == 1 and literals[0] == (type_names[0] if type_names else None):
+                        const_value = discriminator_field.extras.get("const")
+                        expected_value = type_names[0] if type_names else None
+
+                        # Check if literals match (existing behavior)
+                        literals_match = len(literals) == 1 and literals[0] == expected_value
+                        # Check if const value matches (for msgspec with type: string + const)
+                        const_match = const_value is not None and const_value == expected_value
+
+                        if literals_match:
                             has_one_literal = True
                             if isinstance(discriminator_model, msgspec_model.Struct):  # pragma: no cover
                                 discriminator_model.add_base_class_kwarg("tag_field", f"'{field_name}'")
@@ -1106,6 +1114,14 @@ def check_paths(
                             # Found the discriminator field, no need to keep looking
                             break
 
+                        # For msgspec with const value but no literal (type: string + const case)
+                        if const_match and isinstance(discriminator_model, msgspec_model.Struct):  # pragma: no cover
+                            has_one_literal = True
+                            discriminator_model.add_base_class_kwarg("tag_field", f"'{field_name}'")
+                            discriminator_model.add_base_class_kwarg("tag", repr(const_value))
+                            discriminator_field.extras["is_classvar"] = True
+                            break
+
                         enum_source: Enum | None = None
                         if self.use_enum_values_in_discriminator:
                             enum_source = (  # pragma: no cover

src/datamodel_code_generator/parser/jsonschema.py

@@ -392,8 +392,15 @@ def ref_type(self) -> JSONReference | None:
 
     @cached_property
     def type_has_null(self) -> bool:
-        """Check if the type list contains null."""
-        return isinstance(self.type, list) and "null" in self.type
+        """Check if the type list or oneOf/anyOf contains null."""
+        if isinstance(self.type, list) and "null" in self.type:
+            return True
+        for item in self.oneOf + self.anyOf:
+            if item.type == "null":
+                return True
+            if isinstance(item.type, list) and "null" in item.type:
+                return True
+        return False
 
     @cached_property
     def has_multiple_types(self) -> bool:

tests/data/expected/main/jsonschema/discriminator_with_meta_msgspec.py

@@ -0,0 +1,23 @@
+# generated by datamodel-codegen:
+#   filename:  discriminator_with_meta_msgspec.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import Annotated, ClassVar, Literal, Union
+
+from msgspec import UNSET, Meta, Struct, UnsetType
+
+
+class SystemMessage(Struct, tag_field='role', tag='system'):
+    role: ClassVar[Annotated[Literal['system'], Meta(title='Message Role')]]
+    content: str
+
+
+class UserMessage(Struct, tag_field='role', tag='user'):
+    role: ClassVar[Annotated[Literal['user'], Meta(title='Message Role')]]
+    content: str
+
+
+class Model(Struct):
+    message: Union[SystemMessage, UserMessage, UnsetType] = UNSET

tests/data/expected/main/jsonschema/discriminator_with_type_string_msgspec.py

@@ -0,0 +1,23 @@
+# generated by datamodel-codegen:
+#   filename:  discriminator_with_type_string.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import ClassVar, Literal, Union
+
+from msgspec import UNSET, Struct, UnsetType
+
+
+class SystemMessage(Struct, tag_field='role', tag='system'):
+    role: ClassVar[Literal['system']]
+    content: str
+
+
+class UserMessage(Struct, tag_field='role', tag='user'):
+    role: ClassVar[Literal['user']]
+    content: str
+
+
+class Model(Struct):
+    message: Union[SystemMessage, UserMessage, UnsetType] = UNSET

tests/data/jsonschema/discriminator_with_meta_msgspec.json

@@ -0,0 +1,50 @@
+{
+  "$defs": {
+    "SystemMessage": {
+      "type": "object",
+      "properties": {
+        "role": {
+          "type": "string",
+          "const": "system",
+          "default": "system",
+          "title": "Message Role"
+        },
+        "content": {
+          "type": "string"
+        }
+      },
+      "required": ["role", "content"]
+    },
+    "UserMessage": {
+      "type": "object",
+      "properties": {
+        "role": {
+          "type": "string",
+          "const": "user",
+          "default": "user",
+          "title": "Message Role"
+        },
+        "content": {
+          "type": "string"
+        }
+      },
+      "required": ["role", "content"]
+    }
+  },
+  "type": "object",
+  "properties": {
+    "message": {
+      "discriminator": {
+        "propertyName": "role",
+        "mapping": {
+          "system": "#/$defs/SystemMessage",
+          "user": "#/$defs/UserMessage"
+        }
+      },
+      "oneOf": [
+        {"$ref": "#/$defs/SystemMessage"},
+        {"$ref": "#/$defs/UserMessage"}
+      ]
+    }
+  }
+}

tests/data/jsonschema/discriminator_with_type_string.json

@@ -0,0 +1,48 @@
+{
+  "$defs": {
+    "SystemMessage": {
+      "type": "object",
+      "properties": {
+        "role": {
+          "type": "string",
+          "const": "system",
+          "default": "system"
+        },
+        "content": {
+          "type": "string"
+        }
+      },
+      "required": ["role", "content"]
+    },
+    "UserMessage": {
+      "type": "object",
+      "properties": {
+        "role": {
+          "type": "string",
+          "const": "user",
+          "default": "user"
+        },
+        "content": {
+          "type": "string"
+        }
+      },
+      "required": ["role", "content"]
+    }
+  },
+  "type": "object",
+  "properties": {
+    "message": {
+      "discriminator": {
+        "propertyName": "role",
+        "mapping": {
+          "system": "#/$defs/SystemMessage",
+          "user": "#/$defs/UserMessage"
+        }
+      },
+      "oneOf": [
+        {"$ref": "#/$defs/SystemMessage"},
+        {"$ref": "#/$defs/UserMessage"}
+      ]
+    }
+  }
+}

tests/main/jsonschema/test_main_jsonschema.py

@@ -2485,6 +2485,42 @@ def test_main_jsonschema_openapi_keyword_only_msgspec_with_extra_data(tmp_path:
     assert_file_content(output_file, "discriminator_literals_msgspec_keyword_only_omit_defaults.py")
 
 
+@MSGSPEC_LEGACY_BLACK_SKIP
+def test_main_msgspec_discriminator_with_type_string(output_file: Path) -> None:
+    """Test msgspec Struct generation with discriminator using type: string + const."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "discriminator_with_type_string.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        assert_func=assert_file_content,
+        expected_file="discriminator_with_type_string_msgspec.py",
+        extra_args=[
+            "--output-model-type",
+            "msgspec.Struct",
+            "--target-python-version",
+            "3.10",
+        ],
+    )
+
+
+@MSGSPEC_LEGACY_BLACK_SKIP
+def test_main_msgspec_discriminator_with_meta(output_file: Path) -> None:
+    """Test msgspec Struct generation with discriminator ClassVar having Meta constraints."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "discriminator_with_meta_msgspec.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        assert_func=assert_file_content,
+        expected_file="discriminator_with_meta_msgspec.py",
+        extra_args=[
+            "--output-model-type",
+            "msgspec.Struct",
+            "--target-python-version",
+            "3.10",
+        ],
+    )
+
+
 @MSGSPEC_LEGACY_BLACK_SKIP
 def test_main_msgspec_null_field(output_file: Path) -> None:
     """Test msgspec Struct generation with null type fields."""