Coverage for faststream / _internal / fastapi / _compat.py: 66%

1from dataclasses import dataclass (empty)

2from typing import TYPE_CHECKING, Any (empty)

3 

4from fastapi import __version__ as FASTAPI_VERSION # noqa: N812 (empty)

5from fastapi.dependencies.utils import solve_dependencies (empty)

6from starlette.background import BackgroundTasks (empty)

7from typing_extensions import Never (empty)

8 

9if TYPE_CHECKING: (empty)

10 from fastapi.dependencies.models import Dependant 

11 from fastapi.requests import Request 

12 

13major, minor, patch, *_ = FASTAPI_VERSION.split(".") (empty)

14 

15_FASTAPI_MAJOR, _FASTAPI_MINOR = int(major), int(minor) (empty)

16 

17FASTAPI_V2 = _FASTAPI_MAJOR > 0 or _FASTAPI_MINOR > 100 (empty)

18FASTAPI_V106 = _FASTAPI_MAJOR > 0 or _FASTAPI_MINOR >= 106 (empty)

19FASTAPI_V121 = _FASTAPI_MAJOR > 0 or _FASTAPI_MINOR >= 121 (empty)

20FASTAPI_V128 = _FASTAPI_MAJOR > 0 or _FASTAPI_MINOR >= 128 (empty)

21 

22try: (empty)

23 _FASTAPI_PATCH = int(patch) (empty)

24except ValueError: 

25 FASTAPI_v102_3 = True 

26 FASTAPI_v102_4 = True 

27else: 

28 FASTAPI_v102_3 = ( (empty)

29 _FASTAPI_MAJOR > 0 

30 or _FASTAPI_MINOR > 112 

31 or (_FASTAPI_MINOR == 112 and _FASTAPI_PATCH > 2) 

32 ) 

33 FASTAPI_v102_4 = ( (empty)

34 _FASTAPI_MAJOR > 0 

35 or _FASTAPI_MINOR > 112 

36 or (_FASTAPI_MINOR == 112 and _FASTAPI_PATCH > 3) 

37 ) 

38 

39__all__ = ( (empty)

40 "RequestValidationError", 

41 "create_response_field", 

42 "raise_fastapi_validation_error", 

43 "solve_faststream_dependency", 

44) 

45 

46 

47@dataclass (empty)

48class SolvedDependency: (empty)

49 values: dict[str, Any] (empty)

50 errors: list[Any] (empty)

51 background_tasks: BackgroundTasks | None (empty)

52 

53 

54if FASTAPI_V128: (empty)

55 from fastapi.exceptions import RequestValidationError (empty)

56 

57 def raise_fastapi_validation_error(errors: list[Any], body: dict[str, Any]) -> Never: (empty)

58 raise RequestValidationError(errors, body=body) (empty)

59 

60elif FASTAPI_V2: 

61 from fastapi._compat import _normalize_errors # type: ignore[attr-defined] 

62 from fastapi.exceptions import RequestValidationError 

63 

64 def raise_fastapi_validation_error(errors: list[Any], body: dict[str, Any]) -> Never: 

65 raise RequestValidationError(_normalize_errors(errors), body=body) 

66 

67else: 

68 from pydantic import ( # type: ignore[assignment] 

69 ValidationError as RequestValidationError, 

70 create_model, 

71 ) 

72 

73 ROUTER_VALIDATION_ERROR_MODEL = create_model("StreamRoute") 

74 

75 def raise_fastapi_validation_error(errors: list[Any], body: dict[str, Any]) -> Never: 

76 raise RequestValidationError(errors, ROUTER_VALIDATION_ERROR_MODEL) # type: ignore[misc] 

77 

78 

79if FASTAPI_v102_3: (empty)

80 from fastapi.utils import ( (empty)

81 create_model_field as create_response_field, 

82 ) 

83 

84 extra = {"embed_body_fields": False} if FASTAPI_v102_4 else {} (empty)

85 

86 async def solve_faststream_dependency( (empty)

87 request: "Request", 

88 dependant: "Dependant", 

89 dependency_overrides_provider: Any | None, 

90 **kwargs: Any, 

91 ) -> SolvedDependency: 

92 solved_result = await solve_dependencies( (empty)

93 request=request, 

94 body=request._body, # type: ignore[arg-type] 

95 dependant=dependant, 

96 dependency_overrides_provider=dependency_overrides_provider, 

97 **extra, # type: ignore[arg-type] 

98 **kwargs, 

99 ) 

100 values, errors, background = ( (empty)

101 solved_result.values, 

102 solved_result.errors, 

103 solved_result.background_tasks, 

104 ) 

105 

106 return SolvedDependency( (empty)

107 values=values, 

108 errors=errors, 

109 background_tasks=background, 

110 ) 

111 

112else: 

113 from fastapi.utils import ( # type: ignore[attr-defined,no-redef] 

114 create_response_field, 

115 ) 

116 

117 async def solve_faststream_dependency( 

118 request: "Request", 

119 dependant: "Dependant", 

120 dependency_overrides_provider: Any | None, 

121 **kwargs: Any, 

122 ) -> SolvedDependency: 

123 solved_result = await solve_dependencies( 

124 request=request, 

125 body=request._body, # type: ignore[arg-type] 

126 dependant=dependant, 

127 dependency_overrides_provider=dependency_overrides_provider, 

128 **kwargs, 

129 ) 

130 

131 ( 

132 values, 

133 errors, 

134 background, 

135 _response, 

136 _dependency_cache, 

137 ) = solved_result # type: ignore[misc] 

138 

139 return SolvedDependency( 

140 values=values, # type: ignore[has-type] 

141 errors=errors, # type: ignore[has-type] 

142 background_tasks=background, # type: ignore[has-type] 

143 )