Add --type-mappings option to customize type mappings in JSON schema (#2559)

* Add --type-mappings option to customize type mappings in JSON schema

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

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

* Add type_mappings parameter to GraphQL and OpenAPI parsers

* Add tests for --type-mappings option and validation logic

* Add --type-mappings option to override default type mappings in documentation

* Add test for fallback behavior in _get_type_with_mappings method

---------

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

Author: koxudaxi

README.md

@@ -380,6 +380,11 @@ Typing customization:
                         Set enum members as default values for enum field
   --strict-types {str,bytes,int,float,bool} [{str,bytes,int,float,bool} ...]
                         Use strict types
+  --type-mappings TYPE_MAPPINGS [TYPE_MAPPINGS ...]
+                        Override default type mappings. Format: "type+format=target" (e.g.,
+                        "string+binary=string" to map binary format to string type) or
+                        "format=target" (e.g., "binary=string"). Can be specified multiple
+                        times.
   --use-annotated       Use typing.Annotated for Field(). Also, `--field-constraints` option
                         will be enabled.
   --use-generic-container-types

docs/index.md

@@ -372,6 +372,11 @@ Typing customization:
                         Set enum members as default values for enum field
   --strict-types {str,bytes,int,float,bool} [{str,bytes,int,float,bool} ...]
                         Use strict types
+  --type-mappings TYPE_MAPPINGS [TYPE_MAPPINGS ...]
+                        Override default type mappings. Format: "type+format=target" (e.g.,
+                        "string+binary=string" to map binary format to string type) or
+                        "format=target" (e.g., "binary=string"). Can be specified multiple
+                        times.
   --use-annotated       Use typing.Annotated for Field(). Also, `--field-constraints` option
                         will be enabled.
   --use-generic-container-types

src/datamodel_code_generator/__init__.py

@@ -340,6 +340,7 @@ def generate(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
     formatters: list[Formatter] = DEFAULT_FORMATTERS,
     parent_scoped_naming: bool = False,
     disable_future_imports: bool = False,
+    type_mappings: list[str] | None = None,
 ) -> None:
     """Generate Python data models from schema definitions or structured data.
 
@@ -560,6 +561,7 @@ def get_header_and_first_line(csv_file: IO[str]) -> dict[str, Any]:
         formatters=formatters,
         encoding=encoding,
         parent_scoped_naming=parent_scoped_naming,
+        type_mappings=type_mappings,
         **kwargs,
     )
 

src/datamodel_code_generator/__main__.py

@@ -406,6 +406,7 @@ def validate_root(cls, values: dict[str, Any]) -> dict[str, Any]:  # noqa: N805
     formatters: list[Formatter] = DEFAULT_FORMATTERS
     parent_scoped_naming: bool = False
     disable_future_imports: bool = False
+    type_mappings: Optional[list[str]] = None  # noqa: UP045
 
     def merge_args(self, args: Namespace) -> None:
         """Merge command-line arguments into config."""
@@ -660,6 +661,7 @@ def main(args: Sequence[str] | None = None) -> Exit:  # noqa: PLR0911, PLR0912,
             formatters=config.formatters,
             parent_scoped_naming=config.parent_scoped_naming,
             disable_future_imports=config.disable_future_imports,
+            type_mappings=config.type_mappings,
         )
     except InvalidClassNameError as e:
         print(f"{e} You have to set `--class-name` option", file=sys.stderr)  # noqa: T201

src/datamodel_code_generator/arguments.py

@@ -336,6 +336,16 @@ def start_section(self, heading: str | None) -> None:
     action="store_true",
     default=None,
 )
+typing_options.add_argument(
+    "--type-mappings",
+    help="Override default type mappings. "
+    'Format: "type+format=target" (e.g., "string+binary=string" to map binary format to string type) '
+    'or "format=target" (e.g., "binary=string"). '
+    "Can be specified multiple times.",
+    nargs="+",
+    type=str,
+    default=None,
+)
 
 # ======================================================================================
 # Customization options for generated model fields

src/datamodel_code_generator/parser/base.py

@@ -429,6 +429,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         no_alias: bool = False,
         formatters: list[Formatter] = DEFAULT_FORMATTERS,
         parent_scoped_naming: bool = False,
+        type_mappings: list[str] | None = None,
     ) -> None:
         """Initialize the Parser with configuration options."""
         self.keyword_only = keyword_only
@@ -548,6 +549,38 @@ def __init__(  # noqa: PLR0913, PLR0915
         self.treat_dot_as_module = treat_dot_as_module
         self.default_field_extras: dict[str, Any] | None = default_field_extras
         self.formatters: list[Formatter] = formatters
+        self.type_mappings: dict[tuple[str, str], str] = Parser._parse_type_mappings(type_mappings)
+
+    @staticmethod
+    def _parse_type_mappings(type_mappings: list[str] | None) -> dict[tuple[str, str], str]:
+        """Parse type mappings from CLI format to internal format.
+
+        Supports two formats:
+        - "type+format=target" (e.g., "string+binary=string")
+        - "format=target" (e.g., "binary=string", assumes type="string")
+
+        Returns a dict mapping (type, format) tuples to target type names.
+        """
+        if not type_mappings:
+            return {}
+
+        result: dict[tuple[str, str], str] = {}
+        for mapping in type_mappings:
+            if "=" not in mapping:
+                msg = f"Invalid type mapping format: {mapping!r}. Expected 'type+format=target' or 'format=target'."
+                raise ValueError(msg)
+
+            source, target = mapping.split("=", 1)
+            if "+" in source:
+                type_, format_ = source.split("+", 1)
+            else:
+                # Default to "string" type if only format is specified
+                type_ = "string"
+                format_ = source
+
+            result[type_, format_] = target
+
+        return result
 
     @property
     def iter_source(self) -> Iterator[Source]:

src/datamodel_code_generator/parser/graphql.py

@@ -173,6 +173,7 @@ def __init__(  # noqa: PLR0913
         no_alias: bool = False,
         formatters: list[Formatter] = DEFAULT_FORMATTERS,
         parent_scoped_naming: bool = False,
+        type_mappings: list[str] | None = None,
     ) -> None:
         """Initialize the GraphQL parser with configuration options."""
         super().__init__(
@@ -254,6 +255,7 @@ def __init__(  # noqa: PLR0913
             no_alias=no_alias,
             formatters=formatters,
             parent_scoped_naming=parent_scoped_naming,
+            type_mappings=type_mappings,
         )
 
         self.data_model_scalar_type = data_model_scalar_type

src/datamodel_code_generator/parser/jsonschema.py

@@ -525,6 +525,7 @@ def __init__(  # noqa: PLR0913
         no_alias: bool = False,
         formatters: list[Formatter] = DEFAULT_FORMATTERS,
         parent_scoped_naming: bool = False,
+        type_mappings: list[str] | None = None,
     ) -> None:
         """Initialize the JSON Schema parser with configuration options."""
         target_datetime_class = target_datetime_class or DatetimeClassType.Awaredatetime
@@ -607,6 +608,7 @@ def __init__(  # noqa: PLR0913
             no_alias=no_alias,
             formatters=formatters,
             parent_scoped_naming=parent_scoped_naming,
+            type_mappings=type_mappings,
         )
 
         self.remote_object_cache: DefaultPutDict[str, dict[str, YamlValue]] = DefaultPutDict()
@@ -645,6 +647,22 @@ def get_field_extras(self, obj: JsonSchemaObject) -> dict[str, Any]:
             extras.update(self.default_field_extras)
         return extras
 
+    def _get_type_with_mappings(self, type_: str, format_: str | None = None) -> Types:
+        """Get the Types enum for a given type and format, applying custom type mappings.
+
+        Custom mappings from --type-mappings are checked first, then falls back to
+        the default json_schema_data_formats mappings.
+        """
+        if self.type_mappings and format_ is not None and (type_, format_) in self.type_mappings:
+            target_format = self.type_mappings[type_, format_]
+            for type_formats in json_schema_data_formats.values():
+                if target_format in type_formats:
+                    return type_formats[target_format]
+            if target_format in json_schema_data_formats:
+                return json_schema_data_formats[target_format]["default"]
+
+        return _get_type(type_, format_)
+
     @cached_property
     def schema_paths(self) -> list[tuple[str, list[str]]]:
         """Get schema paths for definitions and defs."""
@@ -713,7 +731,7 @@ def get_data_type(self, obj: JsonSchemaObject) -> DataType:
 
         def _get_data_type(type_: str, format__: str) -> DataType:
             return self.data_type_manager.get_data_type(
-                _get_type(type_, format__),
+                self._get_type_with_mappings(type_, format__),
                 **obj.dict() if not self.field_constraints else {},
             )
 
@@ -1493,7 +1511,9 @@ def parse_enum(
             )
 
         def create_enum(reference_: Reference) -> DataType:
-            type_: Types | None = _get_type(obj.type, obj.format) if isinstance(obj.type, str) else None
+            type_: Types | None = (
+                self._get_type_with_mappings(obj.type, obj.format) if isinstance(obj.type, str) else None
+            )
 
             enum_cls: type[Enum] = Enum
             if (

src/datamodel_code_generator/parser/openapi.py

@@ -254,6 +254,7 @@ def __init__(  # noqa: PLR0913
         no_alias: bool = False,
         formatters: list[Formatter] = DEFAULT_FORMATTERS,
         parent_scoped_naming: bool = False,
+        type_mappings: list[str] | None = None,
     ) -> None:
         """Initialize the OpenAPI parser with extensive configuration options."""
         target_datetime_class = target_datetime_class or DatetimeClassType.Awaredatetime
@@ -336,6 +337,7 @@ def __init__(  # noqa: PLR0913
             no_alias=no_alias,
             formatters=formatters,
             parent_scoped_naming=parent_scoped_naming,
+            type_mappings=type_mappings,
         )
         self.open_api_scopes: list[OpenAPIScope] = openapi_scopes or [OpenAPIScope.Schemas]
         self.include_path_parameters: bool = include_path_parameters

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

@@ -0,0 +1,15 @@
+# generated by datamodel-codegen:
+#   filename:  type_mappings.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from pydantic import Base64Str, BaseModel, Field
+
+
+class BlobModel(BaseModel):
+    content: str = Field(
+        ..., description='Binary content that should be mapped to string'
+    )
+    data: Base64Str = Field(..., description='Base64 encoded data')
+    name: str = Field(..., description='Regular string field')