Coverage for filip/clients/base_http_client.py: 85%

1""" 

2Base http client module 

3""" 

4import logging 

5from pydantic import AnyHttpUrl 

6from typing import Dict, ByteString, List, IO, Tuple, Union 

7import requests 

8 

9from filip.models.base import FiwareHeader 

10from filip.utils import validate_http_url 

11 

12 

13class BaseHttpClient: 

14 """ 

15 Base client for all derived api-clients. 

16 

17 Args: 

18 session: request session object. This is required for reusing 

19 the same connection 

20 reuse_session (bool): 

21 fiware_header: Fiware header object required for multi tenancy 

22 **kwargs: Optional arguments that ``request`` takes. 

23 

24 """ 

25 def __init__(self, 

26 url: Union[AnyHttpUrl, str] = None, 

27 *, 

28 session: requests.Session = None, 

29 fiware_header: Union[Dict, FiwareHeader] = None, 

30 **kwargs): 

31 

32 self.logger = logging.getLogger( 

33 name=f"{self.__class__.__name__}") 

34 self.logger.addHandler(logging.NullHandler()) 

35 self.logger.debug("Creating %s", self.__class__.__name__) 

36 

37 if url: 

38 self.logger.debug("Checking url style...") 

39 self.base_url = validate_http_url(url) 

40 

41 if session: 

42 self.session = session 

43 self._external_session = True 

44 else: 

45 self.session = None 

46 self._headers = {} 

47 

48 if not fiware_header: 

49 self.fiware_headers = FiwareHeader() 

50 else: 

51 self.fiware_headers = fiware_header 

52 

53 self.headers.update(kwargs.pop('headers', {})) 

54 self.kwargs: Dict = kwargs 

55 

56 # Context Manager Protocol 

57 def __enter__(self): 

58 if not self.session: 

59 self.session = requests.Session() 

60 self.headers.update(self.fiware_headers.model_dump(by_alias=True)) 

61 self._external_session = False 

62 return self 

63 

64 def __exit__(self, exc_type, exc_val, exc_tb): 

65 self.close() 

66 

67 @property 

68 def fiware_headers(self) -> FiwareHeader: 

69 """ 

70 Get fiware header 

71 

72 Returns: 

73 FiwareHeader 

74 """ 

75 return self._fiware_headers 

76 

77 @fiware_headers.setter 

78 def fiware_headers(self, headers: Union[Dict, FiwareHeader]) -> None: 

79 """ 

80 Sets new fiware header 

81 

82 Args: 

83 headers (Dict, FiwareHeader): New headers either as FiwareHeader 

84 object or as dict. 

85 

86 Example: 

87 {fiware-service: "MyService", 

88 fiware-servicepath: "/MyServicePath"} 

89 

90 Returns: 

91 None 

92 """ 

93 if isinstance(headers, FiwareHeader): 

94 self._fiware_headers = headers 

95 elif isinstance(headers, dict): 

96 self._fiware_headers = FiwareHeader.model_validate(headers) 

97 elif isinstance(headers, str): 

98 self._fiware_headers = FiwareHeader.model_validate_json(headers) 

99 else: 

100 raise TypeError(f'Invalid headers! {type(headers)}') 

101 self.headers.update(self.fiware_headers.model_dump(by_alias=True)) 

102 

103 @property 

104 def fiware_service(self) -> str: 

105 """ 

106 Get current fiware service 

107 Returns: 

108 str 

109 """ 

110 return self.fiware_headers.service 

111 

112 @fiware_service.setter 

113 def fiware_service(self, service: str) -> None: 

114 """ 

115 Set new fiware service 

116 Args: 

117 service: 

118 

119 Returns: 

120 None 

121 """ 

122 self._fiware_headers.service = service 

123 self.headers.update(self.fiware_headers.model_dump(by_alias=True)) 

124 

125 @property 

126 def fiware_service_path(self) -> str: 

