Add --reuse-scope option for cross-file model deduplication (#2573)

* feat: add reuse scope for model deduplication across modules

* fix: update default value for reuse scope argument to None

* docs: update command help in README

🤖 Generated by GitHub Actions

* test: add tests for --reuse-scope=tree functionality in JSON schema processing

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* feat: add shared module name option for reuse scope tree functionality

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* docs: update command help in README

🤖 Generated by GitHub Actions

* feat: implement reuse scope tree functionality with model deduplication and shared module validation

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* refactor: streamline function definitions and improve code readability

* fix: update shared module name help text for --reuse-scope=tree

* fix: add error handling for duplicate and canonical models in reuse scope tree

* fix: update documentation for reuse scope tree to reflect shared.py changes

* feat: enhance reuse scope tree support for dataclasses and TypedDict

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: 3 people

README.md

@@ -478,6 +478,13 @@ Model customization:
                         Set name of models defined inline from the parent model
   --reuse-model         Reuse models on the field when a module has the model with the same
                         content
+  --reuse-scope {module,tree}
+                        Scope for model reuse deduplication: module (per-file, default) or
+                        tree (cross-file with shared module). Only effective when --reuse-
+                        model is set.
+  --shared-module-name SHARED_MODULE_NAME
+                        Name of the shared module for --reuse-scope=tree (default:
+                        "shared"). Use this option if your schema has a file named "shared".
   --skip-root-model     Skip generating the model for the root schema element
   --target-python-version {3.9,3.10,3.11,3.12,3.13,3.14}
                         target python version

docs/index.md

@@ -470,6 +470,13 @@ Model customization:
                         Set name of models defined inline from the parent model
   --reuse-model         Reuse models on the field when a module has the model with the same
                         content
+  --reuse-scope {module,tree}
+                        Scope for model reuse deduplication: module (per-file, default) or
+                        tree (cross-file with shared module). Only effective when --reuse-
+                        model is set.
+  --shared-module-name SHARED_MODULE_NAME
+                        Name of the shared module for --reuse-scope=tree (default:
+                        "shared"). Use this option if your schema has a file named "shared".
   --skip-root-model     Skip generating the model for the root schema element
   --target-python-version {3.9,3.10,3.11,3.12,3.13,3.14}
                         target python version

src/datamodel_code_generator/__init__.py

@@ -53,6 +53,7 @@
 
 MIN_VERSION: Final[int] = 9
 MAX_VERSION: Final[int] = 13
+DEFAULT_SHARED_MODULE_NAME: Final[str] = "shared"
 
 T = TypeVar("T")
 
@@ -224,6 +225,17 @@ class DataModelType(Enum):
     MsgspecStruct = "msgspec.Struct"
 
 
+class ReuseScope(Enum):
+    """Scope for model reuse deduplication.
+
+    module: Deduplicate identical models within each module (default).
+    tree: Deduplicate identical models across all modules, placing shared models in shared.py.
+    """
+
+    Module = "module"
+    Tree = "tree"
+
+
 class OpenAPIScope(Enum):
     """Scopes for OpenAPI model generation."""
 
@@ -306,6 +318,8 @@ def generate(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
     use_inline_field_description: bool = False,
     use_default_kwarg: bool = False,
     reuse_model: bool = False,
+    reuse_scope: ReuseScope = ReuseScope.Module,
+    shared_module_name: str = DEFAULT_SHARED_MODULE_NAME,
     encoding: str = "utf-8",
     enum_field_as_literal: LiteralType | None = None,
     use_one_literal_as_default: bool = False,
@@ -537,6 +551,8 @@ def get_header_and_first_line(csv_file: IO[str]) -> dict[str, Any]:
         use_inline_field_description=use_inline_field_description,
         use_default_kwarg=use_default_kwarg,
         reuse_model=reuse_model,
+        reuse_scope=reuse_scope,
+        shared_module_name=shared_module_name,
         enum_field_as_literal=LiteralType.All
         if output_model_type == DataModelType.TypingTypedDict
         else enum_field_as_literal,

src/datamodel_code_generator/__main__.py

@@ -19,12 +19,14 @@
 from typing_extensions import TypeAlias
 
 from datamodel_code_generator import (
+    DEFAULT_SHARED_MODULE_NAME,
     DataclassArguments,
     DataModelType,
     Error,
     InputFileType,
     InvalidClassNameError,
     OpenAPIScope,
+    ReuseScope,
     enable_debug_message,
     generate,
 )
@@ -357,6 +359,8 @@ def validate_root(cls, values: dict[str, Any]) -> dict[str, Any]:  # noqa: N805
     use_inline_field_description: bool = False
     use_default_kwarg: bool = False
     reuse_model: bool = False
+    reuse_scope: ReuseScope = ReuseScope.Module
+    shared_module_name: str = DEFAULT_SHARED_MODULE_NAME
     encoding: str = DEFAULT_ENCODING
     enum_field_as_literal: Optional[LiteralType] = None  # noqa: UP045
     use_one_literal_as_default: bool = False
@@ -535,6 +539,13 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
 
     if config.disable_warnings:
         warnings.simplefilter("ignore")
+
+    if config.reuse_scope == ReuseScope.Tree and not config.reuse_model:
+        print(  # noqa: T201
+            "Warning: --reuse-scope=tree has no effect without --reuse-model",
+            file=sys.stderr,
+        )
+
     extra_template_data: defaultdict[str, dict[str, Any]] | None
     if config.extra_template_data is None:
         extra_template_data = None
@@ -616,6 +627,8 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
             use_inline_field_description=config.use_inline_field_description,
             use_default_kwarg=config.use_default_kwarg,
             reuse_model=config.reuse_model,
+            reuse_scope=config.reuse_scope,
+            shared_module_name=config.shared_module_name,
             encoding=config.encoding,
             enum_field_as_literal=config.enum_field_as_literal,
             use_one_literal_as_default=config.use_one_literal_as_default,

src/datamodel_code_generator/arguments.py

@@ -14,7 +14,14 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
-from datamodel_code_generator import DataclassArguments, DataModelType, InputFileType, OpenAPIScope
+from datamodel_code_generator import (
+    DEFAULT_SHARED_MODULE_NAME,
+    DataclassArguments,
+    DataModelType,
+    InputFileType,
+    OpenAPIScope,
+    ReuseScope,
+)
 from datamodel_code_generator.format import DatetimeClassType, Formatter, PythonVersion
 from datamodel_code_generator.model.pydantic_v2 import UnionMode
 from datamodel_code_generator.parser import LiteralType
@@ -224,6 +231,19 @@ def start_section(self, heading: str | None) -> None:
     action="store_true",
     default=None,
 )
+model_options.add_argument(
+    "--reuse-scope",
+    help="Scope for model reuse deduplication: module (per-file, default) or tree (cross-file with shared module). "
+    "Only effective when --reuse-model is set.",
+    choices=[s.value for s in ReuseScope],
+    default=None,
+)
+model_options.add_argument(
+    "--shared-module-name",
+    help=f'Name of the shared module for --reuse-scope=tree (default: "{DEFAULT_SHARED_MODULE_NAME}"). '
+    f'Use this option if your schema has a file named "{DEFAULT_SHARED_MODULE_NAME}".',
+    default=None,
+)
 model_options.add_argument(
     "--target-python-version",
     help="target python version",

src/datamodel_code_generator/parser/base.py

@@ -20,6 +20,7 @@
 
 from pydantic import BaseModel
 
+from datamodel_code_generator import DEFAULT_SHARED_MODULE_NAME, Error, ReuseScope
 from datamodel_code_generator.format import (
     DEFAULT_FORMATTERS,
     CodeFormatter,
@@ -405,6 +406,8 @@ def __init__(  # noqa: PLR0913, PLR0915
         use_inline_field_description: bool = False,
         use_default_kwarg: bool = False,
         reuse_model: bool = False,
+        reuse_scope: ReuseScope | None = None,
+        shared_module_name: str = DEFAULT_SHARED_MODULE_NAME,
         encoding: str = "utf-8",
         enum_field_as_literal: LiteralType | None = None,
         set_default_enum_member: bool = False,
@@ -495,6 +498,8 @@ def __init__(  # noqa: PLR0913, PLR0915
         self.use_inline_field_description: bool = use_inline_field_description
         self.use_default_kwarg: bool = use_default_kwarg
         self.reuse_model: bool = reuse_model
+        self.reuse_scope: ReuseScope | None = reuse_scope
+        self.shared_module_name: str = shared_module_name
         self.encoding: str = encoding
         self.enum_field_as_literal: LiteralType | None = enum_field_as_literal
         self.set_default_enum_member: bool = set_default_enum_member
@@ -1002,7 +1007,7 @@ def __set_reference_default_value_to_field(cls, models: list[DataModel]) -> None
                     model_field.default = model_field.data_type.reference.source.default
 
     def __reuse_model(self, models: list[DataModel], require_update_action_models: list[str]) -> None:
-        if not self.reuse_model:
+        if not self.reuse_model or self.reuse_scope == ReuseScope.Tree:
             return
         model_cache: dict[tuple[HashableComparable, ...], Reference] = {}
         duplicates = []
@@ -1041,6 +1046,133 @@ def __reuse_model(self, models: list[DataModel], require_update_action_models: l
         for duplicate in duplicates:
             models.remove(duplicate)
 
+    def __find_duplicate_models_across_modules(  # noqa: PLR6301
+        self,
+        module_models: list[tuple[tuple[str, ...], list[DataModel]]],
+    ) -> list[tuple[tuple[str, ...], DataModel, tuple[str, ...], DataModel]]:
+        """Find duplicate models across all modules by comparing render output and imports."""
+        all_models: list[tuple[tuple[str, ...], DataModel]] = []
+        for module, models in module_models:
+            all_models.extend((module, model) for model in models)
+
+        model_cache: dict[tuple[HashableComparable, ...], tuple[tuple[str, ...], DataModel]] = {}
+        duplicates: list[tuple[tuple[str, ...], DataModel, tuple[str, ...], DataModel]] = []
+
+        for module, model in all_models:
+            model_key = tuple(to_hashable(v) for v in (model.render(class_name="M"), model.imports))
+            cached = model_cache.get(model_key)
+            if cached:
+                canonical_module, canonical_model = cached
+                duplicates.append((module, model, canonical_module, canonical_model))
+            else:
+                model_cache[model_key] = (module, model)
+
+        return duplicates
+
+    def __validate_shared_module_name(
+        self,
+        module_models: list[tuple[tuple[str, ...], list[DataModel]]],
+    ) -> None:
+        """Validate that the shared module name doesn't conflict with existing modules."""
+        shared_module = self.shared_module_name
+        existing_module_names = {module[0] for module, _ in module_models}
+        if shared_module in existing_module_names:
+            msg = (
+                f"Schema file or directory '{shared_module}' conflicts with the shared module name. "
+                f"Use --shared-module-name to specify a different name."
+            )
+            raise Error(msg)
+
+    def __create_shared_module_from_duplicates(  # noqa: PLR0912
+        self,
+        module_models: list[tuple[tuple[str, ...], list[DataModel]]],
+        duplicates: list[tuple[tuple[str, ...], DataModel, tuple[str, ...], DataModel]],
+        require_update_action_models: list[str],
+    ) -> tuple[tuple[str, ...], list[DataModel]]:
+        """Create shared module with canonical models and replace duplicates with inherited models."""
+        shared_module = self.shared_module_name
+
+        shared_models: list[DataModel] = []
+        canonical_to_shared_ref: dict[DataModel, Reference] = {}
+        canonical_models_seen: set[DataModel] = set()
+
+        # Process in order of first appearance in duplicates to ensure stable ordering
+        for _, _, _, canonical in duplicates:
+            if canonical in canonical_models_seen:
+                continue
+            canonical_models_seen.add(canonical)
+            canonical.file_path = Path(f"{shared_module}.py")
+            canonical_to_shared_ref[canonical] = canonical.reference
+            shared_models.append(canonical)
+
+        supports_inheritance = issubclass(
+            self.data_model_type,
+            (
+                pydantic_model.BaseModel,
+                pydantic_model_v2.BaseModel,
+                dataclass_model.DataClass,
+            ),
+        )
+
+        for duplicate_module, duplicate_model, _, canonical_model in duplicates:
+            shared_ref = canonical_to_shared_ref[canonical_model]
+            for module, models in module_models:
+                if module != duplicate_module or duplicate_model not in models:
+                    continue
+                if isinstance(duplicate_model, Enum) or not supports_inheritance:
+                    for child in duplicate_model.reference.children[:]:
+                        data_model = get_most_of_parent(child)
+                        if data_model in models and isinstance(child, DataType):
+                            child.replace_reference(shared_ref)
+                    models.remove(duplicate_model)
+                else:
+                    index = models.index(duplicate_model)
+                    inherited_model = duplicate_model.__class__(
+                        fields=[],
+                        base_classes=[shared_ref],
+                        description=duplicate_model.description,
+                        reference=Reference(
+                            name=duplicate_model.name,
+                            path=duplicate_model.reference.path + "/reuse",
+                        ),
+                        custom_template_dir=duplicate_model._custom_template_dir,  # noqa: SLF001
+                    )
+                    if shared_ref.path in require_update_action_models:
+                        add_model_path_to_list(require_update_action_models, inherited_model)
+                    models.insert(index, inherited_model)
+                    models.remove(duplicate_model)
+                break
+            else:  # pragma: no cover
+                msg = f"Duplicate model {duplicate_model.name} not found in module {duplicate_module}"
+                raise RuntimeError(msg)
+
+        for canonical in canonical_models_seen:
+            for _module, models in module_models:
+                if canonical in models:
+                    models.remove(canonical)
+                    break
+            else:  # pragma: no cover
+                msg = f"Canonical model {canonical.name} not found in any module"
+                raise RuntimeError(msg)
+
+        return (shared_module,), shared_models
+
+    def __reuse_model_tree_scope(
+        self,
+        module_models: list[tuple[tuple[str, ...], list[DataModel]]],
+        require_update_action_models: list[str],
+    ) -> tuple[tuple[str, ...], list[DataModel]] | None:
+        """Deduplicate models across all modules, placing shared models in shared.py."""
+        if not self.reuse_model or self.reuse_scope != ReuseScope.Tree:
+            return None
+
+        duplicates = self.__find_duplicate_models_across_modules(module_models)
+        if not duplicates:
+            return None
+
+        self.__validate_shared_module_name(module_models)
+        return self.__create_shared_module_from_duplicates(module_models, duplicates, require_update_action_models)
+
     def __collapse_root_models(  # noqa: PLR0912
         self,
         models: list[DataModel],
@@ -1499,6 +1631,10 @@ def sort_key(data_model: DataModel) -> tuple[int, tuple[str, ...]]:
             ))
             previous_module = module
 
+        shared_module_entry = self.__reuse_model_tree_scope(module_models, require_update_action_models)
+        if shared_module_entry:
+            module_models.insert(0, shared_module_entry)
+
         class Processed(NamedTuple):
             module: tuple[str, ...]
             models: list[DataModel]

src/datamodel_code_generator/parser/graphql.py

@@ -15,11 +15,13 @@
 from urllib.parse import ParseResult
 
 from datamodel_code_generator import (
+    DEFAULT_SHARED_MODULE_NAME,
     DataclassArguments,
     DefaultPutDict,
     LiteralType,
     PythonVersion,
     PythonVersionMin,
+    ReuseScope,
     snooper_to_methods,
 )
 from datamodel_code_generator.format import DEFAULT_FORMATTERS, DatetimeClassType, Formatter
@@ -126,6 +128,8 @@ def __init__(  # noqa: PLR0913
         use_inline_field_description: bool = False,
         use_default_kwarg: bool = False,
         reuse_model: bool = False,
+        reuse_scope: ReuseScope | None = None,
+        shared_module_name: str = DEFAULT_SHARED_MODULE_NAME,
         encoding: str = "utf-8",
         enum_field_as_literal: LiteralType | None = None,
         set_default_enum_member: bool = False,
@@ -211,6 +215,8 @@ def __init__(  # noqa: PLR0913
             use_inline_field_description=use_inline_field_description,
             use_default_kwarg=use_default_kwarg,
             reuse_model=reuse_model,
+            reuse_scope=reuse_scope,
+            shared_module_name=shared_module_name,
             encoding=encoding,
             enum_field_as_literal=enum_field_as_literal,
             use_one_literal_as_default=use_one_literal_as_default,

src/datamodel_code_generator/parser/jsonschema.py

@@ -20,8 +20,10 @@
 )
 
 from datamodel_code_generator import (
+    DEFAULT_SHARED_MODULE_NAME,
     DataclassArguments,
     InvalidClassNameError,
+    ReuseScope,
     YamlValue,
     load_yaml,
     load_yaml_dict,
@@ -478,6 +480,8 @@ def __init__(  # noqa: PLR0913
         use_inline_field_description: bool = False,
         use_default_kwarg: bool = False,
         reuse_model: bool = False,
+        reuse_scope: ReuseScope | None = None,
+        shared_module_name: str = DEFAULT_SHARED_MODULE_NAME,
         encoding: str = "utf-8",
         enum_field_as_literal: LiteralType | None = None,
         use_one_literal_as_default: bool = False,
@@ -564,6 +568,8 @@ def __init__(  # noqa: PLR0913
             use_inline_field_description=use_inline_field_description,
             use_default_kwarg=use_default_kwarg,
             reuse_model=reuse_model,
+            reuse_scope=reuse_scope,
+            shared_module_name=shared_module_name,
             encoding=encoding,
             enum_field_as_literal=enum_field_as_literal,
             use_one_literal_as_default=use_one_literal_as_default,