Coverage for pydantic/_internal/_schema_gather.py: 95.42%

1# pyright: reportTypedDictNotRequiredAccess=false, reportGeneralTypeIssues=false, reportArgumentType=false, reportAttributeAccessIssue=false 

2from __future__ import annotations 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

3 

4from dataclasses import dataclass, field 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

5from typing import TypedDict 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

6 

7from pydantic_core.core_schema import ComputedField, CoreSchema, DefinitionReferenceSchema, SerSchema 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

8from typing_extensions import TypeAlias 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

9 

10AllSchemas: TypeAlias = 'CoreSchema | SerSchema | ComputedField' 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

11 

12 

13class GatherResult(TypedDict): 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

14 """Schema traversing result.""" 

15 

16 collected_references: dict[str, DefinitionReferenceSchema | None] 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

17 """The collected definition references. 35 ctx1abcdefghijklmnopqrstuvwPxyzABCDEFGH

18 

19 If a definition reference schema can be inlined, it means that there is 

20 only one in the whole core schema. As such, it is stored as the value. 

21 Otherwise, the value is set to `None`. 

22 """ 

23 

24 deferred_discriminator_schemas: list[CoreSchema] 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

25 """The list of core schemas having the discriminator application deferred.""" 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

26 

27 

28class MissingDefinitionError(LookupError): 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

29 """A reference was pointing to a non-existing core schema.""" 

30 

31 def __init__(self, schema_reference: str, /) -> None: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

32 self.schema_reference = schema_reference 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

33 

34 

35@dataclass 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

36class GatherContext: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

37 """The current context used during core schema traversing. 

38 

39 Context instances should only be used during schema traversing. 

40 """ 

41 

42 definitions: dict[str, CoreSchema] 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

43 """The available definitions.""" 35 ctx1abcdefghijklmnopqrstuvwPxyzABCDEFGH

44 

45 deferred_discriminator_schemas: list[CoreSchema] = field(init=False, default_factory=list) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

46 """The list of core schemas having the discriminator application deferred. 35 ctx1abcdefghijklmnopqrstuvwPxyzABCDEFGH

47 

48 Internally, these core schemas have a specific key set in the core metadata dict. 

49 """ 

50 

51 collected_references: dict[str, DefinitionReferenceSchema | None] = field(init=False, default_factory=dict) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

52 """The collected definition references. 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

53 

54 If a definition reference schema can be inlined, it means that there is 

55 only one in the whole core schema. As such, it is stored as the value. 

56 Otherwise, the value is set to `None`. 

57 

58 During schema traversing, definition reference schemas can be added as candidates, or removed 

59 (by setting the value to `None`). 

60 """ 

61 

62 

63def traverse_metadata(schema: AllSchemas, ctx: GatherContext) -> None: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

64 meta = schema.get('metadata') 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

65 if meta is not None and 'pydantic_internal_union_discriminator' in meta: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

66 ctx.deferred_discriminator_schemas.append(schema) # pyright: ignore[reportArgumentType] 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

67 

68 

69def traverse_definition_ref(def_ref_schema: DefinitionReferenceSchema, ctx: GatherContext) -> None: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

70 schema_ref = def_ref_schema['schema_ref'] 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

71 

72 if schema_ref not in ctx.collected_references: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

73 definition = ctx.definitions.get(schema_ref) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

74 if definition is None: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

75 raise MissingDefinitionError(schema_ref) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

76 

77 # The `'definition-ref'` schema was only encountered once, make it 

78 # a candidate to be inlined: 

79 ctx.collected_references[schema_ref] = def_ref_schema 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

80 traverse_schema(definition, ctx) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

81 if 'serialization' in def_ref_schema: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

82 traverse_schema(def_ref_schema['serialization'], ctx) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

83 traverse_metadata(def_ref_schema, ctx) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

84 else: 

85 # The `'definition-ref'` schema was already encountered, meaning 

86 # the previously encountered schema (and this one) can't be inlined: 

87 ctx.collected_references[schema_ref] = None 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

88 

89 

90def traverse_schema(schema: AllSchemas, context: GatherContext) -> None: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

91 # TODO When we drop 3.9, use a match statement to get better type checking and remove 

92 # file-level type ignore. 

93 # (the `'type'` could also be fetched in every `if/elif` statement, but this alters performance). 

94 schema_type = schema['type'] 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

95 

96 if schema_type == 'definition-ref': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

97 traverse_definition_ref(schema, context) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

98 # `traverse_definition_ref` handles the possible serialization and metadata schemas: 

99 return 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

100 elif schema_type == 'definitions': 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

101 traverse_schema(schema['schema'], context) 

102 for definition in schema['definitions']: 

103 traverse_schema(definition, context) 

104 elif schema_type in {'list', 'set', 'frozenset', 'generator'}: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

105 if 'items_schema' in schema: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

106 traverse_schema(schema['items_schema'], context) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

107 elif schema_type == 'tuple': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

