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

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

2from __future__ import annotations 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

3 

4from dataclasses import dataclass, field 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

5from typing import TypedDict 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

6 

7from pydantic_core.core_schema import ComputedField, CoreSchema, DefinitionReferenceSchema, SerSchema 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

8from typing_extensions import TypeAlias 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

9 

10AllSchemas: TypeAlias = 'CoreSchema | SerSchema | ComputedField' 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

11 

12 

13class GatherResult(TypedDict): 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

14 """Schema traversing result.""" 

15 

16 collected_references: dict[str, DefinitionReferenceSchema | None] 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

17 """The collected definition references. 31 ctx1abcdefghijklmnopqGHIJKLrstuvwxy

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] 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

25 """The list of core schemas having the discriminator application deferred.""" 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

26 

27 

28class MissingDefinitionError(LookupError): 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

30 

31 def __init__(self, schema_reference: str, /) -> None: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

32 self.schema_reference = schema_reference 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

33 

34 

35@dataclass 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

36class GatherContext: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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] 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

43 """The available definitions.""" 31 ctx1abcdefghijklmnopqGHIJKLrstuvwxy

44 

45 deferred_discriminator_schemas: list[CoreSchema] = field(init=False, default_factory=list) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

46 """The list of core schemas having the discriminator application deferred. 31 ctx1abcdefghijklmnopqGHIJKLrstuvwxy

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) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

52 """The collected definition references. 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

64 meta = schema.get('metadata') 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

65 if meta is not None and 'pydantic_internal_union_discriminator' in meta: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

66 ctx.deferred_discriminator_schemas.append(schema) # pyright: ignore[reportArgumentType] 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

67 

68 

69def traverse_definition_ref(def_ref_schema: DefinitionReferenceSchema, ctx: GatherContext) -> None: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

70 schema_ref = def_ref_schema['schema_ref'] 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

71 

72 if schema_ref not in ctx.collected_references: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

73 definition = ctx.definitions.get(schema_ref) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

74 if definition is None: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

75 raise MissingDefinitionError(schema_ref) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

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 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

80 traverse_schema(definition, ctx) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

81 if 'serialization' in def_ref_schema: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

82 traverse_schema(def_ref_schema['serialization'], ctx) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

83 traverse_metadata(def_ref_schema, ctx) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

88 

89 

90def traverse_schema(schema: AllSchemas, context: GatherContext) -> None: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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'] 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

95 

96 if schema_type == 'definition-ref': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

97 traverse_definition_ref(schema, context) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

99 return 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

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'}: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

105 if 'items_schema' in schema: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

106 traverse_schema(schema['items_schema'], context) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

107 elif schema_type == 'tuple': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

108 if 'items_schema' in schema: 108 ↛ 179line 108 didn't jump to line 179 because the condition on line 108 was always true32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

109 for s in schema['items_schema']: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

110 traverse_schema(s, context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

111 elif schema_type == 'dict': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

113 traverse_schema(schema['keys_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

114 if 'values_schema' in schema: 114 ↛ 179line 114 didn't jump to line 179 because the condition on line 114 was always true32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

115 traverse_schema(schema['values_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

116 elif schema_type == 'union': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

117 for choice in schema['choices']: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

118 if isinstance(choice, tuple): 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

119 traverse_schema(choice[0], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

120 else: 

121 traverse_schema(choice, context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

122 elif schema_type == 'tagged-union': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

123 for v in schema['choices'].values(): 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

124 traverse_schema(v, context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

125 elif schema_type == 'chain': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

126 for step in schema['steps']: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

127 traverse_schema(step, context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

128 elif schema_type == 'lax-or-strict': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

129 traverse_schema(schema['lax_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

130 traverse_schema(schema['strict_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

131 elif schema_type == 'json-or-python': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

132 traverse_schema(schema['json_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

133 traverse_schema(schema['python_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

134 elif schema_type in {'model-fields', 'typed-dict'}: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

135 if 'extras_schema' in schema: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

136 traverse_schema(schema['extras_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

137 if 'computed_fields' in schema: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

138 for s in schema['computed_fields']: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

139 traverse_schema(s, context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

140 for s in schema['fields'].values(): 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

141 traverse_schema(s, context) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

142 elif schema_type == 'dataclass-args': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

144 for s in schema['computed_fields']: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

145 traverse_schema(s, context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

146 for s in schema['fields']: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

147 traverse_schema(s, context) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

148 elif schema_type == 'arguments': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

149 for s in schema['arguments_schema']: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

150 traverse_schema(s['schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

151 if 'var_args_schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

152 traverse_schema(schema['var_args_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

153 if 'var_kwargs_schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

154 traverse_schema(schema['var_kwargs_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

155 elif schema_type == 'call': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

156 traverse_schema(schema['arguments_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

157 if 'return_schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

158 traverse_schema(schema['return_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

159 elif schema_type == 'computed-field': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

160 traverse_schema(schema['return_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

161 elif schema_type == 'function-plain': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

163 if 'return_schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

164 traverse_schema(schema['return_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

165 if 'json_schema_input_schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

166 traverse_schema(schema['json_schema_input_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

167 elif schema_type == 'function-wrap': 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

169 if 'return_schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

170 traverse_schema(schema['return_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

171 if 'schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

172 traverse_schema(schema['schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

173 if 'json_schema_input_schema' in schema: 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

174 traverse_schema(schema['json_schema_input_schema'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

175 else: 

176 if 'schema' in schema: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

177 traverse_schema(schema['schema'], context) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

178 

179 if 'serialization' in schema: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

180 traverse_schema(schema['serialization'], context) 32 ctx1zAabcdefghBiCDjklmnopqEFrstuvwxy

181 traverse_metadata(schema, context) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

182 

183 

184def gather_schemas_for_cleaning(schema: CoreSchema, definitions: dict[str, CoreSchema]) -> GatherResult: 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

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

186 

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

188 

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

190 `MissingDefinitionError` exception is raised. 

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

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

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

194 """ 

195 context = GatherContext(definitions) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

196 traverse_schema(schema, context) 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

197 

198 return { 39 ctx1zAabcdefghBiCDjklmnopqMGHIJKLEFrstuvwxy

199 'collected_references': context.collected_references, 

200 'deferred_discriminator_schemas': context.deferred_discriminator_schemas, 

201 }