Coverage for /pythoncovmergedfiles/medio/medio/src/pydantic/pydantic/_internal/_core_metadata.py: 72%

1from __future__ import annotations as _annotations 

2 

3import typing 

4from typing import Any 

5 

6import typing_extensions 

7from pydantic_core import CoreSchema, core_schema 

8 

9if typing.TYPE_CHECKING: 

10 from ..json_schema import JsonSchemaValue 

11 from ._json_schema_shared import ( 

12 CoreSchemaOrField as CoreSchemaOrField, 

13 ) 

14 from ._json_schema_shared import ( 

15 GetJsonSchemaFunction, 

16 GetJsonSchemaHandler, 

17 ) 

18 

19 

20class CoreMetadata(typing_extensions.TypedDict, total=False): 

21 # `pydantic_cs_update_function Retrieves the function that will be used to update the CoreSchema. 

22 # This is generally obtained from a `__pydantic_update_schema__` function 

23 pydantic_cs_update_function: UpdateCoreSchemaCallable | None 

24 

25 pydantic_js_functions: list[GetJsonSchemaFunction] 

26 

27 # If `pydantic_js_prefer_positional_arguments` is True, the JSON schema generator will 

28 # prefer positional over keyword arguments for an 'arguments' schema. 

29 pydantic_js_prefer_positional_arguments: bool | None 

30 

31 

32class UpdateCoreSchemaCallable(typing_extensions.Protocol): 

33 def __call__(self, schema: CoreSchema, **kwargs: Any) -> None: 

34 ... 

35 

36 

37class CoreMetadataHandler: 

38 """ 

39 Because the metadata field in pydantic_core is of type `Any`, we can't assume much about its contents. 

40 

41 This class is used to interact with the metadata field on a CoreSchema object in a consistent 

42 way throughout pydantic. 

43 """ 

44 

45 __slots__ = ('_schema',) 

46 

47 def __init__(self, schema: CoreSchema | core_schema.TypedDictField | core_schema.DataclassField): 

48 self._schema = schema 

49 

50 metadata = schema.get('metadata') 

51 if metadata is None: 

52 schema['metadata'] = CoreMetadata() 

53 elif not isinstance(metadata, dict): 

54 raise TypeError(f'CoreSchema metadata should be a dict; got {metadata!r}.') 

55 

56 @property 

57 def metadata(self) -> CoreMetadata: 

58 """ 

59 Retrieves the metadata dict off the schema, initializing it to a dict if it is None 

60 and raises an error if it is not a dict. 

61 """ 

62 metadata = self._schema.get('metadata') 

63 if metadata is None: 

64 self._schema['metadata'] = metadata = CoreMetadata() 

65 if not isinstance(metadata, dict): 

66 raise TypeError(f'CoreSchema metadata should be a dict; got {metadata!r}.') 

67 return metadata # type: ignore[return-value] 

68 

69 def get_json_schema( 

70 self, 

71 core_schema: CoreSchema | core_schema.TypedDictField | core_schema.DataclassField, 

72 handler: GetJsonSchemaHandler, 

73 ) -> JsonSchemaValue: 

74 js_function = self.metadata.get('pydantic_js_function') 

75 if js_function is None: 

76 return handler(core_schema) 

77 return js_function(core_schema, handler) 

78 

79 

80def build_metadata_dict( 

81 *, # force keyword arguments to make it easier to modify this signature in a backwards-compatible way 

82 cs_update_function: UpdateCoreSchemaCallable | None = None, 

83 js_functions: list[GetJsonSchemaFunction] | None = None, 

84 js_prefer_positional_arguments: bool | None = None, 

85 initial_metadata: Any | None = None, 

86) -> Any: 

87 """ 

88 Builds a dict to use as the metadata field of a CoreSchema object in a manner that is consistent 

89 with the CoreMetadataHandler class. 

90 """ 

91 if initial_metadata is not None and not isinstance(initial_metadata, dict): 

92 raise TypeError(f'CoreSchema metadata should be a dict; got {initial_metadata!r}.') 

93 

94 metadata = CoreMetadata( 

95 pydantic_cs_update_function=cs_update_function, 

96 pydantic_js_functions=js_functions or [], 

97 pydantic_js_prefer_positional_arguments=js_prefer_positional_arguments, 

98 ) 

99 metadata = {k: v for k, v in metadata.items() if v is not None} # type: ignore[assignment] 

100 

101 if initial_metadata is not None: 

102 metadata = {**initial_metadata, **metadata} # type: ignore[misc] 

103 

104 return metadata