Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/web_server.py: 51%

1"""Low level HTTP server.""" 

2 

3import asyncio 

4import warnings 

5from typing import ( 

6 Any, 

7 Awaitable, 

8 Callable, 

9 Dict, 

10 Generic, 

11 List, 

12 Optional, 

13 TypeVar, 

14 overload, 

15) 

16 

17from .abc import AbstractStreamWriter 

18from .http_parser import RawRequestMessage 

19from .streams import StreamReader 

20from .web_protocol import RequestHandler 

21from .web_request import BaseRequest 

22from .web_response import StreamResponse 

23 

24__all__ = ("Server",) 

25 

26_Request = TypeVar("_Request", bound=BaseRequest) 

27_RequestFactory = Callable[ 

28 [ 

29 RawRequestMessage, 

30 StreamReader, 

31 "RequestHandler[_Request]", 

32 AbstractStreamWriter, 

33 "asyncio.Task[None]", 

34 ], 

35 _Request, 

36] 

37 

38 

39class Server(Generic[_Request]): 

40 request_factory: _RequestFactory[_Request] 

41 

42 @overload 

43 def __init__( # type: ignore[misc] 

44 self: "Server[BaseRequest]", 

45 handler: Callable[[_Request], Awaitable[StreamResponse]], 

46 *, 

47 debug: Optional[bool] = None, 

48 handler_cancellation: bool = False, 

49 **kwargs: Any, # TODO(PY311): Use Unpack to define kwargs from RequestHandler 

50 ) -> None: ... 

51 @overload 

52 def __init__( # type: ignore[misc] 

53 self, 

54 handler: Callable[[_Request], Awaitable[StreamResponse]], 

55 *, 

56 request_factory: Optional[_RequestFactory[_Request]], 

57 debug: Optional[bool] = None, 

58 handler_cancellation: bool = False, 

59 **kwargs: Any, 

60 ) -> None: ... 

61 def __init__( 

62 self, 

63 handler: Callable[[_Request], Awaitable[StreamResponse]], 

64 *, 

65 request_factory: Optional[_RequestFactory[_Request]] = None, 

66 debug: Optional[bool] = None, 

67 handler_cancellation: bool = False, 

68 **kwargs: Any, 

69 ) -> None: 

70 if debug is not None: 

71 warnings.warn( 

72 "debug argument is no-op since 4.0 and scheduled for removal in 5.0", 

73 DeprecationWarning, 

74 stacklevel=2, 

75 ) 

76 self._loop = asyncio.get_running_loop() 

77 self._connections: Dict[RequestHandler[_Request], asyncio.Transport] = {} 

78 self._kwargs = kwargs 

79 # requests_count is the number of requests being processed by the server 

80 # for the lifetime of the server. 

81 self.requests_count = 0 

82 self.request_handler = handler 

83 self.request_factory = request_factory or self._make_request # type: ignore[assignment] 

84 self.handler_cancellation = handler_cancellation 

85 

86 @property 

87 def connections(self) -> List[RequestHandler[_Request]]: 

88 return list(self._connections.keys()) 

89 

90 def connection_made( 

91 self, handler: RequestHandler[_Request], transport: asyncio.Transport 

92 ) -> None: 

93 self._connections[handler] = transport 

94 

95 def connection_lost( 

96 self, handler: RequestHandler[_Request], exc: Optional[BaseException] = None 

97 ) -> None: 

98 if handler in self._connections: 

99 if handler._task_handler: 

100 handler._task_handler.add_done_callback( 

101 lambda f: self._connections.pop(handler, None) 

102 ) 

103 else: 

104 del self._connections[handler] 

105 

106 def _make_request( 

107 self, 

108 message: RawRequestMessage, 

109 payload: StreamReader, 

110 protocol: RequestHandler[BaseRequest], 

111 writer: AbstractStreamWriter, 

112 task: "asyncio.Task[None]", 

113 ) -> BaseRequest: 

114 return BaseRequest(message, payload, protocol, writer, task, self._loop) 

115 

116 def pre_shutdown(self) -> None: 

117 for conn in self._connections: 

118 conn.close() 

119 

120 async def shutdown(self, timeout: Optional[float] = None) -> None: 

121 coros = (conn.shutdown(timeout) for conn in self._connections) 

122 await asyncio.gather(*coros) 

123 self._connections.clear() 

124 

125 def __call__(self) -> RequestHandler[_Request]: 

126 try: 

127 return RequestHandler(self, loop=self._loop, **self._kwargs) 

128 except TypeError: 

129 # Failsafe creation: remove all custom handler_args 

130 kwargs = { 

131 k: v 

132 for k, v in self._kwargs.items() 

133 if k in ["debug", "access_log_class"] 

134 } 

135 return RequestHandler(self, loop=self._loop, **kwargs)