1from types import NoneType, UnionType
2from typing import (
3 Any,
4 Literal,
5 NotRequired,
6 Required,
7 Union,
8 get_args,
9 get_origin,
10 get_type_hints,
11)
12from uuid import UUID
13
14
15def merge_data(data1: dict[str, Any], data2: dict[str, Any]) -> dict[str, Any]:
16 merged = data1.copy()
17 for key, value in data2.items():
18 if key in merged:
19 if isinstance(merged[key], dict) and isinstance(value, dict):
20 merged[key] = merge_data(merged[key], value)
21 else:
22 merged[key] = value
23 else:
24 merged[key] = value
25 return merged
26
27
28def is_typed_dict(t: Any) -> bool:
29 """TypedDict classes carry `__required_keys__` / `__optional_keys__` set by the metaclass."""
30 return (
31 isinstance(t, type)
32 and hasattr(t, "__required_keys__")
33 and hasattr(t, "__optional_keys__")
34 )
35
36
37_PRIMITIVE_SCHEMAS: dict[Any, dict[str, Any]] = {
38 str: {"type": "string"},
39 int: {"type": "integer"},
40 float: {"type": "number"},
41 bool: {"type": "boolean"},
42 dict: {"type": "object"},
43 list: {"type": "array"},
44 UUID: {"type": "string", "format": "uuid"},
45}
46
47
48def schema_from_type(
49 t: Any,
50 *,
51 components: dict[str, Any] | None = None,
52) -> dict[str, Any]:
53 """Translate a Python type to an OpenAPI schema. With `components`, register TypedDicts there and return `$ref`s; without, inline them."""
54 origin = get_origin(t)
55
56 if origin in (NotRequired, Required):
57 return schema_from_type(get_args(t)[0], components=components)
58
59 if origin is Literal:
60 values = list(get_args(t))
61 value_types = {type(v) for v in values}
62 if value_types == {str}:
63 return {"type": "string", "enum": values}
64 if value_types == {int}:
65 return {"type": "integer", "enum": values}
66 if value_types == {bool}:
67 return {"type": "boolean", "enum": values}
68 return {"enum": values}
69
70 if is_typed_dict(t):
71 if components is not None:
72 schemas = components.setdefault("schemas", {})
73 name = t.__name__
74 if name not in schemas:
75 # Placeholder breaks self-referential cycles.
76 schemas[name] = {}
77 schemas[name] = _typed_dict_body(t, components=components)
78 return {"$ref": f"#/components/schemas/{name}"}
79 return _typed_dict_body(t, components=None)
80
81 # Unions are matched before the subscripted-generic branches below. Python
82 # 3.14 unified `types.UnionType` with `typing.Union`, so `str | None` now
83 # reports an origin (`typing.Union`) just like `list[str]` reports `list`.
84 # An origin check that ran first and rejected what it didn't recognize
85 # would turn every optional field into "Unknown type".
86 if origin in (Union, UnionType):
87 return _union_schema(t, components=components)
88
89 if origin is list:
90 return {
91 "type": "array",
92 "items": schema_from_type(get_args(t)[0], components=components),
93 }
94
95 if origin is dict:
96 return {
97 "type": "object",
98 "additionalProperties": schema_from_type(
99 get_args(t)[1], components=components
100 ),
101 }
102
103 if origin is not None:
104 raise ValueError(f"Unknown type: {t}")
105
106 schema = _PRIMITIVE_SCHEMAS.get(t) or _PRIMITIVE_SCHEMAS.get(t.__class__)
107 if schema is None:
108 raise ValueError(f"Unknown type: {t}")
109 return dict(schema)
110
111
112def _union_schema(
113 t: Any,
114 *,
115 components: dict[str, Any] | None,
116) -> dict[str, Any]:
117 """Translate a union — `X | None`, `Optional[X]`, `Union[X, Y]` — to an OpenAPI 3.0 schema.
118
119 One named member (the `X | None` case) keeps that member's schema and adds
120 `"nullable": True`. Several named members become an `anyOf`, since OpenAPI
121 3.0 has no sum type of its own. `None` is always carried by `nullable`,
122 which is what 3.0 has in place of a `null` type.
123 """
124 args = get_args(t)
125 named = [arg for arg in args if arg is not NoneType]
126
127 if len(named) == 1:
128 schema = schema_from_type(named[0], components=components)
129 else:
130 schema = {
131 "anyOf": [schema_from_type(arg, components=components) for arg in named]
132 }
133
134 if len(named) < len(args):
135 schema["nullable"] = True
136
137 return schema
138
139
140def _typed_dict_body(
141 t: Any,
142 *,
143 components: dict[str, Any] | None,
144) -> dict[str, Any]:
145 """Render a TypedDict body; `get_type_hints` resolves string-form annotations."""
146 try:
147 hints = get_type_hints(t, include_extras=True)
148 except Exception:
149 hints = t.__annotations__
150 return {
151 "type": "object",
152 "properties": {
153 k: schema_from_type(v, components=components) for k, v in hints.items()
154 },
155 }
156
157
158def typed_dict_from_annotation(annotation: Any) -> type | None:
159 """Return the first TypedDict found in `annotation`, walking unions like `MyDict | Response | None`."""
160 if is_typed_dict(annotation):
161 return annotation
162
163 if get_origin(annotation) in (Union, UnionType):
164 for arg in get_args(annotation):
165 if arg is NoneType:
166 continue
167 if is_typed_dict(arg):
168 return arg
169 return None