108 if 'items_schema' in schema: 108 ↛ 187line 108 didn't jump to line 187 because the condition on line 108 was always true41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

109 for s in schema['items_schema']: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

110 traverse_schema(s, context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

111 elif schema_type == 'dict': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

112 if 'keys_schema' in schema: 112 ↛ 114line 112 didn't jump to line 114 because the condition on line 112 was always true41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

113 traverse_schema(schema['keys_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

114 if 'values_schema' in schema: 114 ↛ 187line 114 didn't jump to line 187 because the condition on line 114 was always true41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

115 traverse_schema(schema['values_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

116 elif schema_type == 'union': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

117 for choice in schema['choices']: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

118 if isinstance(choice, tuple): 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

119 traverse_schema(choice[0], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

120 else: 

121 traverse_schema(choice, context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

122 elif schema_type == 'tagged-union': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

123 for v in schema['choices'].values(): 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

124 traverse_schema(v, context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

125 elif schema_type == 'chain': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

126 for step in schema['steps']: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

127 traverse_schema(step, context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

128 elif schema_type == 'lax-or-strict': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

129 traverse_schema(schema['lax_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

130 traverse_schema(schema['strict_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

131 elif schema_type == 'json-or-python': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

132 traverse_schema(schema['json_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

133 traverse_schema(schema['python_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

134 elif schema_type in {'model-fields', 'typed-dict'}: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

135 if 'extras_schema' in schema: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

136 traverse_schema(schema['extras_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

137 if 'computed_fields' in schema: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

138 for s in schema['computed_fields']: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

139 traverse_schema(s, context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

140 for s in schema['fields'].values(): 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

141 traverse_schema(s, context) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

142 elif schema_type == 'dataclass-args': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

143 if 'computed_fields' in schema: 143 ↛ 146line 143 didn't jump to line 146 because the condition on line 143 was always true42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

144 for s in schema['computed_fields']: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

145 traverse_schema(s, context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

146 for s in schema['fields']: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

147 traverse_schema(s, context) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

148 elif schema_type == 'arguments': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

149 for s in schema['arguments_schema']: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

150 traverse_schema(s['schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

151 if 'var_args_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

152 traverse_schema(schema['var_args_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

153 if 'var_kwargs_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

154 traverse_schema(schema['var_kwargs_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

155 elif schema_type == 'arguments-v3': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

156 for s in schema['arguments_schema']: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

157 traverse_schema(s['schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

158 elif schema_type == 'call': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

159 traverse_schema(schema['arguments_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

160 if 'return_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

161 traverse_schema(schema['return_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

162 elif schema_type == 'computed-field': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

163 traverse_schema(schema['return_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

164 elif schema_type == 'function-before': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

165 if 'schema' in schema: 165 ↛ 167line 165 didn't jump to line 167 because the condition on line 165 was always true41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

166 traverse_schema(schema['schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

167 if 'json_schema_input_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

168 traverse_schema(schema['json_schema_input_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

169 elif schema_type == 'function-plain': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

170 # TODO duplicate schema types for serializers and validators, needs to be deduplicated. 

171 if 'return_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

172 traverse_schema(schema['return_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

173 if 'json_schema_input_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

174 traverse_schema(schema['json_schema_input_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

175 elif schema_type == 'function-wrap': 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

176 # TODO duplicate schema types for serializers and validators, needs to be deduplicated. 

177 if 'return_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

178 traverse_schema(schema['return_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

179 if 'schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

180 traverse_schema(schema['schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

181 if 'json_schema_input_schema' in schema: 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

182 traverse_schema(schema['json_schema_input_schema'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

183 else: 

184 if 'schema' in schema: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

185 traverse_schema(schema['schema'], context) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

186 

187 if 'serialization' in schema: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

188 traverse_schema(schema['serialization'], context) 41 ctx1IJabcdefghijkKlLMmnopqrstuvwNOxyzABCDEFGH

189 traverse_metadata(schema, context) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

190 

191 

192def gather_schemas_for_cleaning(schema: CoreSchema, definitions: dict[str, CoreSchema]) -> GatherResult: 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

193 """Traverse the core schema and definitions and return the necessary information for schema cleaning. 

194 

195 During the core schema traversing, any `'definition-ref'` schema is: 

196 

197 - Validated: the reference must point to an existing definition. If this is not the case, a 

198 `MissingDefinitionError` exception is raised. 

199 - Stored in the context: the actual reference is stored in the context. Depending on whether 

200 the `'definition-ref'` schema is encountered more that once, the schema itself is also 

201 saved in the context to be inlined (i.e. replaced by the definition it points to). 

202 """ 

203 context = GatherContext(definitions) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

204 traverse_schema(schema, context) 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

205 

206 return { 42 ctx1IJabcdefghijkKlLMmnopqrstuvwPNOxyzABCDEFGH

207 'collected_references': context.collected_references, 

208 'deferred_discriminator_schemas': context.deferred_discriminator_schemas, 

209 }