Coverage for pydantic/_internal/_core_utils.py: 55.93%

86 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-02-13 19:35 +0000

1from __future__ import annotations 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

2 

3import inspect 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

4import os 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

5from collections.abc import Mapping, Sequence 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

6from typing import TYPE_CHECKING, Any, Union 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

7 

8from pydantic_core import CoreSchema, core_schema 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

9from pydantic_core import validate_core_schema as _validate_core_schema 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

10from typing_extensions import TypeGuard, get_args, get_origin 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

11 

12from . import _repr 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

13from ._typing_extra import is_generic_alias, is_type_alias_type 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

14 

15if TYPE_CHECKING: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

16 from rich.console import Console 

17 

18AnyFunctionSchema = Union[ 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

19 core_schema.AfterValidatorFunctionSchema, 

20 core_schema.BeforeValidatorFunctionSchema, 

21 core_schema.WrapValidatorFunctionSchema, 

22 core_schema.PlainValidatorFunctionSchema, 

23] 

24 

25 

26FunctionSchemaWithInnerSchema = Union[ 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

27 core_schema.AfterValidatorFunctionSchema, 

28 core_schema.BeforeValidatorFunctionSchema, 

29 core_schema.WrapValidatorFunctionSchema, 

30] 

31 

32CoreSchemaField = Union[ 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

33 core_schema.ModelField, core_schema.DataclassField, core_schema.TypedDictField, core_schema.ComputedField 

34] 

35CoreSchemaOrField = Union[core_schema.CoreSchema, CoreSchemaField] 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

36 

37_CORE_SCHEMA_FIELD_TYPES = {'typed-dict-field', 'dataclass-field', 'model-field', 'computed-field'} 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

38_FUNCTION_WITH_INNER_SCHEMA_TYPES = {'function-before', 'function-after', 'function-wrap'} 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

39_LIST_LIKE_SCHEMA_WITH_ITEMS_TYPES = {'list', 'set', 'frozenset'} 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

40 

41 

42def is_core_schema( 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

43 schema: CoreSchemaOrField, 

44) -> TypeGuard[CoreSchema]: 

45 return schema['type'] not in _CORE_SCHEMA_FIELD_TYPES 1abcdefghijklmnopqrstuvwxyzABCDEF

46 

47 

48def is_core_schema_field( 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

49 schema: CoreSchemaOrField, 

50) -> TypeGuard[CoreSchemaField]: 

51 return schema['type'] in _CORE_SCHEMA_FIELD_TYPES 1abcdefghijklmnopqrstuvwxyzABCDEF

52 

53 

54def is_function_with_inner_schema( 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

55 schema: CoreSchemaOrField, 

56) -> TypeGuard[FunctionSchemaWithInnerSchema]: 

57 return schema['type'] in _FUNCTION_WITH_INNER_SCHEMA_TYPES 1abcdefghijklmnopqrstuvwxyzABCDEF

58 

59 

60def is_list_like_schema_with_items_schema( 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

61 schema: CoreSchema, 

62) -> TypeGuard[core_schema.ListSchema | core_schema.SetSchema | core_schema.FrozenSetSchema]: 

63 return schema['type'] in _LIST_LIKE_SCHEMA_WITH_ITEMS_TYPES 1abcdefghijklmnopqrstuvwxyzABCDEF

64 

65 

66def get_type_ref(type_: Any, args_override: tuple[type[Any], ...] | None = None) -> str: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

67 """Produces the ref to be used for this type by pydantic_core's core schemas. 

68 

69 This `args_override` argument was added for the purpose of creating valid recursive references 

70 when creating generic models without needing to create a concrete class. 

71 """ 

72 origin = get_origin(type_) or type_ 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

73 

74 args = get_args(type_) if is_generic_alias(type_) else (args_override or ()) 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

75 generic_metadata = getattr(type_, '__pydantic_generic_metadata__', None) 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

76 if generic_metadata: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

77 origin = generic_metadata['origin'] or origin 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

78 args = generic_metadata['args'] or args 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

79 

80 module_name = getattr(origin, '__module__', '<No __module__>') 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

81 if is_type_alias_type(origin): 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

82 type_ref = f'{module_name}.{origin.__name__}:{id(origin)}' 1abcdefghijklmnopqrstuvwxyzABCDEF

83 else: 

84 try: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

85 qualname = getattr(origin, '__qualname__', f'<No __qualname__: {origin}>') 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

