Skip to content

Commit 12fc77c

Browse files
authored
Add defaultdict and Any to PYTHON_TYPE_IMPORTS (#2860)
* Add defaultdict and Any to PYTHON_TYPE_IMPORTS * Expand PYTHON_TYPE_IMPORTS and add dynamic fallback * Use try-except with noqa instead of suppress * Use suppress for ImportError handling * Add test for dynamic type import resolution
1 parent 8c7a9b2 commit 12fc77c

5 files changed

Lines changed: 173 additions & 2 deletions

File tree

docs/python-model.md

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -200,6 +200,66 @@ datamodel-codegen --input-model ./mymodule.py:OPENAPI_SPEC --input-file-type ope
200200

201201
---
202202

203+
## Custom Python Types with x-python-type {#x-python-type}
204+
205+
When using `x-python-type` in JSON Schema (via `WithJsonSchema` in Pydantic), the generator automatically resolves and generates the required imports.
206+
207+
### Automatic Import Resolution {#import-resolution}
208+
209+
The generator supports many common Python types out of the box:
210+
211+
| Module | Supported Types |
212+
|--------|-----------------|
213+
| `typing` | `Any`, `Union`, `Optional`, `Literal`, `Final`, `ClassVar`, `Annotated`, `TypeVar`, `TypeAlias`, `Never`, `NoReturn`, `Self`, `LiteralString`, `TypeGuard`, `Type` |
214+
| `collections` | `defaultdict`, `OrderedDict`, `Counter`, `deque`, `ChainMap` |
215+
| `collections.abc` | `Callable`, `Iterable`, `Iterator`, `Generator`, `Awaitable`, `Coroutine`, `AsyncIterable`, `AsyncIterator`, `AsyncGenerator`, `Mapping`, `MutableMapping`, `Sequence`, `MutableSequence`, `Set`, `MutableSet`, `Collection`, `Reversible` |
216+
| `pathlib` | `Path`, `PurePath` |
217+
| `decimal` | `Decimal` |
218+
| `uuid` | `UUID` |
219+
| `datetime` | `datetime`, `date`, `time`, `timedelta` |
220+
| `enum` | `Enum`, `IntEnum`, `StrEnum`, `Flag`, `IntFlag` |
221+
| `re` | `Pattern`, `Match` |
222+
223+
For types not in this list, the generator dynamically searches common modules to resolve imports.
224+
225+
### Example {#x-python-type-example}
226+
227+
**mymodule.py**
228+
```python
229+
from collections import defaultdict
230+
from typing import Any, Annotated
231+
from pydantic import BaseModel, Field, WithJsonSchema
232+
233+
class Config(BaseModel):
234+
data: Annotated[
235+
defaultdict[str, Annotated[dict[str, Any], Field(default_factory=dict)]],
236+
WithJsonSchema({'type': 'object', 'x-python-type': 'defaultdict[str, dict[str, Any]]'})
237+
] | None = None
238+
```
239+
240+
```bash
241+
datamodel-codegen --input-model ./mymodule.py:Config --output-model-type typing.TypedDict
242+
```
243+
244+
**✨ Generated output**
245+
```python
246+
from __future__ import annotations
247+
248+
from collections import defaultdict
249+
from typing import Any, TypedDict
250+
251+
from typing_extensions import NotRequired
252+
253+
254+
class Config(TypedDict):
255+
data: NotRequired[defaultdict[str, dict[str, Any]] | None]
256+
```
257+
258+
!!! tip "Fully Qualified Paths"
259+
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.
260+
261+
---
262+
203263
## Mutual Exclusion {#mutual-exclusion}
204264

205265
`--input-model` cannot be used with:

src/datamodel_code_generator/parser/jsonschema.py

Lines changed: 80 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
from __future__ import annotations
88

99
import enum as _enum
10+
import importlib
1011
import json
1112
import re
1213
from collections import defaultdict
@@ -561,6 +562,7 @@ class JsonSchemaParser(Parser):
561562
}
562563

