Coverage for faststream / _internal / _compat.py: 60%

1import json (empty)

2import sys (empty)

3import warnings (empty)

4from collections import UserString (empty)

5from collections.abc import Callable, Iterable, Mapping (empty)

6from importlib.util import find_spec (empty)

7from typing import ( (empty)

8 Any, 

9 TypeVar, 

10) 

11 

12from pydantic import BaseModel (empty)

13from pydantic.version import VERSION as PYDANTIC_VERSION (empty)

14 

15IS_WINDOWS = sys.platform in {"win32", "cygwin", "msys"} (empty)

16IS_MACOS = sys.platform == "darwin" (empty)

17 

18__all__ = ( (empty)

19 "HAS_TYPER", 

20 "PYDANTIC_V2", 

21 "BaseModel", 

22 "CoreSchema", 

23 "EmailStr", 

24 "ExceptionGroup", 

25 "GetJsonSchemaHandler", 

26 "PydanticUndefined", 

27 "json_dumps", 

28 "json_loads", 

29 "with_info_plain_validator_function", 

30) 

31 

32try: (empty)

33 HAS_TYPER = find_spec("typer") is not None (empty)

34except ImportError: 

35 HAS_TYPER = False 

36 

37 

38json_dumps: Callable[..., bytes] (empty)

39orjson: Any (empty)

40 

41try: (empty)

42 import orjson # type: ignore[no-redef] (empty)

43except ImportError: (empty)

44 orjson = None (empty)

45 

46if orjson: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true(empty)

47 json_loads = orjson.loads 

48 json_dumps = orjson.dumps 

49else: 

50 json_loads = json.loads (empty)

51 

52 def json_dumps(*a: Any, **kw: Any) -> bytes: (empty)

53 return json.dumps(*a, **kw).encode() (empty)

54 

55 

56ModelVar = TypeVar("ModelVar", bound=BaseModel) (empty)

57 

58JsonSchemaValue = Mapping[str, Any] (empty)

59major, minor, *_ = PYDANTIC_VERSION.split(".") (empty)

60_PYDANTCI_MAJOR, _PYDANTIC_MINOR = int(major), int(minor) (empty)

61 

62PYDANTIC_V2 = _PYDANTCI_MAJOR >= 2 (empty)

63 

64if PYDANTIC_V2: (empty)

65 if _PYDANTIC_MINOR >= 4: (empty)

66 from pydantic.annotated_handlers import ( (empty)

67 GetJsonSchemaHandler, 

68 ) 

69 from pydantic_core.core_schema import ( (empty)

70 with_info_plain_validator_function, 

71 ) 

72 else: 

73 from pydantic._internal._annotated_handlers import ( # type: ignore[no-redef] 

74 GetJsonSchemaHandler, 

75 ) 

76 from pydantic_core.core_schema import ( 

77 general_plain_validator_function as with_info_plain_validator_function, 

78 ) 

79 

80 from pydantic_core import CoreSchema, PydanticUndefined, to_jsonable_python (empty)

81 

82 SCHEMA_FIELD = "json_schema_extra" (empty)

83 DEF_KEY = "$defs" (empty)

84 

85 def model_to_jsonable( (empty)

86 model: BaseModel, 

87 **kwargs: Any, 

88 ) -> Any: 

89 return to_jsonable_python(model, **kwargs) (empty)

90 

91 def dump_json(data: Any) -> bytes: (empty)

92 return json_dumps(model_to_jsonable(data)) (empty)

93 

94 def get_model_fields(model: type[BaseModel]) -> dict[str, Any]: (empty)

95 return model.model_fields (empty)

96 

97 def model_to_json(model: BaseModel, **kwargs: Any) -> str: (empty)

98 return model.model_dump_json(**kwargs) (empty)

99 

100 def model_parse( (empty)

101 model: type[ModelVar], 

102 data: str | bytes, 

103 **kwargs: Any, 

104 ) -> ModelVar: 

105 return model.model_validate_json(data, **kwargs) 

106 

107 def model_schema(model: type[BaseModel], **kwargs: Any) -> dict[str, Any]: (empty)

108 return model.model_json_schema(**kwargs) (empty)

109 

110else: 

111 from pydantic.json import pydantic_encoder 

112 

113 GetJsonSchemaHandler = Any # type: ignore[assignment,misc] 

114 CoreSchema = Any # type: ignore[assignment,misc] 

115 

