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

1""" 

2Base http client module 

3""" 

4 

5import logging 

6from pydantic import AnyHttpUrl 

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

8import requests 

9from filip.models.base import FiwareHeader, FiwareLDHeader 

10from filip.utils import validate_http_url 

11from enum import Enum 

12 

13 

14class NgsiURLVersion(str, Enum): 

15 """ 

16 URL part that defines the NGSI version for the API. 

17 """ 

18 

19 v2_url = "v2" 

20 ld_url = "ngsi-ld/v1" 

21 

22 

23class BaseHttpClient: 

24 """ 

25 Base client for all derived api-clients. 

26 

27 Args: 

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

29 the same connection 

30 reuse_session (bool): 

31 fiware_header: Fiware header object required for multi tenancy 

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

33 

34 """ 

35 

36 def __init__( 

37 self, 

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

39 *, 

40 session: requests.Session = None, 

41 fiware_header: Union[Dict, FiwareHeader, FiwareLDHeader] = None, 

42 **kwargs, 

43 ): 

44 

45 self.logger = logging.getLogger(name=f"{self.__class__.__name__}") 

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

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

48 

49 if url: 

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

51 self.base_url = validate_http_url(url) 

52 

53 if session: 

54 self.session = session 

55 self._external_session = True 

56 else: 

57 self.session = None 

58 self._headers = {} 

59 

60 if not fiware_header: 

61 self.fiware_headers = FiwareHeader() 

62 else: 

63 self.fiware_headers = fiware_header 

64 

65 self.headers.update(kwargs.pop("headers", {})) 

66 self.kwargs: Dict = kwargs 

67 

68 # Context Manager Protocol 

69 def __enter__(self): 

70 if not self.session: 

71 self.session = requests.Session() 

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

73 self._external_session = False 

74 return self 

75 

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

77 self.close() 

78 

79 @property 

80 def fiware_headers(self) -> FiwareHeader: 

81 """ 

82 Get fiware header 

83 

84 Returns: 

85 FiwareHeader 

86 """ 

87 return self._fiware_headers 

88 

89 @fiware_headers.setter 

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

91 """ 

92 Sets new fiware header 

93 

94 Args: 

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

96 object or as dict. 

97 

98 Example: 

99 {fiware-service: "MyService", 

100 fiware-servicepath: "/MyServicePath"} 

101 

102 Returns: 

103 None 

104 """ 

105 if isinstance(headers, FiwareHeader): 

106 self._fiware_headers = headers 

107 elif isinstance(headers, FiwareLDHeader): 

108 self._fiware_headers = headers 

109 elif isinstance(headers, dict): 

110 self._fiware_headers = FiwareHeader.model_validate(headers) 

111 elif isinstance(headers, str): 

112 self._fiware_headers = FiwareHeader.model_validate_json(headers) 

113 elif isinstance(headers, dict): 

114 self._fiware_headers = FiwareLDHeader.parse_obj(headers) 

115 elif isinstance(headers, str): 

116 self._fiware_headers = FiwareLDHeader.parse_raw(headers) 

117 else: 

118 raise TypeError(f"Invalid headers! {type(headers)}") 

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

120 

121 @property 

122 def fiware_service(self) -> str: 

123 """ 

124 Get current fiware service 

125 Returns: 

126 str 

127 """ 

128 return self.fiware_headers.service 

129 

130 @fiware_service.setter 

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

132 """ 

133 Set new fiware service 

134 Args: 

135 service: 

136 

137 Returns: 

138 None 

139 """ 

140 self._fiware_headers.service = service 

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

142 

143 @property 

144 def fiware_service_path(self) -> str: 

145 """ 

146 Get current fiware service path 

147 Returns: 

148 str 

149 """ 

150 return self.fiware_headers.service_path 

151 

152 @fiware_service_path.setter 

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

154 """ 

155 Set new fiware service path 

156 Args: 

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

158 

159 Returns: 

160 None 

161 """ 

162 self._fiware_headers.service_path = service_path 

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

