Coverage for an_website/utils/elastic_transport_async_http_node.py: 32.692%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13"""Async elasticsearch.""" 

14 

15import logging 

16from typing import Final 

17 

18from elastic_transport import ( 

19 ApiResponse, 

20 ApiResponseMeta, 

21 BaseAsyncNode, 

22 ConnectionError as ESConnectionError, 

23 ConnectionTimeout, 

24 HttpHeaders, 

25 NodeConfig, 

26 TlsError, 

27 TransportError, 

28) 

29from elastic_transport.client_utils import DefaultType 

30from tornado.httpclient import AsyncHTTPClient, HTTPClientError, HTTPRequest 

31 

32try: 

33 import pycurl 

34 

35 # pylint: disable-next=ungrouped-imports 

36 from tornado.curl_httpclient import CurlError 

37except ImportError: 

38 CurlError = None # type: ignore[misc, assignment] 

39 pycurl = None # type: ignore[assignment] 

40 

41 

42LOGGER: Final = logging.getLogger(__name__) 

43 

44 

45class TornadoConnectionError(ESConnectionError): 

46 """An error occured while connecting.""" 

47 

48 __str__ = TransportError.__str__ 

49 

50 

51class TornadoAsyncNode(BaseAsyncNode): 

52 """A node that performs requests.""" 

53 

54 __client: AsyncHTTPClient | None 

55 

56 def __init__(self, config: NodeConfig) -> None: 

57 """Initialise self.""" 

58 super().__init__(config) 

59 self.__client = None 

60 

61 @property 

62 def _client(self) -> AsyncHTTPClient: 

63 """Get the AsyncHTTPClient.""" 

64 if self.__client is None: 

65 self.__client = AsyncHTTPClient( 

66 force_instance=True, 

67 defaults={ 

68 "ca_certs": self.config.ca_certs, 

69 "client_cert": self.config.client_cert, 

70 "client_key": self.config.client_key, 

71 "ssl_options": self.config.ssl_context, 

72 "use_gzip": self._http_compress, 

73 "validate_cert": self.config.verify_certs, 

74 }, 

75 ) 

76 return self.__client 

77 

78 async def close(self) -> None: # type: ignore[override] 

79 """Close the connection.""" 

80 if self.__client: 

81 self.__client.close() 

82 self.__client = None 

83 

84 async def perform_request( # type: ignore[override] 

85 self, 

86 method: str, 

87 target: str, 

88 body: bytes | None = None, 

89 headers: HttpHeaders | None = None, 

90 request_timeout: float | DefaultType | None = None, 

91 ) -> ApiResponse[bytes]: 

92 """Perform a request.""" 

93 if isinstance(request_timeout, DefaultType): 

94 request_timeout = None 

95 url = self.base_url + target 

96 

97 request = HTTPRequest( 

98 url=url, 

99 method=method, 

100 body=body, 

101 headers={ 

102 **self._headers, 

103 **(headers or {}), 

104 }, 

105 request_timeout=request_timeout, 

106 ) 

107 

108 try: 

109 response = await self._client.fetch( 

110 request, 

111 raise_error=False, 

112 ) 

113 except HTTPClientError as e: 

114 err: TransportError | None = None 

115 if CurlError is not None and isinstance(e, CurlError): 

116 assert pycurl is not None, "pycurl is present if CurlError is" 

117 curl_errno: int = e.errno # pylint: disable=no-member 

118 if curl_errno == pycurl.E_COULDNT_CONNECT: 

119 err = ESConnectionError("Could not connect to server") 

120 elif curl_errno == pycurl.E_OPERATION_TIMEOUTED: 

121 err = ConnectionTimeout("Timeout was reached") 

122 elif curl_errno == pycurl.E_SSL_CONNECT_ERROR: 

123 err = TlsError("SSL connect error") 

124 elif e.code == 599: 

125 # TODO: maybe use curl_easy_strerror to get a string 

126 # added in pycurl 17ecb45612b99232cc0908ffa535f960eb485800 

127 err = TornadoConnectionError(f"{e.message} ({curl_errno})") 

128 err = err or TornadoConnectionError(str(e)) 

129 err.errors = (e,) 

130 raise err from e 

131 

132 LOGGER.debug( 

133 "%s %s [status:%s duration:%fs]", 

134 method, 

135 url, 

136 response.code, 

137 response.request_time, 

138 ) 

139 

140 return ApiResponse( 

141 meta=ApiResponseMeta( 

142 status=response.code, 

143 duration=( 

144 -1.0 

145 if response.request_time is None 

146 else response.request_time 

147 ), 

148 headers=HttpHeaders(response.headers), 

149 node=self._config, 

150 http_version="1.1", 

151 ), 

152 body=response.body, 

153 )