127 """ 

128 Get current fiware service path 

129 Returns: 

130 str 

131 """ 

132 return self.fiware_headers.service_path 

133 

134 @fiware_service_path.setter 

135 def fiware_service_path(self, service_path: str) -> None: 

136 """ 

137 Set new fiware service path 

138 Args: 

139 service_path (str): New fiware service path. Must start with '/' 

140 

141 Returns: 

142 None 

143 """ 

144 self._fiware_headers.service_path = service_path 

145 self.headers.update(self.fiware_headers.model_dump(by_alias=True)) 

146 

147 @property 

148 def headers(self): 

149 """ 

150 Return current session headers 

151 Returns: 

152 dict with headers 

153 """ 

154 if self.session: 

155 return self.session.headers 

156 return self._headers 

157 

158 # modification to requests api 

159 def get(self, 

160 url: str, 

161 params: Union[Dict, List[Tuple], ByteString] = None, 

162 **kwargs) -> requests.Response: 

163 """ 

164 Sends a GET request either using the provided session or the single 

165 session. 

166 

167 Args: 

168 url (str): URL for the new :class:`Request` object. 

169 params (optional): (optional) Dictionary, list of tuples or bytes 

170 to send in the query string for the :class:`Request`. 

171 **kwargs: Optional arguments that ``request`` takes. 

172 

173 Returns: 

174 requests.Response 

175 """ 

176 

177 kwargs.update({k: v for k, v in self.kwargs.items() 

178 if k not in kwargs.keys()}) 

179 

180 if self.session: 

181 return self.session.get(url=url, params=params, **kwargs) 

182 return requests.get(url=url, params=params, **kwargs) 

183 

184 def options(self, url: str, **kwargs) -> requests.Response: 

185 """ 

186 Sends an OPTIONS request either using the provided session or the 

187 single session. 

188 

189 Args: 

190 url (str): 

191 **kwargs: Optional arguments that ``request`` takes. 

192 

193 Returns: 

194 requests.Response 

195 """ 

196 kwargs.update({k: v for k, v in self.kwargs.items() 

197 if k not in kwargs.keys()}) 

198 

199 if self.session: 

200 return self.session.options(url=url, **kwargs) 

201 return requests.options(url=url, **kwargs) 

202 

203 def head(self, url: str, 

204 params: Union[Dict, List[Tuple], ByteString] = None, 

205 **kwargs) -> requests.Response: 

206 """ 

207 Sends a HEAD request either using the provided session or the 

208 single session. 

209 

210 Args: 

211 url (str): URL for the new :class:`Request` object. 

212 params (optional): Dictionary, list of tuples or bytes 

213 to send in the query string for the :class:`Request`. 

214 **kwargs: Optional arguments that ``request`` takes. 

215 

216 Returns: 

217 requests.Response 

218 """ 

219 kwargs.update({k: v for k, v in self.kwargs.items() 

220 if k not in kwargs.keys()}) 

221 

222 if self.session: 

223 return self.session.head(url=url, params=params, **kwargs) 

224 return requests.head(url=url, params=params, **kwargs) 

225 

226 def post(self, 

227 url: str, 

228 data: Union[Dict, ByteString, List[Tuple], IO, str] = None, 

229 json: Dict = None, 

230 **kwargs) -> requests.Response: 

231 """ 

232 Sends a POST request either using the provided session or the 

233 single session. 

234 

235 Args: 

236 url: URL for the new :class:`Request` object. 

237 data: Dictionary, list of tuples, bytes, or file-like object to 

238 send in the body of the :class:`Request`. 

239 json: A JSON serializable Python object to send in the 

240 body of the :class:`Request`. 

241 **kwargs: Optional arguments that ``request`` takes. 

242 

243 Returns: 

244 

245 """ 

246 kwargs.update({k: v for k, v in self.kwargs.items() 

247 if k not in kwargs.keys()}) 

248 

249 if self.session: 