116 SCHEMA_FIELD = "schema_extra" 

117 DEF_KEY = "definitions" 

118 

119 PydanticUndefined = Ellipsis # type: ignore[assignment] 

120 

121 def dump_json(data: Any) -> bytes: 

122 return json_dumps(data, default=pydantic_encoder) 

123 

124 def get_model_fields(model: type[BaseModel]) -> dict[str, Any]: 

125 return model.__fields__ # type: ignore[return-value] 

126 

127 def model_to_json(model: BaseModel, **kwargs: Any) -> str: 

128 return model.json(**kwargs) 

129 

130 def model_parse( 

131 model: type[ModelVar], 

132 data: str | bytes, 

133 **kwargs: Any, 

134 ) -> ModelVar: 

135 return model.parse_raw(data, **kwargs) 

136 

137 def model_schema(model: type[BaseModel], **kwargs: Any) -> dict[str, Any]: 

138 return model.schema(**kwargs) 

139 

140 def model_to_jsonable( 

141 model: BaseModel, 

142 **kwargs: Any, 

143 ) -> Any: 

144 return json_loads(model.json(**kwargs)) 

145 

146 # TODO: pydantic types misc 

147 def with_info_plain_validator_function( # type: ignore[misc] 

148 function: Callable[..., Any], 

149 *, 

150 ref: str | None = None, 

151 metadata: Any = None, 

152 serialization: Any = None, 

153 ) -> JsonSchemaValue: 

154 return {} 

155 

156 

157if sys.version_info >= (3, 11): (empty)

158 ExceptionGroup = ExceptionGroup # noqa: F821,PLW0127 (empty)

159else: 

160 from exceptiongroup import ExceptionGroup (empty)

161 

162try: (empty)

163 import email_validator (empty)

164 

165 if email_validator is None: (empty)

166 raise ImportError 

167 from pydantic import EmailStr (empty)

168except ImportError: # pragma: no cover 

169 # NOTE: EmailStr mock was copied from the FastAPI 

170 # https://github.com/tiangolo/fastapi/blob/master/fastapi/openapi/models.py#24 

171 class EmailStr(UserString): # type: ignore[no-redef] 

172 """EmailStr is a string that should be an email. 

173 

174 Note: EmailStr mock was copied from the FastAPI: 

175 https://github.com/tiangolo/fastapi/blob/master/fastapi/openapi/models.py#24 

176 """ 

177 

178 @classmethod 

179 def __get_validators__(cls) -> Iterable[Callable[..., Any]]: 

180 """Returns the validators for the EmailStr class.""" 

181 yield cls.validate 

182 

183 @classmethod 

184 def validate(cls, v: Any) -> str: 

185 """Validates the EmailStr class.""" 

186 warnings.warn( 

187 "email-validator not installed, email fields will be treated as str.\n" 

188 "To install, run: pip install email-validator", 

189 category=RuntimeWarning, 

190 stacklevel=1, 

191 ) 

192 return str(v) 

193 

194 @classmethod 

195 def _validate(cls, __input_value: Any, _: Any) -> str: 

196 warnings.warn( 

197 "email-validator not installed, email fields will be treated as str.\n" 

198 "To install, run: pip install email-validator", 

199 category=RuntimeWarning, 

200 stacklevel=1, 

201 ) 

202 return str(__input_value) 

203 

204 @classmethod 

205 def __get_pydantic_json_schema__( 

206 cls, 

207 core_schema: CoreSchema, 

208 handler: GetJsonSchemaHandler, 

209 ) -> JsonSchemaValue: 

210 """Returns the JSON schema for the EmailStr class. 

211 

212 Args: 

213 core_schema : the core schema 

214 handler : the handler 

215 """ 

216 return {"type": "string", "format": "email"} 

217 

218 @classmethod 

219 def __get_pydantic_core_schema__( 

220 cls, 

221 source: type[Any], 

222 handler: Callable[[Any], CoreSchema], 

223 ) -> JsonSchemaValue: 

224 """Returns the core schema for the EmailStr class. 

225 

226 Args: 

227 source : the source 

228 handler : the handler 

229 """ 

230 return with_info_plain_validator_function(cls._validate) 

231 

232 

233uvicorn: Any (empty)

234 

235try: (empty)

236 import uvicorn (empty)

237 

238 HAS_UVICORN = True (empty)

239 

240except ImportError: 

241 uvicorn = None 

242 HAS_UVICORN = False