Coverage for faststream / asgi / handlers.py: 95%

1import logging (empty)

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

3from typing import TYPE_CHECKING, Any, Optional, Union, overload (empty)

4 

5from fast_depends.exceptions import ValidationError as FDValidationError (empty)

6 

7from faststream import apply_types (empty)

8from faststream._internal.di.config import FastDependsConfig (empty)

9from faststream._internal.utils.functions import to_async (empty)

10 

11from .request import AsgiRequest (empty)

12from .response import AsgiResponse (empty)

13 

14if TYPE_CHECKING: (empty)

15 from faststream._internal.basic_types import LoggerProto 

16 from faststream.specification.schema import Tag, TagDict 

17 

18 from .types import ASGIApp, Receive, Scope, Send, UserApp 

19 

20 

21class HttpHandler: (empty)

22 def __init__( (empty)

23 self, 

24 func: "UserApp", 

25 *, 

26 include_in_schema: bool = True, 

27 description: str | None = None, 

28 methods: Sequence[str] | None = None, 

29 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, 

30 unique_id: str | None = None, 

31 fd_config: FastDependsConfig | None = None, 

32 ) -> None: 

33 self.__original_func = func (empty)

34 self.func = func (empty)

35 self.methods = methods or () (empty)

36 self.include_in_schema = include_in_schema (empty)

37 self.description = description or func.__doc__ (empty)

38 self.tags = tags (empty)

39 self.unique_id = unique_id (empty)

40 self.fd_config = fd_config or FastDependsConfig() (empty)

41 self.logger: LoggerProto | None = None (empty)

42 

43 async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None: (empty)

44 if scope["method"] not in self.methods: (empty)

45 response: ASGIApp = _get_method_not_allowed_response(self.methods) 

46 

47 else: 

48 with ( (empty)

49 self.fd_config.context.scope( 

50 "request", AsgiRequest(scope, receive, send) 

51 ), 

52 self.fd_config.context.scope( 

53 "logger", 

54 self.logger, 

55 ), 

56 ): 

57 try: (empty)

58 response = await self.func(scope) (empty)

59 except FDValidationError: (empty)

60 if self.logger is not None: (empty)

61 message = "Validation error" (empty)

62 self.logger.log(logging.ERROR, message, exc_info=True) (empty)

63 response = AsgiResponse( (empty)

64 body=b"Validation error", 

65 status_code=422, 

66 ) 

67 except Exception: 

68 if self.logger is not None: 

69 self.logger.log( 

70 logging.ERROR, 

71 "Exception occurred while processing request", 

72 exc_info=True, 

73 ) 

74 response = AsgiResponse( 

75 body=b"Internal Server Error", status_code=500 

76 ) 

77 await response(scope, receive, send) (empty)

78 

79 def update_fd_config(self, config: FastDependsConfig) -> None: (empty)

80 self.fd_config = config | self.fd_config (empty)

81 self.func = apply_types( (empty)

82 to_async(self.__original_func), context__=self.fd_config.context 

83 ) 

84 

85 def set_logger(self, logger: "LoggerProto | None") -> None: (empty)

86 self.logger = logger (empty)

87 

88 

89class GetHandler(HttpHandler): (empty)

90 def __init__( (empty)

91 self, 

92 func: "UserApp", 

93 *, 

94 include_in_schema: bool = True, 

95 description: str | None = None, 

96 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, 

97 unique_id: str | None = None, 

98 ) -> None: 

99 super().__init__( (empty)

100 func, 

101 include_in_schema=include_in_schema, 

102 description=description, 

103 methods=("GET", "HEAD"), 

104 tags=tags, 

105 unique_id=unique_id, 

106 ) 

107 

108 

109@overload (empty)

110def get( (empty)

111 func: "UserApp", (empty)

112 *, 

113 include_in_schema: bool = True, (empty)

114 description: str | None = None, (empty)

115 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, (empty)

116 unique_id: str | None = None, (empty)

117) -> "GetHandler": ... (empty)

118 

119 

120@overload (empty)

121def get( (empty)

122 func: None = None, (empty)

123 *, 

124 include_in_schema: bool = True, (empty)

125 description: str | None = None, (empty)

126 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, (empty)

127 unique_id: str | None = None, (empty)

128) -> Callable[["UserApp"], "GetHandler"]: ... (empty)

129 

130 

131def get( (empty)

132 func: Optional["UserApp"] = None, 

133 *, 

134 include_in_schema: bool = True, 

135 description: str | None = None, 

136 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, 

137 unique_id: str | None = None, 

138) -> Union[Callable[["UserApp"], "GetHandler"], "GetHandler"]: 

139 def decorator(inner_func: "UserApp") -> "GetHandler": (empty)

140 return GetHandler( (empty)

141 inner_func, 

142 include_in_schema=include_in_schema, 

143 description=description, 

144 tags=tags, 

145 unique_id=unique_id, 

146 ) 

147 

148 if func is None: (empty)

149 return decorator (empty)

150 

151 return decorator(func) (empty)

152 

153 

154class PostHandler(HttpHandler): (empty)

155 def __init__( (empty)

156 self, 

157 func: "UserApp", 

158 *, 

159 include_in_schema: bool = True, 

160 description: str | None = None, 

161 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, 

162 unique_id: str | None = None, 

163 ) -> None: 

164 super().__init__( (empty)

165 func, 

166 include_in_schema=include_in_schema, 

167 description=description, 

168 methods=("POST", "HEAD"), 

169 tags=tags, 

170 unique_id=unique_id, 

171 ) 

172 

173 

174@overload (empty)

175def post( (empty)

176 func: "UserApp", (empty)

177 *, 

178 include_in_schema: bool = True, (empty)

179 description: str | None = None, (empty)

180 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, (empty)

181 unique_id: str | None = None, (empty)

182) -> "PostHandler": ... (empty)

183 

184 

185@overload (empty)

186def post( (empty)

187 func: None = None, (empty)

188 *, 

189 include_in_schema: bool = True, (empty)

190 description: str | None = None, (empty)

191 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, (empty)

192 unique_id: str | None = None, (empty)

193) -> Callable[["UserApp"], "PostHandler"]: ... (empty)

194 

195 

196def post( (empty)

197 func: Optional["UserApp"] = None, 

198 *, 

199 include_in_schema: bool = True, 

200 description: str | None = None, 

201 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, 

202 unique_id: str | None = None, 

203) -> Union[Callable[["UserApp"], "PostHandler"], "PostHandler"]: 

204 def decorator(inner_func: "UserApp") -> "PostHandler": (empty)

205 return PostHandler( (empty)

206 inner_func, 

207 include_in_schema=include_in_schema, 

208 description=description, 

209 tags=tags, 

210 unique_id=unique_id, 

211 ) 

212 

213 if func is None: (empty)

214 return decorator (empty)

215 

216 return decorator(func) (empty)

217 

218 

219def _get_method_not_allowed_response(methods: Sequence[str]) -> AsgiResponse: (empty)

220 return AsgiResponse( 

221 body=b"Method Not Allowed", 

222 status_code=405, 

223 headers={ 

224 "Allow": ", ".join(methods), 

225 }, 

226 )