250 return self.session.post(url=url, data=data, json=json, **kwargs) 

251 return requests.post(url=url, data=data, json=json, **kwargs) 

252 

253 def put(self, 

254 url: str, 

255 data: Union[Dict, ByteString, List[Tuple], IO, str] = None, 

256 json: Dict = None, 

257 **kwargs) -> requests.Response: 

258 """ 

259 Sends a PUT request either using the provided session or the 

260 single session. 

261 

262 Args: 

263 url: URL for the new :class:`Request` object. 

264 data (Union[Dict, ByteString, List[Tuple], IO]): 

265 Dictionary, list of tuples, bytes, or file-like 

266 object to send in the body of the :class:`Request`. 

267 json (Dict): A JSON serializable Python object to send in the 

268 body of the :class:`Request`.. 

269 **kwargs: Optional arguments that ``request`` takes. 

270 

271 Returns: 

272 request.Response 

273 """ 

274 kwargs.update({k: v for k, v in self.kwargs.items() 

275 if k not in kwargs.keys()}) 

276 

277 if self.session: 

278 return self.session.put(url=url, data=data, json=json, **kwargs) 

279 return requests.put(url=url, data=data, json=json, **kwargs) 

280 

281 def patch(self, 

282 url: str, 

283 data: Union[Dict, ByteString, List[Tuple], IO, str] = None, 

284 json: Dict = None, 

285 **kwargs) -> requests.Response: 

286 """ 

287 Sends a PATCH request either using the provided session or the 

288 single session. 

289 

290 Args: 

291 url: URL for the new :class:`Request` object. 

292 data (Union[Dict, ByteString, List[Tuple], IO]): 

293 Dictionary, list of tuples, bytes, or file-like 

294 object to send in the body of the :class:`Request`. 

295 json (Dict): A JSON serializable Python object to send in the 

296 body of the :class:`Request`.. 

297 **kwargs: Optional arguments that ``request`` takes. 

298 

299 Returns: 

300 request.Response 

301 """ 

302 kwargs.update({k: v for k, v in self.kwargs.items() 

303 if k not in kwargs.keys()}) 

304 

305 if self.session: 

306 return self.session.patch(url=url, data=data, json=json, **kwargs) 

307 return requests.patch(url=url, data=data, json=json, **kwargs) 

308 

309 def delete(self, url: str, **kwargs) -> requests.Response: 

310 """ 

311 Sends a DELETE request either using the provided session or the 

312 single session. 

313 

314 Args: 

315 url (str): URL for the new :class:`Request` object. 

316 **kwargs: Optional arguments that ``request`` takes. 

317 

318 Returns: 

319 request.Response 

320 """ 

321 kwargs.update({k: v for k, v in self.kwargs.items() 

322 if k not in kwargs.keys()}) 

323 

324 if self.session: 

325 return self.session.delete(url=url, **kwargs) 

326 return requests.delete(url=url, **kwargs) 

327 

328 def log_error(self, 

329 err: requests.RequestException, 

330 msg: str = None) -> None: 

331 """ 

332 Outputs the error messages from the client request function. If 

333 additional information is available in the server response this will 

334 be forwarded to the logging output. 

335 

336 Note: 

337 The user is responsible to setup the logging system 

338 

339 Args: 

340 err: Request Error 

341 msg: error message from calling function 

342 

343 Returns: 

344 None 

345 """ 

346 if err.response is not None: 

347 if err.response.text and msg: 

348 self.logger.error("%s \n Reason: %s", msg, err.response.text) 

349 elif err.response.text and not msg: 

350 self.logger.error("%s", err.response.text) 

351 elif not err.response and msg: 

352 self.logger.error("%s \n Reason: %s", msg, err) 

353 else: 

354 self.logger.error(err) 

355 

356 def close(self) -> None: 

357 """ 

358 Close http session 

359 Returns: 

360 None 

361 """ 

362 if self.session and not self._external_session: 

363 self.session.close()