86 except Exception: 

87 qualname = getattr(origin, '__qualname__', '<No __qualname__>') 

88 type_ref = f'{module_name}.{qualname}:{id(origin)}' 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

89 

90 arg_refs: list[str] = [] 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

91 for arg in args: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

92 if isinstance(arg, str): 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

93 # Handle string literals as a special case; we may be able to remove this special handling if we 

94 # wrap them in a ForwardRef at some point. 

95 arg_ref = f'{arg}:str-{id(arg)}' 1abcdefghijklmnopqrstuvwxyzABCDEF

96 else: 

97 arg_ref = f'{_repr.display_as_type(arg)}:{id(arg)}' 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

98 arg_refs.append(arg_ref) 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

99 if arg_refs: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

100 type_ref = f'{type_ref}[{",".join(arg_refs)}]' 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

101 return type_ref 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

102 

103 

104def get_ref(s: core_schema.CoreSchema) -> None | str: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

105 """Get the ref from the schema if it has one. 

106 This exists just for type checking to work correctly. 

107 """ 

108 return s.get('ref', None) 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

109 

110 

111def validate_core_schema(schema: CoreSchema) -> CoreSchema: 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

112 if os.getenv('PYDANTIC_VALIDATE_CORE_SCHEMAS'): 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

113 return _validate_core_schema(schema) 

114 return schema 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

115 

116 

117def _clean_schema_for_pretty_print(obj: Any, strip_metadata: bool = True) -> Any: # pragma: nocover 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

118 """A utility function to remove irrelevant information from a core schema.""" 

119 if isinstance(obj, Mapping): 

120 new_dct = {} 

121 for k, v in obj.items(): 

122 if k == 'metadata' and strip_metadata: 

123 new_metadata = {} 

124 

125 for meta_k, meta_v in v.items(): 

126 if meta_k in ('pydantic_js_functions', 'pydantic_js_annotation_functions'): 

127 new_metadata['js_metadata'] = '<stripped>' 

128 else: 

129 new_metadata[meta_k] = _clean_schema_for_pretty_print(meta_v, strip_metadata=strip_metadata) 

130 

131 if list(new_metadata.keys()) == ['js_metadata']: 

132 new_metadata = {'<stripped>'} 

133 

134 new_dct[k] = new_metadata 

135 # Remove some defaults: 

136 elif k in ('custom_init', 'root_model') and not v: 

137 continue 

138 else: 

139 new_dct[k] = _clean_schema_for_pretty_print(v, strip_metadata=strip_metadata) 

140 

141 return new_dct 

142 elif isinstance(obj, Sequence) and not isinstance(obj, str): 

143 return [_clean_schema_for_pretty_print(v, strip_metadata=strip_metadata) for v in obj] 

144 else: 

145 return obj 

146 

147 

148def pretty_print_core_schema( 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF

149 val: Any, 

150 *, 

151 console: Console | None = None, 

152 max_depth: int | None = None, 

153 strip_metadata: bool = True, 

154) -> None: # pragma: nocover 

155 """Pretty-print a core schema using the `rich` library. 

156 

157 Args: 

158 val: The core schema to print, or a Pydantic model/dataclass/type adapter 

159 (in which case the cached core schema is fetched and printed). 

160 console: A rich console to use when printing. Defaults to the global rich console instance. 

161 max_depth: The number of nesting levels which may be printed. 

162 strip_metadata: Whether to strip metadata in the output. If `True` any known core metadata 

163 attributes will be stripped (but custom attributes are kept). Defaults to `True`. 

164 """ 

165 # lazy import: 

166 from rich.pretty import pprint 

167 

168 # circ. imports: 

169 from pydantic import BaseModel, TypeAdapter 

170 from pydantic.dataclasses import is_pydantic_dataclass 

171 

172 if (inspect.isclass(val) and issubclass(val, BaseModel)) or is_pydantic_dataclass(val): 

173 val = val.__pydantic_core_schema__ 

174 if isinstance(val, TypeAdapter): 

175 val = val.core_schema 

176 cleaned_schema = _clean_schema_for_pretty_print(val, strip_metadata=strip_metadata) 

177 

178 pprint(cleaned_schema, console=console, max_depth=max_depth) 

179 

180 

181pps = pretty_print_core_schema 1abcdefghijklmnopqrstuvGHIJKLMwxyzABCDEF