1from typing import Any
2from uuid import UUID
3
4
5def merge_data(data1, data2):
6 merged = data1.copy()
7 for key, value in data2.items():
8 if key in merged:
9 if isinstance(merged[key], dict) and isinstance(value, dict):
10 merged[key] = merge_data(merged[key], value)
11 else:
12 merged[key] = value
13 else:
14 merged[key] = value
15 return merged
16
17
18def schema_from_type(t) -> dict[str, Any]:
19 # if it's a union with None, add nullable: true
20
21 # if t has a comment, add description
22 # import inspect
23 # if description := inspect.getdoc(t):
24 # extra_fields = {"description": description}
25 # else:
26 extra_fields: dict[str, Any] = {}
27
28 if hasattr(t, "__annotations__") and t.__annotations__:
29 # It's a TypedDict...
30 return {
31 "type": "object",
32 "properties": {
33 k: schema_from_type(v) for k, v in t.__annotations__.items()
34 },
35 **extra_fields,
36 }
37
38 if hasattr(t, "__origin__"):
39 if t.__origin__ is list:
40 return {
41 "type": "array",
42 "items": schema_from_type(t.__args__[0]),
43 **extra_fields,
44 }
45 elif t.__origin__ is dict:
46 return {
47 "type": "object",
48 "properties": {
49 k: schema_from_type(v)
50 for k, v in t.__args__[1].__annotations__.items()
51 },
52 **extra_fields,
53 }
54 else:
55 raise ValueError(f"Unknown type: {t}")
56
57 if hasattr(t, "__args__") and len(t.__args__) == 2 and type(None) in t.__args__:
58 return {
59 **schema_from_type(t.__args__[0]),
60 "nullable": True,
61 **extra_fields,
62 }
63
64 type_mappings: dict[Any, dict] = {
65 str: {
66 "type": "string",
67 },
68 int: {
69 "type": "integer",
70 },
71 float: {
72 "type": "number",
73 },
74 bool: {
75 "type": "boolean",
76 },
77 dict: {
78 "type": "object",
79 },
80 list: {
81 "type": "array",
82 },
83 UUID: {
84 "type": "string",
85 "format": "uuid",
86 },
87 }
88
89 schema = type_mappings.get(t, {})
90 if not schema:
91 schema = type_mappings.get(t.__class__, {})
92 if not schema:
93 raise ValueError(f"Unknown type: {t}")
94
95 return {**schema, **extra_fields}