1from __future__ import annotations
2
3from typing import TypedDict
4
5from pydantic_core._pydantic_core import _schema_gather
6from pydantic_core.core_schema import CoreSchema, DefinitionReferenceSchema
7
8__all__ = [
9 'GatherResult',
10 'MissingDefinitionError',
11 'gather_schemas_for_cleaning',
12]
13
14MissingDefinitionError = _schema_gather.MissingDefinitionError
15
16
17class GatherResult(TypedDict):
18 """Schema traversing result."""
19
20 collected_references: dict[str, DefinitionReferenceSchema | None]
21 """The collected definition references.
22
23 If a definition reference schema can be inlined, it means that there is
24 only one in the whole core schema. As such, it is stored as the value.
25 Otherwise, the value is set to `None`.
26 """
27
28 deferred_discriminator_schemas: list[CoreSchema]
29 """The list of core schemas having the discriminator application deferred."""
30
31
32def gather_schemas_for_cleaning(schema: CoreSchema, definitions: dict[str, CoreSchema]) -> GatherResult:
33 """Traverse the core schema and definitions and return the necessary information for schema cleaning.
34
35 During the core schema traversing, any `'definition-ref'` schema is:
36
37 - Validated: the reference must point to an existing definition. If this is not the case, a
38 `MissingDefinitionError` exception is raised.
39 - Stored in the context: the actual reference is stored in the context. Depending on whether
40 the `'definition-ref'` schema is encountered more that once, the schema itself is also
41 saved in the context to be inlined (i.e. replaced by the definition it points to).
42 """
43 collected_references, deferred_discriminator_schemas = _schema_gather.gather_schemas_for_cleaning(
44 schema, definitions
45 )
46 return {
47 'collected_references': collected_references,
48 'deferred_discriminator_schemas': deferred_discriminator_schemas,
49 }