Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions docs/python-model.md
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,66 @@ datamodel-codegen --input-model ./mymodule.py:OPENAPI_SPEC --input-file-type ope

---

## Custom Python Types with x-python-type {#x-python-type}

When using `x-python-type` in JSON Schema (via `WithJsonSchema` in Pydantic), the generator automatically resolves and generates the required imports.

### Automatic Import Resolution {#import-resolution}

The generator supports many common Python types out of the box:

| Module | Supported Types |
|--------|-----------------|
| `typing` | `Any`, `Union`, `Optional`, `Literal`, `Final`, `ClassVar`, `Annotated`, `TypeVar`, `TypeAlias`, `Never`, `NoReturn`, `Self`, `LiteralString`, `TypeGuard`, `Type` |
| `collections` | `defaultdict`, `OrderedDict`, `Counter`, `deque`, `ChainMap` |
| `collections.abc` | `Callable`, `Iterable`, `Iterator`, `Generator`, `Awaitable`, `Coroutine`, `AsyncIterable`, `AsyncIterator`, `AsyncGenerator`, `Mapping`, `MutableMapping`, `Sequence`, `MutableSequence`, `Set`, `MutableSet`, `Collection`, `Reversible` |
| `pathlib` | `Path`, `PurePath` |
| `decimal` | `Decimal` |
| `uuid` | `UUID` |
| `datetime` | `datetime`, `date`, `time`, `timedelta` |
| `enum` | `Enum`, `IntEnum`, `StrEnum`, `Flag`, `IntFlag` |
| `re` | `Pattern`, `Match` |

For types not in this list, the generator dynamically searches common modules to resolve imports.

### Example {#x-python-type-example}

**mymodule.py**
```python
from collections import defaultdict
from typing import Any, Annotated
from pydantic import BaseModel, Field, WithJsonSchema

class Config(BaseModel):
data: Annotated[
defaultdict[str, Annotated[dict[str, Any], Field(default_factory=dict)]],
WithJsonSchema({'type': 'object', 'x-python-type': 'defaultdict[str, dict[str, Any]]'})
] | None = None
```

```bash
datamodel-codegen --input-model ./mymodule.py:Config --output-model-type typing.TypedDict
```

**✨ Generated output**
```python
from __future__ import annotations

from collections import defaultdict
from typing import Any, TypedDict

from typing_extensions import NotRequired


class Config(TypedDict):
data: NotRequired[defaultdict[str, dict[str, Any]] | None]
```

!!! tip "Fully Qualified Paths"
You can also use fully qualified paths in `x-python-type` (e.g., `collections.defaultdict`), which are always resolved correctly regardless of the static mapping.

---

## Mutual Exclusion {#mutual-exclusion}

`--input-model` cannot be used with:
Expand Down
82 changes: 80 additions & 2 deletions src/datamodel_code_generator/parser/jsonschema.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from __future__ import annotations

import enum as _enum
import importlib
import json
import re
from collections import defaultdict
Expand Down Expand Up @@ -561,6 +562,7 @@ class JsonSchemaParser(Parser):
}

PYTHON_TYPE_IMPORTS: ClassVar[dict[str, Import]] = {
# collections.abc
"Callable": Import.from_full_path("collections.abc.Callable"),
"Iterable": Import.from_full_path("collections.abc.Iterable"),
"Iterator": Import.from_full_path("collections.abc.Iterator"),
Expand All @@ -570,9 +572,57 @@ class JsonSchemaParser(Parser):
"AsyncIterable": Import.from_full_path("collections.abc.AsyncIterable"),
"AsyncIterator": Import.from_full_path("collections.abc.AsyncIterator"),
"AsyncGenerator": Import.from_full_path("collections.abc.AsyncGenerator"),
"Mapping": Import.from_full_path("collections.abc.Mapping"),
"MutableMapping": Import.from_full_path("collections.abc.MutableMapping"),
"Sequence": Import.from_full_path("collections.abc.Sequence"),
"MutableSequence": Import.from_full_path("collections.abc.MutableSequence"),
"Set": Import.from_full_path("collections.abc.Set"),
"MutableSet": Import.from_full_path("collections.abc.MutableSet"),
"Collection": Import.from_full_path("collections.abc.Collection"),
"Reversible": Import.from_full_path("collections.abc.Reversible"),
# collections
"defaultdict": Import.from_full_path("collections.defaultdict"),
"OrderedDict": Import.from_full_path("collections.OrderedDict"),
"Counter": Import.from_full_path("collections.Counter"),
"deque": Import.from_full_path("collections.deque"),
"ChainMap": Import.from_full_path("collections.ChainMap"),
# re
"Pattern": Import.from_full_path("re.Pattern"),
"Match": Import.from_full_path("re.Match"),
# typing
"Any": Import.from_full_path("typing.Any"),
"Type": Import.from_full_path("typing.Type"),
"Union": Import.from_full_path("typing.Union"),
"Optional": Import.from_full_path("typing.Optional"),
"Literal": Import.from_full_path("typing.Literal"),
"Final": Import.from_full_path("typing.Final"),
"ClassVar": Import.from_full_path("typing.ClassVar"),
"Annotated": Import.from_full_path("typing.Annotated"),
"TypeVar": Import.from_full_path("typing.TypeVar"),
"TypeAlias": Import.from_full_path("typing.TypeAlias"),
"Never": Import.from_full_path("typing.Never"),
"NoReturn": Import.from_full_path("typing.NoReturn"),
"Self": Import.from_full_path("typing.Self"),
"LiteralString": Import.from_full_path("typing.LiteralString"),
"TypeGuard": Import.from_full_path("typing.TypeGuard"),
# pathlib
"Path": Import.from_full_path("pathlib.Path"),
"PurePath": Import.from_full_path("pathlib.PurePath"),
# decimal
"Decimal": Import.from_full_path("decimal.Decimal"),
# uuid
"UUID": Import.from_full_path("uuid.UUID"),
# datetime
"datetime": Import.from_full_path("datetime.datetime"),
"date": Import.from_full_path("datetime.date"),
"time": Import.from_full_path("datetime.time"),
"timedelta": Import.from_full_path("datetime.timedelta"),
# enum
"Enum": Import.from_full_path("enum.Enum"),
"IntEnum": Import.from_full_path("enum.IntEnum"),
"StrEnum": Import.from_full_path("enum.StrEnum"),
"Flag": Import.from_full_path("enum.Flag"),
"IntFlag": Import.from_full_path("enum.IntFlag"),
}

