Fix generation reliability edge cases (#3561)

* Fix generation reliability edge cases

* Fix command header redaction test compatibility

* Address reliability review feedback

* Avoid path monkeypatch leakage

* Fix RootModel config rendering

* Sync RootModel config before rendering

* Stabilize RootModel config rendering

* Harden RootModel config sync

* Preserve RootModel config object

* Tighten Ruff stdout fallback

Author: koxudaxi

src/datamodel_code_generator/__init__.py

@@ -106,6 +106,8 @@
 Returned by generate() when output=None and multiple modules are generated.
 """
 
+_SCHEMA_TEXT_START_CHARS = ("{", "[", "<")
+
 
 def _apply_generate_config_preset(config: GenerateConfig) -> GenerateConfig:
     """Return a generate config with preset defaults applied."""
@@ -759,6 +761,23 @@ def _generate_config_values(generate_config: GenerateConfig) -> dict[str, Any]:
     return values
 
 
+def _coerce_existing_path_input(
+    input_: Path | str | ParseResult | Mapping[str, Any] | list[Any],
+) -> Path | str | ParseResult | Mapping[str, Any] | list[Any]:
+    if not isinstance(input_, str):
+        return input_
+    if not input_ or "\n" in input_ or "\r" in input_:
+        return input_
+    if (stripped := input_.lstrip()).startswith(_SCHEMA_TEXT_START_CHARS) or "://" in stripped:
+        return input_
+    try:
+        path = Path(input_).expanduser()
+        path_exists = path.exists()
+    except (OSError, RuntimeError, ValueError):  # pragma: no cover
+        return input_
+    return path if path_exists else input_
+
+
 def _create_parser_config(
     generate_config: GenerateConfig,
     additional_options: ParserConfigDict
@@ -1148,7 +1167,7 @@ def _build_parser(  # noqa: PLR0911, PLR0913
     raise Error(msg)
 
 
-def _emit_results(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
+def _emit_results(  # noqa: PLR0912, PLR0913, PLR0915
     results: str | dict[tuple[str, ...], Any],
     input_: Path | str | ParseResult | Mapping[str, Any] | list[Any],
     input_filename: str | None,
@@ -1225,25 +1244,24 @@ def _emit_results(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
             for name, result in sorted(results.items())
         }
 
-    file: IO[Any] | None
     for path, (body, future_imports, filename) in modules.items():
         if not path.parent.exists():
             path.parent.mkdir(parents=True)
 
         safe_filename = filename.replace("\n", " ").replace("\r", " ") if filename else ""
         effective_header = custom_file_header or header.format(safe_filename)
-        file = path.open("wt", encoding=config.encoding)
-        if custom_file_header and body:
-            file.write(
-                _build_module_content(body, effective_header, custom_file_header, future_imports=future_imports) + "\n"
-            )
-        else:
-            file.write(effective_header)
-            if body:
-                file.write("\n\n")
-                file.write(body.rstrip())
-            file.write("\n")
-        file.close()
+        with path.open("wt", encoding=config.encoding) as file:
+            if custom_file_header and body:
+                file.write(
+                    _build_module_content(body, effective_header, custom_file_header, future_imports=future_imports)
+                    + "\n"
+                )
+            else:
+                file.write(effective_header)
+                if body:
+                    file.write("\n\n")
+                    file.write(body.rstrip())
+                file.write("\n")
 
     if defer_formatting and config.formatters:
         from datamodel_code_generator._format_types import Formatter  # noqa: PLC0415
@@ -1343,6 +1361,8 @@ def generate(  # noqa: PLR0912, PLR0914, PLR0915
     skip_root_model = config.skip_root_model
     source_override: Mapping[str, Any] | None = None
 
+    input_ = _coerce_existing_path_input(input_)
+
     if (
         isinstance(input_, list)
         and input_file_type != InputFileType.MCPTools

src/datamodel_code_generator/__main__.py

@@ -188,6 +188,8 @@
 })
 
 ORIGINAL_FIELD_NAME_DELIMITER_ERROR = "`--original-field-name-delimiter` can not be used without `--snake-case-field`."
+SENSITIVE_COMMAND_OPTIONS: frozenset[str] = frozenset({"--http-headers", "--http-query-parameters"})
+REDACTED_COMMAND_ARGUMENT = "<redacted>"
 
 
 class Exit(IntEnum):
@@ -204,6 +206,31 @@ def sig_int_handler(_: int, __: Any) -> None:  # pragma: no cover
     sys.exit(Exit.OK)
 
 
+def _redact_command_args(args: Sequence[str]) -> list[str]:
+    redacted: list[str] = []
+    redact_values = False
+    for arg in args:
+        match arg:
+            case option if option in SENSITIVE_COMMAND_OPTIONS:
+                redacted.append(option)
+                redact_values = True
+            case option_value if (option := option_value.partition("="))[1] and option[0] in SENSITIVE_COMMAND_OPTIONS:
+                redacted.append(f"{option[0]}={REDACTED_COMMAND_ARGUMENT}")
+                redact_values = False
+            case option if redact_values and option.startswith("-"):
+                redacted.append(option)
+                redact_values = False
+            case _ if redact_values:
+                redacted.append(REDACTED_COMMAND_ARGUMENT)
+            case _:
+                redacted.append(arg)
+    return redacted
+
+
+def _command_header(args: Sequence[str]) -> str:
+    return shlex.join(["datamodel-codegen", *_redact_command_args(args)])
+
+
 signal.signal(signal.SIGINT, sig_int_handler)
 
 
@@ -1393,6 +1420,11 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
         generate_output = config.output
         is_directory_output = False
 
+    def cleanup_and_return(exit_code: Exit) -> Exit:
+        if temp_context is not None:
+            temp_context.cleanup()
+        return exit_code
+
     try:
         input_: Path | str | ParseResult
         if config.input_model:
@@ -1421,29 +1453,23 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
             extra_template_data=extra_template_data,
             aliases=aliases,
             serialization_aliases=serialization_aliases,
-            command_line=shlex.join(["datamodel-codegen", *args]) if config.enable_command_header else None,
+            command_line=_command_header(args) if config.enable_command_header else None,
             custom_formatters_kwargs=custom_formatters_kwargs,
             settings_path=config.output,
             validators=validators_config,
             default_value_overrides=default_value_overrides,
         )
     except InvalidClassNameError as e:
         print(f"{e} You have to set `--class-name` option", file=sys.stderr)  # noqa: T201
-        if temp_context is not None:
-            temp_context.cleanup()
-        return Exit.ERROR
+        return cleanup_and_return(Exit.ERROR)
     except Error as e:
         print(str(e), file=sys.stderr)  # noqa: T201
-        if temp_context is not None:
-            temp_context.cleanup()
-        return Exit.ERROR
+        return cleanup_and_return(Exit.ERROR)
     except Exception:  # noqa: BLE001
         import traceback  # noqa: PLC0415
 
         print(traceback.format_exc(), file=sys.stderr)  # noqa: T201
-        if temp_context is not None:
-            temp_context.cleanup()
-        return Exit.ERROR
+        return cleanup_and_return(Exit.ERROR)
 
     if writes_json_output_file and generate_output is not None and config.output is not None:
         _copy_generated_output(generate_output, config.output, is_directory_output=is_directory_output)
@@ -1459,10 +1485,6 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
             )
             + "\n"
         )
-        if temp_context is not None:  # pragma: no branch
-            temp_context.cleanup()
-            temp_context = None
-
     if config.check and config.output is not None and generate_output is not None:
         from datamodel_code_generator._structured_output import CheckDifferencePayload  # noqa: PLC0415
 
@@ -1536,9 +1558,6 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
                     print("".join(single_file_diff_lines), end="")  # noqa: T201
                 has_differences = True
 
-        if temp_context is not None:  # pragma: no branch
-            temp_context.cleanup()
-
         if namespace.output_format == "json":
             content = "".join(single_file_diff_lines)
             content += "".join("".join(changed_file.diff_lines) for changed_file in changed_files)
@@ -1553,25 +1572,27 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
                 + "\n"
             )
 
-        return Exit.DIFF if has_differences else Exit.OK
+        return cleanup_and_return(Exit.DIFF if has_differences else Exit.OK)
 
     if config.watch:
         try:
             from datamodel_code_generator.watch import watch_and_regenerate  # noqa: PLC0415
 
-            return watch_and_regenerate(
-                config,
-                extra_template_data,
-                aliases,
-                serialization_aliases,
-                custom_formatters_kwargs,
-                default_value_overrides,
+            return cleanup_and_return(
+                watch_and_regenerate(
+                    config,
+                    extra_template_data,
+                    aliases,
+                    serialization_aliases,
+                    custom_formatters_kwargs,
+                    default_value_overrides,
+                )
             )
         except Exception as e:  # noqa: BLE001
             print(str(e), file=sys.stderr)  # noqa: T201
-            return Exit.ERROR
+            return cleanup_and_return(Exit.ERROR)
 
-    return Exit.OK
+    return cleanup_and_return(Exit.OK)
 
 
 if __name__ == "__main__":

src/datamodel_code_generator/format.py

@@ -504,45 +504,63 @@ def apply_builtin_formatter(  # noqa: PLR0913
 
     def apply_ruff_lint(self, code: str) -> str:
         """Run ruff check with auto-fix on code."""
-        result = subprocess.run(  # noqa: S603
+        result = self._run_ruff_command(
             self._ruff_check_command("-"),
-            input=code.encode(self.encoding),
-            capture_output=True,
-            check=False,
-            cwd=self.settings_path,
+            stdin=code.encode(self.encoding),
+            allow_stdout_on_error=True,
         )
         return result.stdout.decode(self.encoding)
 
     def apply_ruff_formatter(self, code: str) -> str:
         """Format code using ruff format."""
         ruff_path = self._find_ruff_path()
-        result = subprocess.run(  # noqa: S603
+        result = self._run_ruff_command(
             (ruff_path, "format", "-"),
-            input=code.encode(self.encoding),
-            capture_output=True,
-            check=False,
-            cwd=self.settings_path,
+            stdin=code.encode(self.encoding),
         )
         return result.stdout.decode(self.encoding)
 
     def apply_ruff_check_and_format(self, code: str) -> str:
         """Run ruff check and format sequentially for reliable processing."""
         ruff_path = self._find_ruff_path()
-        check_result = subprocess.run(  # noqa: S603
+        check_result = self._run_ruff_command(
             self._ruff_check_command("-", ruff_path=ruff_path),
-            input=code.encode(self.encoding),
-            capture_output=True,
-            check=False,
-            cwd=self.settings_path,
+            stdin=code.encode(self.encoding),
+            allow_stdout_on_error=True,
         )
-        format_result = subprocess.run(  # noqa: S603
+        format_result = self._run_ruff_command(
             (ruff_path, "format", "-"),
-            input=check_result.stdout,
+            stdin=check_result.stdout,
+        )
+        return format_result.stdout.decode(self.encoding)
+
+    def _run_ruff_command(
+        self,
+        command: tuple[str, ...],
+        *,
+        stdin: bytes | None = None,
+        allow_stdout_on_error: bool = False,
+    ) -> subprocess.CompletedProcess[bytes]:
+        kwargs: dict[str, Any] = {}
+        if stdin is not None:
+            kwargs["input"] = stdin
+        result = subprocess.run(  # noqa: S603
+            command,
             capture_output=True,
             check=False,
             cwd=self.settings_path,
+            **kwargs,
         )
-        return format_result.stdout.decode(self.encoding)
+        if result.returncode == 0 or (allow_stdout_on_error and result.returncode == 1 and result.stdout):
+            return result
+        if (message := result.stderr.decode(self.encoding, errors="replace").strip()) or (
+            message := result.stdout.decode(self.encoding, errors="replace").strip()
+        ):
+            detail = message
+        else:
+            detail = "no output"
+        msg = f"Ruff command failed with exit code {result.returncode}: {' '.join(command)}\n{detail}"
+        raise RuntimeError(msg)
 
     def _ruff_check_command(self, *paths: str, ruff_path: str | None = None) -> tuple[str, ...]:
         """Build the Ruff check command for the current formatter settings."""
@@ -561,7 +579,10 @@ def _find_ruff_path() -> str:
         ruff_in_venv = bin_dir / ruff_name
         if ruff_in_venv.exists():
             return str(ruff_in_venv)
-        return shutil.which("ruff") or "ruff"  # pragma: no cover
+        if ruff_path := shutil.which("ruff"):
+            return ruff_path
+        msg = "Ruff executable was not found. Install it with `pip install 'datamodel-code-generator[ruff]'`."
+        raise RuntimeError(msg)
 
     def apply_isort(self, code: str) -> str:
         """Sort imports using isort."""
@@ -578,18 +599,12 @@ def format_directory(self, directory: Path) -> None:
         """Apply ruff formatting to all Python files in a directory."""
         ruff_path = self._find_ruff_path()
         if Formatter.RUFF_CHECK in self.formatters:
-            subprocess.run(  # noqa: S603
+            self._run_ruff_command(
                 self._ruff_check_command(str(directory), ruff_path=ruff_path),
-                capture_output=True,
-                check=False,
-                cwd=self.settings_path,
             )
         if Formatter.RUFF_FORMAT in self.formatters:
-            subprocess.run(  # noqa: S603
+            self._run_ruff_command(
                 (ruff_path, "format", str(directory)),
-                capture_output=True,
-                check=False,
-                cwd=self.settings_path,
             )
 
 

src/datamodel_code_generator/model/pydantic_v2/base_model.py

@@ -137,13 +137,11 @@ def _config_dict_items(config: Any) -> list[tuple[str, Any]]:
     if isinstance(config, dict):
         return list(config.items())
 
-    if model_dump := getattr(config, "model_dump", None):
-        return list(model_dump(exclude_unset=True).items())
-
-    if dict_ := getattr(config, "dict", None):
-        return list(dict_(exclude_unset=True).items())
-
-    return []
+    dump = getattr(config, "model_dump", None) or getattr(config, "dict", None)
+    if not dump:
+        return []
+    values = dump(exclude_unset=True)
+    return list(values.items()) if isinstance(values, dict) else []
 
 
 _PYDANTIC_V2_BASE_FIELD_KEYS: frozenset[str] = frozenset({
@@ -643,8 +641,11 @@ def __init__(  # noqa: PLR0913
             config_parameters[_ALIAS_GENERATOR_TEMPLATE_DATA_KEY] = alias_generator
             self._additional_imports.append(_ALIAS_GENERATOR_IMPORTS[alias_generator])
 
-        if isinstance(config := self.extra_template_data.get("config"), dict):
+        config = self.extra_template_data.get("config")
+        if isinstance(config, dict):
             config_parameters.update(dict(config.items()))
+        elif config_items := _config_dict_items(config):
+            config_parameters.update(dict(config_items))
 
         # Handle json_schema_extra from schema extensions (x-* fields)
         model_extras = self.extra_template_data.get("model_extras")
@@ -661,6 +662,7 @@ def __init__(  # noqa: PLR0913
             )
             self._additional_imports.append(IMPORT_CONFIG_DICT)
         else:
+            self.extra_template_data.pop("config", None)
             self.extra_template_data.pop(_CONFIG_ITEMS_TEMPLATE_DATA_KEY, None)
 
         self._process_schema_runtime_validation()