fix #3034 and #3035 by introducing ValidatedDefault (#3040)

* new fixtures

* one more fixture

* fix

* fix: use resolved.data_types for union branch filtering in _needs_validated_default

* more fixtures

* another fixture

* use model-level capability, and defer set

* fix overrides

* another fix

* one more

* test: stub my_app for py312 exec validation

* test: cover validated default parser helpers

* test: cover remaining validated default branches

---------

Co-authored-by: Koudai Aono <koxudaxi@gmail.com>

Author: keyz

src/datamodel_code_generator/model/_types.py

@@ -16,3 +16,15 @@ class WrappedDefault:
     def __repr__(self) -> str:
         """Return type constructor representation, e.g., 'CountType(10)'."""
         return f"{self.type_name}({self.value!r})"
+
+
+@dataclass(repr=False)
+class ValidatedDefault:
+    """Represents a structured default validated through the field type."""
+
+    value: Any
+    type_name: str
+
+    def __repr__(self) -> str:
+        """Return TypeAdapter validation representation for structured defaults."""
+        return f"TypeAdapter({self.type_name}).validate_python({self.value!r})"

src/datamodel_code_generator/model/base.py

@@ -25,7 +25,7 @@
     IMPORT_UNION,
     Import,
 )
-from datamodel_code_generator.model._types import WrappedDefault
+from datamodel_code_generator.model._types import ValidatedDefault, WrappedDefault
 from datamodel_code_generator.reference import Reference, _BaseModel
 from datamodel_code_generator.types import (
     ANY,
@@ -38,7 +38,7 @@
     get_optional_type,
 )
 
-__all__ = ["WrappedDefault"]
+__all__ = ["ValidatedDefault", "WrappedDefault"]
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -599,6 +599,7 @@ class DataModel(TemplateBase, Nullable, ABC):  # noqa: PLR0904
     SUPPORTS_DISCRIMINATOR: ClassVar[bool] = False
     SUPPORTS_FIELD_RENAMING: ClassVar[bool] = False
     SUPPORTS_WRAPPED_DEFAULT: ClassVar[bool] = False
+    SUPPORTS_VALIDATED_DEFAULT: ClassVar[bool] = False
     SUPPORTS_KW_ONLY: ClassVar[bool] = False
     has_forward_reference: bool = False
 

src/datamodel_code_generator/model/pydantic_base.py

@@ -19,14 +19,15 @@
     DataModel,
     DataModelFieldBase,
 )
-from datamodel_code_generator.model._types import WrappedDefault
+from datamodel_code_generator.model._types import ValidatedDefault, WrappedDefault
 from datamodel_code_generator.model.base import UNDEFINED, repr_set_sorted
 from datamodel_code_generator.types import STANDARD_LIST, UnionIntFloat, chain_as_tuple
 
 # Defined here instead of importing from pydantic_v2.imports to avoid circular import
 # (pydantic_base -> pydantic_v2.imports -> pydantic_v2/__init__ -> pydantic_v2.base_model -> pydantic_base)
 IMPORT_ANYURL = Import.from_full_path("pydantic.AnyUrl")
 IMPORT_FIELD = Import.from_full_path("pydantic.Field")
+IMPORT_TYPE_ADAPTER = Import.from_full_path("pydantic.TypeAdapter")
 
 if TYPE_CHECKING:
     from collections import defaultdict
@@ -126,7 +127,7 @@ def _get_strict_field_constraint_value(self, constraint: str, value: Any) -> Any
         return int(value)
 
     def _get_default_as_pydantic_model(self) -> str | None:  # noqa: PLR0911, PLR0912
-        if isinstance(self.default, WrappedDefault):
+        if isinstance(self.default, (ValidatedDefault, WrappedDefault)):
             return f"lambda :{self.default!r}"
         if self.data_type.is_list and len(self.data_type.data_types) == 1:
             data_type_child = self.data_type.data_types[0]
@@ -296,8 +297,12 @@ def annotated(self) -> str | None:
     @property
     def imports(self) -> tuple[Import, ...]:
         """Get all required imports including Field if needed."""
-        if self.field:
-            return chain_as_tuple(super().imports, (IMPORT_FIELD,))
+        field = self.field
+        if field:
+            extra_imports = [IMPORT_FIELD]
+            if isinstance(self.default, ValidatedDefault):
+                extra_imports.append(IMPORT_TYPE_ADAPTER)
+            return chain_as_tuple(super().imports, extra_imports)
         return super().imports
 
 

src/datamodel_code_generator/model/pydantic_v2/base_model.py

@@ -223,6 +223,7 @@ class BaseModel(BaseModelBase):
     SUPPORTS_DISCRIMINATOR: ClassVar[bool] = True
     SUPPORTS_FIELD_RENAMING: ClassVar[bool] = True
     SUPPORTS_WRAPPED_DEFAULT: ClassVar[bool] = True