# Types that require x-python-type override regardless of schema type
Expand Down Expand Up @@ -1355,6 +1405,34 @@ def _extract_all_type_names(self, type_str: str) -> list[str]: # noqa: PLR6301
pattern = r"(?<![.\w])([A-Z]\w*)"
return re.findall(pattern, type_str)

@staticmethod
@lru_cache(maxsize=256)
def _resolve_type_import_dynamic(type_name: str) -> Import | None:
"""Dynamically resolve import for a type name from known modules."""
modules_to_check = (
"typing",
"collections.abc",
"collections",
"pathlib",
"decimal",
"uuid",
"datetime",
"enum",
"re",
)
for module_name in modules_to_check:
with suppress(ImportError):
module = importlib.import_module(module_name)
if hasattr(module, type_name):
return Import.from_full_path(f"{module_name}.{type_name}")
return None

def _resolve_type_import(self, type_name: str) -> Import | None:
"""Resolve import for a type name, with dynamic fallback."""
if type_name in self.PYTHON_TYPE_IMPORTS:
return self.PYTHON_TYPE_IMPORTS[type_name]
return self._resolve_type_import_dynamic(type_name)

def _get_python_type_override(self, obj: JsonSchemaObject) -> DataType | None:
"""Get DataType from x-python-type if it's incompatible with schema type."""
x_python_type = obj.extras.get("x-python-type")
Expand All @@ -1366,7 +1444,7 @@ def _get_python_type_override(self, obj: JsonSchemaObject) -> DataType | None:
return None

base_type = self._get_python_type_base(x_python_type)
import_ = self.PYTHON_TYPE_IMPORTS.get(base_type)
import_ = self._resolve_type_import(base_type)

# Convert fully qualified path to short name when import is added
type_str = x_python_type
Expand All @@ -1389,7 +1467,7 @@ def _get_python_type_override(self, obj: JsonSchemaObject) -> DataType | None:
# Collect imports for all nested types (e.g., Iterable inside Callable[[Iterable[str]], str])
for type_name in self._extract_all_type_names(type_str):
if type_name != base_type:
nested_import = self.PYTHON_TYPE_IMPORTS.get(type_name)
nested_import = self._resolve_type_import(type_name)
if nested_import:
nested_imports.append(self.data_type(import_=nested_import))

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
# generated by datamodel-codegen:
# filename: x_python_type_dynamic_resolve.json
# timestamp: 2019-07-26T00:00:00+00:00

from __future__ import annotations

from typing import NamedTuple, TypedDict

from typing_extensions import NotRequired


class Model(TypedDict):
point: NotRequired[NamedTuple]
9 changes: 9 additions & 0 deletions tests/data/jsonschema/x_python_type_dynamic_resolve.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
{
"type": "object",
"properties": {
"point": {
"type": "string",
"x-python-type": "NamedTuple"
}
}
}
11 changes: 11 additions & 0 deletions tests/main/jsonschema/test_main_jsonschema.py
Original file line number Diff line number Diff line change
Expand Up @@ -7162,3 +7162,14 @@ def test_x_python_type_nested_unknown_type(output_file: Path) -> None:
assert_func=assert_file_content,
extra_args=["--output-model-type", "typing.TypedDict"],
)


def test_x_python_type_dynamic_resolve(output_file: Path) -> None:
"""Test x-python-type with types resolved dynamically (not in static PYTHON_TYPE_IMPORTS)."""
run_main_and_assert(
input_path=JSON_SCHEMA_DATA_PATH / "x_python_type_dynamic_resolve.json",
output_path=output_file,
input_file_type=None,
assert_func=assert_file_content,
extra_args=["--output-model-type", "typing.TypedDict"],
)
Loading