563564
PYTHON_TYPE_IMPORTS: ClassVar[dict[str, Import]] = {
565+
# collections.abc
564566
"Callable": Import.from_full_path("collections.abc.Callable"),
565567
"Iterable": Import.from_full_path("collections.abc.Iterable"),
566568
"Iterator": Import.from_full_path("collections.abc.Iterator"),
@@ -570,9 +572,57 @@ class JsonSchemaParser(Parser):
570572
"AsyncIterable": Import.from_full_path("collections.abc.AsyncIterable"),
571573
"AsyncIterator": Import.from_full_path("collections.abc.AsyncIterator"),
572574
"AsyncGenerator": Import.from_full_path("collections.abc.AsyncGenerator"),
575+
"Mapping": Import.from_full_path("collections.abc.Mapping"),
576+
"MutableMapping": Import.from_full_path("collections.abc.MutableMapping"),
577+
"Sequence": Import.from_full_path("collections.abc.Sequence"),
578+
"MutableSequence": Import.from_full_path("collections.abc.MutableSequence"),
579+
"Set": Import.from_full_path("collections.abc.Set"),
580+
"MutableSet": Import.from_full_path("collections.abc.MutableSet"),
581+
"Collection": Import.from_full_path("collections.abc.Collection"),
582+
"Reversible": Import.from_full_path("collections.abc.Reversible"),
583+
# collections
584+
"defaultdict": Import.from_full_path("collections.defaultdict"),
585+
"OrderedDict": Import.from_full_path("collections.OrderedDict"),
586+
"Counter": Import.from_full_path("collections.Counter"),
587+
"deque": Import.from_full_path("collections.deque"),
588+
"ChainMap": Import.from_full_path("collections.ChainMap"),
589+
# re
573590
"Pattern": Import.from_full_path("re.Pattern"),
574591
"Match": Import.from_full_path("re.Match"),
592+
# typing
593+
"Any": Import.from_full_path("typing.Any"),
575594
"Type": Import.from_full_path("typing.Type"),
595+
"Union": Import.from_full_path("typing.Union"),
596+
"Optional": Import.from_full_path("typing.Optional"),
597+
"Literal": Import.from_full_path("typing.Literal"),
598+
"Final": Import.from_full_path("typing.Final"),
599+
"ClassVar": Import.from_full_path("typing.ClassVar"),
600+
"Annotated": Import.from_full_path("typing.Annotated"),
601+
"TypeVar": Import.from_full_path("typing.TypeVar"),
602+
"TypeAlias": Import.from_full_path("typing.TypeAlias"),
603+
"Never": Import.from_full_path("typing.Never"),
604+
"NoReturn": Import.from_full_path("typing.NoReturn"),
605+
"Self": Import.from_full_path("typing.Self"),
606+
"LiteralString": Import.from_full_path("typing.LiteralString"),
607+
"TypeGuard": Import.from_full_path("typing.TypeGuard"),
608+
# pathlib
609+
"Path": Import.from_full_path("pathlib.Path"),
610+
"PurePath": Import.from_full_path("pathlib.PurePath"),
611+
# decimal
612+
"Decimal": Import.from_full_path("decimal.Decimal"),
613+
# uuid
614+
"UUID": Import.from_full_path("uuid.UUID"),
615+
# datetime
616+
"datetime": Import.from_full_path("datetime.datetime"),
617+
"date": Import.from_full_path("datetime.date"),
618+
"time": Import.from_full_path("datetime.time"),
619+
"timedelta": Import.from_full_path("datetime.timedelta"),
620+
# enum
621+
"Enum": Import.from_full_path("enum.Enum"),
622+
"IntEnum": Import.from_full_path("enum.IntEnum"),
623+
"StrEnum": Import.from_full_path("enum.StrEnum"),
624+
"Flag": Import.from_full_path("enum.Flag"),
625+
"IntFlag": Import.from_full_path("enum.IntFlag"),
576626
}
577627

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

1408+
@staticmethod
1409+
@lru_cache(maxsize=256)
1410+
def _resolve_type_import_dynamic(type_name: str) -> Import | None:
1411+
"""Dynamically resolve import for a type name from known modules."""
1412+
modules_to_check = (
1413+
"typing",
1414+
"collections.abc",
1415+
"collections",
1416+
"pathlib",
1417+
"decimal",
1418+
"uuid",
1419+
"datetime",
1420+
"enum",
1421+
"re",
1422+
)
1423+
for module_name in modules_to_check:
1424+
with suppress(ImportError):
1425+
module = importlib.import_module(module_name)
1426+
if hasattr(module, type_name):
1427+
return Import.from_full_path(f"{module_name}.{type_name}")
1428+
return None
1429+
1430+
def _resolve_type_import(self, type_name: str) -> Import | None:
1431+
"""Resolve import for a type name, with dynamic fallback."""
1432+
if type_name in self.PYTHON_TYPE_IMPORTS:
1433+
return self.PYTHON_TYPE_IMPORTS[type_name]
1434+
return self._resolve_type_import_dynamic(type_name)
1435+
13581436
def _get_python_type_override(self, obj: JsonSchemaObject) -> DataType | None:
13591437
"""Get DataType from x-python-type if it's incompatible with schema type."""
13601438
x_python_type = obj.extras.get("x-python-type")
@@ -1366,7 +1444,7 @@ def _get_python_type_override(self, obj: JsonSchemaObject) -> DataType | None:
13661444
return None
13671445

13681446
base_type = self._get_python_type_base(x_python_type)
1369-
import_ = self.PYTHON_TYPE_IMPORTS.get(base_type)
1447+
import_ = self._resolve_type_import(base_type)
13701448

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
# generated by datamodel-codegen:
2+
# filename: x_python_type_dynamic_resolve.json
3+
# timestamp: 2019-07-26T00:00:00+00:00
4+
5+
from __future__ import annotations
6+
7+
from typing import NamedTuple, TypedDict
8+
9+
from typing_extensions import NotRequired
10+
11+
12+
class Model(TypedDict):
13+
point: NotRequired[NamedTuple]
Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
{
2+
"type": "object",
3+
"properties": {
4+
"point": {
5+
"type": "string",
6+
"x-python-type": "NamedTuple"
7+
}
8+
}
9+
}

tests/main/jsonschema/test_main_jsonschema.py

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7162,3 +7162,14 @@ def test_x_python_type_nested_unknown_type(output_file: Path) -> None:
71627162
assert_func=assert_file_content,
71637163
extra_args=["--output-model-type", "typing.TypedDict"],
71647164
)
7165+
7166+
7167+
def test_x_python_type_dynamic_resolve(output_file: Path) -> None:
7168+
"""Test x-python-type with types resolved dynamically (not in static PYTHON_TYPE_IMPORTS)."""
7169+
run_main_and_assert(
7170+
input_path=JSON_SCHEMA_DATA_PATH / "x_python_type_dynamic_resolve.json",
7171+
output_path=output_file,
7172+
input_file_type=None,
7173+
assert_func=assert_file_content,
7174+
extra_args=["--output-model-type", "typing.TypedDict"],
7175+
)

0 commit comments

Comments
 (0)