+    SUPPORTS_VALIDATED_DEFAULT: ClassVar[bool] = True
     # In Pydantic 2.11+, populate_by_name is deprecated in favor of validate_by_name + validate_by_alias
     # Default to V2 compatible (populate_by_name) unless target_pydantic_version is specified
     _CONFIG_ATTRIBUTES_V2: ClassVar[list[ConfigAttribute]] = [

src/datamodel_code_generator/parser/base.py

@@ -74,6 +74,7 @@
     ConstraintsBase,
     DataModel,
     DataModelFieldBase,
+    ValidatedDefault,
     WrappedDefault,
 )
 from datamodel_code_generator.model.enum import Enum, Member
@@ -82,7 +83,7 @@
 from datamodel_code_generator.parser._graph import stable_toposort
 from datamodel_code_generator.parser._scc import find_circular_sccs, strongly_connected_components
 from datamodel_code_generator.reference import ModelResolver, ModelType, Reference
-from datamodel_code_generator.types import ANY, DataType, DataTypeManager
+from datamodel_code_generator.types import ANY, NONE, DataType, DataTypeManager, _remove_none_from_union
 from datamodel_code_generator.util import camel_to_snake
 
 if TYPE_CHECKING:
@@ -399,6 +400,90 @@ def iter_models_field_data_types(
                 yield model, field, data_type
 
 
+def _unwrap_type_alias(data_type: DataType) -> DataType:
+    current = data_type
+    seen: set[int] = set()
+    while current.reference and isinstance(current.reference.source, TypeAliasBase):
+        source = current.reference.source
+        if id(source) in seen or not source.fields:
+            break
+        seen.add(id(source))
+        current = source.fields[0].data_type
+    return current
+
+
+def _supports_validated_default_model(source: object) -> bool:
+    return isinstance(source, DataModel) and source.SUPPORTS_VALIDATED_DEFAULT and not source.is_alias
+
+
+def _contains_model_reference(data_type: DataType) -> bool:
+    stack = [data_type]
+    seen: set[int] = set()
+
+    while stack:
+        resolved = _unwrap_type_alias(stack.pop())
+        resolved_id = id(resolved)
+        if resolved_id in seen:
+            continue
+        seen.add(resolved_id)
+
+        if resolved.reference and _supports_validated_default_model(resolved.reference.source):
+            return True
+
+        if resolved.dict_key:
+            stack.append(resolved.dict_key)
+        stack.extend(resolved.data_types)
+
+    return False
+
+
+def _uses_existing_model_factory_path(data_type: DataType) -> bool:
+    for candidate in data_type.data_types or (data_type,):
+        if candidate.reference and _supports_validated_default_model(candidate.reference.source):
+            return True
+        if (
+            candidate.is_list
+            and len(candidate.data_types) == 1
+            and candidate.data_types[0].reference
+            and _supports_validated_default_model(candidate.data_types[0].reference.source)
+        ):
+            return True
+    return False
+
+
+def _default_matches_plain_container_branch(default: Any, data_type: DataType) -> bool:
+    resolved = _unwrap_type_alias(data_type)
+    return (isinstance(default, dict) and resolved.is_dict and not _contains_model_reference(resolved)) or (
+        isinstance(default, list) and resolved.is_list and not _contains_model_reference(resolved)
+    )
+
+
+def _get_validated_default_type_name(data_type: DataType) -> str:
+    type_name = data_type.type_hint
+    if data_type.is_optional:
+        return _remove_none_from_union(type_name, use_union_operator=data_type.use_union_operator)
+    return type_name
+
+
+def _needs_validated_default(default: Any, data_type: DataType) -> bool:
+    if not isinstance(default, (dict, list)):
+        return False
+
+    resolved = _unwrap_type_alias(data_type)
+    if not _contains_model_reference(resolved):
+        return False
+
+    if resolved.is_union:
+        non_none_branches = tuple(branch for branch in resolved.data_types if branch.type_hint != NONE)
+        # This must use the declared field type, not the unwrapped branch: alias-backed
+        # unions like `type Alias = A | None` only get the correct factory via `Alias`.
+        if len(non_none_branches) == 1 and _uses_existing_model_factory_path(data_type):
+            return False
+        return not any(_default_matches_plain_container_branch(default, branch) for branch in resolved.data_types)
+
+    return not _uses_existing_model_factory_path(data_type)
+
+
 def _alias_base_class_imports(
     model: DataModel,
     aliased_imports: dict[tuple[str | None, str], Import],
@@ -2152,6 +2237,31 @@ def __wrap_root_model_default_values(
                     type_name=type_name,
                 )
 
+    def __wrap_validated_default_values(
+        self,
+        models: list[DataModel],
+    ) -> None:
+        """Wrap structured defaults that must be validated through the declared field type."""
+        if not self.data_model_type.SUPPORTS_VALIDATED_DEFAULT:
+            return
+        for model in models:
+            if isinstance(model, (Enum, self.data_model_root_type)):
+                continue
+            for model_field in model.fields:
+                if model_field.default is None:
+                    continue
+                if isinstance(model_field.default, (Member, WrappedDefault)):
+                    continue
+                if isinstance(model_field.default, ValidatedDefault):
+                    model_field.default.type_name = _get_validated_default_type_name(model_field.data_type)
+                    continue
+                if not _needs_validated_default(model_field.default, model_field.data_type):
+                    continue
+                model_field.default = ValidatedDefault(
+                    value=model_field.default,
+                    type_name=_get_validated_default_type_name(model_field.data_type),
+                )
+
     def __override_required_field(
         self,
         models: list[DataModel],
@@ -3187,6 +3297,7 @@ def _process_single_module(  # noqa: PLR0913, PLR0917
         models = self.__remove_overridden_models(models)
         self.__apply_type_overrides(models)
         self.__update_type_aliases(models)
+        self.__wrap_validated_default_values(models)
 
         return ModuleContext(module, module_, models, is_init, imports, scoped_model_resolver)
 
@@ -3219,6 +3330,7 @@ def _finalize_modules(
 
         for ctx in contexts:
             self.__change_imported_model_name(ctx.models, ctx.imports, ctx.scoped_model_resolver)
+            self.__wrap_validated_default_values(ctx.models)
 
     def _generate_module_output(  # noqa: PLR0913, PLR0917
         self,

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

@@ -0,0 +1,25 @@
+# generated by datamodel-codegen:
+#   filename:  type_alias_chain_model_default_object_ref.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field, TypeAdapter
+
+
+class A(BaseModel):
+    type: Literal['a'] = 'a'
+
+
+type Alias1 = A
+
+
+type Alias2 = Alias1
+
+
+class Model(BaseModel):
+    x: Alias2 | None = Field(
+        default_factory=lambda: TypeAdapter(Alias2).validate_python({'type': 'a'})
+    )

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

@@ -0,0 +1,32 @@
+# generated by datamodel-codegen:
+#   filename:  type_alias_chain_union_default_object_ref.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field, TypeAdapter
+
+
+class A(BaseModel):
+    type: Literal['a'] = 'a'
+
+
+class B(BaseModel):
+    type: Literal['b'] = 'b'
+
+
+type Alias1 = A | B
+
+
+type Alias2 = Alias1
+
+
+class Model(BaseModel):
+    x_value: Annotated[
+        Alias2 | None,
+        Field(
+            default_factory=lambda: TypeAdapter(Alias2).validate_python({'type': 'b'})
+        ),
+    ]

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

@@ -0,0 +1,32 @@
+# generated by datamodel-codegen:
+#   filename:  type_alias_dict_union_default_object_ref.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field, TypeAdapter
+
+
+class A(BaseModel):
+    type: Literal['a'] = 'a'
+
+
+class B(BaseModel):
+    type: Literal['b']
+    value: int | None = None
+
+
+type Alias = dict[str, A | B]
+
+
+class Model(BaseModel):
+    x_value: Annotated[
+        Alias | None,
+        Field(
+            default_factory=lambda: TypeAdapter(Alias).validate_python(
+                {'k': {'type': 'b', 'value': 1}}
+            )
+        ),
+    ]

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

@@ -0,0 +1,29 @@
+# generated by datamodel-codegen:
+#   filename:  type_alias_inline_dict_union_default_object.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field, TypeAdapter
+
+
+class A(BaseModel):
+    type: Literal['a'] = 'a'
+
+
+class B(BaseModel):
+    type: Literal['b']
+    value: int | None = None
+
+
+class Model(BaseModel):
+    x_value: Annotated[
+        dict[str, A | B] | None,
+        Field(
+            default_factory=lambda: TypeAdapter(dict[str, A | B]).validate_python(
+                {'k': {'type': 'b', 'value': 1}}
+            )
+        ),
+    ]

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

@@ -0,0 +1,29 @@
+# generated by datamodel-codegen:
+#   filename:  type_alias_inline_list_union_default_object.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field, TypeAdapter
+
+
+class A(BaseModel):
+    type: Literal['a'] = 'a'
+
+
+class B(BaseModel):
+    type: Literal['b']
+    value: int | None = None
+
+
+class Model(BaseModel):
+    x_value: Annotated[
+        list[A | B] | None,
+        Field(
+            default_factory=lambda: TypeAdapter(list[A | B]).validate_python(
+                [{'type': 'b', 'value': 1}]
+            )
+        ),
+    ]