Coverage for faststream / _internal / fastapi / get_dependant.py: 92%

1import inspect (empty)

2from collections.abc import Callable, Iterable (empty)

3from typing import TYPE_CHECKING, Annotated, Any, cast, get_args, get_origin (empty)

4 

5from fast_depends.library.serializer import OptionItem (empty)

6from fast_depends.utils import get_typed_annotation (empty)

7from fastapi import params (empty)

8from fastapi.dependencies.utils import ( (empty)

9 get_dependant, 

10 get_parameterless_sub_dependant, 

11 get_typed_signature, 

12) 

13 

14from faststream._internal._compat import PYDANTIC_V2 (empty)

15 

16if TYPE_CHECKING: (empty)

17 from fastapi.dependencies import models 

18 

19 

20def get_fastapi_dependant( (empty)

21 orig_call: Callable[..., Any], (empty)

22 dependencies: Iterable["params.Depends"], (empty)

23) -> Any: (empty)

24 """Generate FastStream-Compatible FastAPI Dependant object.""" 

25 dependent = get_fastapi_native_dependant( (empty)

26 orig_call=orig_call, (empty)

27 dependencies=dependencies, (empty)

28 ) 

29 

30 return _patch_fastapi_dependent(dependent) (empty)

31 

32 

33def get_fastapi_native_dependant( (empty)

34 orig_call: Callable[..., Any], (empty)

35 dependencies: Iterable["params.Depends"], (empty)

36) -> Any: (empty)

37 """Generate native FastAPI Dependant.""" 

38 dependent = get_dependant( (empty)

39 path="", (empty)

40 call=orig_call, (empty)

41 ) 

42 

43 for depends in list(dependencies)[::-1]: (empty)

44 dependent.dependencies.insert( (empty)

45 0, (empty)

46 get_parameterless_sub_dependant(depends=depends, path=""), (empty)

47 ) 

48 

49 return dependent (empty)

50 

51 

52def _patch_fastapi_dependent(dependant: "models.Dependant") -> "models.Dependant": (empty)

53 """Patch FastAPI by adding fields for AsyncAPI schema generation.""" 

54 from pydantic import Field, create_model # FastAPI always has pydantic (empty)

55 

56 from faststream._internal._compat import PydanticUndefined (empty)

57 

58 params = dependant.query_params + dependant.body_params (empty)

59 

60 for d in dependant.dependencies: (empty)

61 params.extend(d.query_params + d.body_params) (empty)

62 

63 params_unique = {} (empty)

64 

65 call = dependant.call (empty)

66 if is_faststream_decorated(call): (empty)

67 call = getattr(call, "__wrapped__", call) (empty)

68 globalns = getattr(call, "__globals__", {}) (empty)

69 

70 for p in params: (empty)

71 if p.name not in params_unique: (empty)

72 info: Any = p.field_info if PYDANTIC_V2 else p (empty)

73 

74 field_data = { (empty)

75 "default": ... if info.default is PydanticUndefined else info.default, 

76 "default_factory": info.default_factory, 

77 "alias": info.alias, 

78 } 

79 

80 if PYDANTIC_V2: (empty)

81 from pydantic.fields import FieldInfo (empty)

82 

83 info = cast("FieldInfo", info) (empty)

84 

85 field_data.update( (empty)

86 { 

87 "title": info.title, 

88 "alias_priority": info.alias_priority, 

89 "validation_alias": info.validation_alias, 

90 "serialization_alias": info.serialization_alias, 

91 "description": info.description, 

92 "discriminator": info.discriminator, 

93 "examples": info.examples, 

94 "exclude": info.exclude, 

95 "json_schema_extra": info.json_schema_extra, 

96 }, 

97 ) 

98 

99 f = next( (empty)

100 filter( 

101 lambda x: isinstance(x, FieldInfo), 

102 p.field_info.metadata or (), 

103 ), 

104 Field(**field_data), # type: ignore[pydantic-field,unused-ignore] 

105 ) 

106 

107 else: 

108 from pydantic.fields import ModelField # type: ignore[attr-defined] 

109 

110 info = cast("ModelField", info) 

111 

112 field_data.update( 

113 { 

114 "title": info.field_info.title, 

115 "description": info.field_info.description, 

116 "discriminator": info.field_info.discriminator, 

117 "exclude": info.field_info.exclude, 

118 "gt": info.field_info.gt, 

119 "ge": info.field_info.ge, 

120 "lt": info.field_info.lt, 

121 "le": info.field_info.le, 

122 }, 

123 ) 

124 f = Field(**field_data) # type: ignore[pydantic-field,unused-ignore] 

125 

126 params_unique[p.name] = ( (empty)

127 get_typed_annotation(info.annotation, globalns, {}), 

128 f, 

129 ) 

130 

131 dependant.model = create_model( # type: ignore[attr-defined] (empty)

132 getattr(call, "__name__", type(call).__name__), 

133 ) 

134 

135 dependant.custom_fields = {} # type: ignore[attr-defined] (empty)

136 dependant.flat_params = [ # type: ignore[attr-defined] (empty)

137 OptionItem(field_name=name, field_type=type_, default_value=default) 

138 for name, (type_, default) in params_unique.items() 

139 ] 

140 

141 return dependant (empty)

142 

143 

144def has_forbidden_types( (empty)

145 orig_call: Callable[..., Any], (empty)

146 forbidden_types: tuple[Any, ...], (empty)

147) -> set[Any]: (empty)

148 """Check if faststream.Depends is used in the handler.""" 

149 endpoint_signature = get_typed_signature(orig_call) (empty)

150 signature_params = endpoint_signature.parameters (empty)

151 

152 founded_types = set() (empty)

153 

154 for param in signature_params.values(): (empty)

155 ann = param.annotation (empty)

156 

157 founded_buffer = set() (empty)

158 has_fastapi_depends = False (empty)

159 if ann is not inspect.Signature.empty and get_origin(ann) is Annotated: (empty)

160 annotated_args = get_args(ann) (empty)

161 

162 for arg in annotated_args[1:]: (empty)

163 if isinstance(arg, params.Depends): (empty)

164 has_fastapi_depends = True (empty)

165 continue (empty)

166 

167 for t in forbidden_types: (empty)

168 if isinstance(arg, t): (empty)

169 founded_buffer.add(t) (empty)

170 

171 if isinstance(param.default, params.Depends): (empty)

172 has_fastapi_depends = True (empty)

173 continue (empty)

174 

175 for t in forbidden_types: (empty)

176 if isinstance(param.default, t): (empty)

177 founded_buffer.add(t) (empty)

178 

179 if not has_fastapi_depends: (empty)

180 founded_types |= founded_buffer (empty)

181 

182 return founded_types (empty)

183 

184 

185FASTSTREAM_FASTAPI_PLUGIN_DECORATOR_MARKER = "__faststream_consumer__" (empty)

186 

187 

188def is_faststream_decorated(func: object) -> bool: (empty)

189 return getattr(func, FASTSTREAM_FASTAPI_PLUGIN_DECORATOR_MARKER, False) (empty)

190 

191 

192def mark_faststream_decorated(func: object) -> None: (empty)

193 setattr(func, FASTSTREAM_FASTAPI_PLUGIN_DECORATOR_MARKER, True) (empty)