164 

165 @property 

166 def headers(self): 

167 """ 

168 Return current session headers 

169 Returns: 

170 dict with headers 

171 """ 

172 if self.session: 

173 return self.session.headers 

174 return self._headers 

175 

176 # modification to requests api 

177 def get( 

178 self, url: str, params: Union[Dict, List[Tuple], ByteString] = None, **kwargs 

179 ) -> requests.Response: 

180 """ 

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

182 session. 

183 

184 Args: 

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

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

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

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

189 

190 Returns: 

191 requests.Response 

192 """ 

193 

194 kwargs.update({k: v for k, v in self.kwargs.items() if k not in kwargs.keys()}) 

195 

196 if self.session: 

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

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

199 

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

201 """ 

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

203 single session. 

204 

205 Args: 

206 url (str): 

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

208 

209 Returns: 

210 requests.Response 

211 """ 

212 kwargs.update({k: v for k, v in self.kwargs.items() if k not in kwargs.keys()}) 

213 

214 if self.session: 

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

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

217 

218 def head( 

219 self, url: str, params: Union[Dict, List[Tuple], ByteString] = None, **kwargs 

220 ) -> requests.Response: 

221 """ 

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

223 single session. 

224 

225 Args: 

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

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

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

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

230 

231 Returns: 

232 requests.Response 

233 """ 

234 kwargs.update({k: v for k, v in self.kwargs.items() if k not in kwargs.keys()}) 

235 

236 if self.session: 

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

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

239 

240 def post( 

241 self, 

242 url: str, 

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

244 json: Dict = None, 

245 **kwargs, 

246 ) -> requests.Response: 

247 """ 

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

249 single session. 

250 

251 Args: 

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

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

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

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

256 body of the :class:`Request`. 

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

258 

259 Returns: 

260 

261 """ 

262 kwargs.update({k: v for k, v in self.kwargs.items() if k not in kwargs.keys()}) 

263 

264 if self.session: 

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

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

267 

268 def put( 

269 self, 

270 url: str, 

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

272 json: Dict = None, 

273 **kwargs, 

274 ) -> requests.Response: 

275 """ 

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

277 single session. 

278 

279 Args: 

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

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

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

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

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

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

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

287 

288 Returns: 

289 request.Response 

290 """ 

291 kwargs.update({k: v for k, v in self.kwargs.items() if k not in kwargs.keys()}) 

292 

293 if self.session: 

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

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

296 

297 def patch( 

298 self, 

299 url: str, 

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

301 json: Dict = None, 

302 **kwargs, 

303 ) -> requests.Response: 

304 """ 

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

306 single session. 

307 

308 Args: 

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

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

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

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

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

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

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

316 

317 Returns: 

318 request.Response 

319 """ 

320 kwargs.update({k: v for k, v in self.kwargs.items() if k not in kwargs.keys()}) 

321 

322 if self.session: 

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

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

325 

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

327 """ 

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

329 single session. 

330 

331 Args: 

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

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

334 

335 Returns: 

336 request.Response 

337 """ 

338 kwargs.update({k: v for k, v in self.kwargs.items() if k not in kwargs.keys()}) 

339 

340 if self.session: 

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

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

343 

344 def log_error(self, err: requests.RequestException, msg: str = None) -> None: 

345 """ 

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

347 additional information is available in the server response this will 

348 be forwarded to the logging output. 

349 

350 Note: 

351 The user is responsible to setup the logging system 

352 

353 Args: 

354 err: Request Error 

355 msg: error message from calling function 

356 

357 Returns: 

358 None 

359 """ 

360 if err.response is not None: 

361 if err.response.text and msg: 

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

363 elif err.response.text and not msg: 

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

365 elif not err.response and msg: 

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

367 else: 

368 self.logger.error(err) 

369 

370 def close(self) -> None: 

371 """ 

372 Close http session 

373 Returns: 

374 None 

375 """ 

376 if self.session and not self._external_session: 

377 self.session.close()