Coverage for fio_wrapper/fio_adapter.py: 100%

1"""Request adapter performing actual API calls towards FIO endpoints 

2""" 

3import logging 

4from typing import Dict 

5from typing import List 

6from typing import Optional 

7from typing import Tuple 

8import importlib.util 

9 

10import requests 

11from requests.packages.urllib3.exceptions import InsecureRequestWarning 

12from fio_wrapper.config import Config 

13 

14from fio_wrapper.exceptions import UnknownFIOResponse 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class FIOAdapter: 

20 """FIO Adapater""" 

21 

22 def __init__(self, config: Config, header: Dict[str, str]): 

23 """Initializes the FIO adapter 

24 

25 Args: 

26 config (Config): Wrapper configuration. 

27 header (Dict[str, str]): FIO Header. 

28 """ 

29 

30 self.config = config 

31 self.header = header 

32 self.ssl_verify = self.config.ssl_verify 

33 self.timeout = self.config.timeout 

34 

35 if not self.ssl_verify: 

36 requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) 

37 

38 # register a session based on requests.session or, if installed 

39 # requests-cache session handler 

40 

41 if self.config.cache and importlib.util.find_spec("requests_cache") is not None: 

42 logger.debug("Using requests-cache Session") 

43 from requests_cache import CachedSession 

44 

45 self._session = CachedSession( 

46 cache_name="fio_wrapper", 

47 backend="memory", 

48 expire_after=self.config.cache_default_expire, 

49 urls_expire_after=self.config.cache_url_expirations(), 

50 cache_control=False, 

51 ) 

52 else: 

53 logger.debug("Using request Session") 

54 self._session = requests.session() 

55 

56 def _do( 

57 self, 

58 http_method: str, 

59 endpoint: str, 

60 params: Dict = None, 

61 data: Dict = None, 

62 err_codes: Optional[List[int]] = None, 

63 timeout: Optional[float] = None, 

64 ) -> Tuple[int, any]: 

65 try: 

66 logger.debug( 

67 "Calling (%s) %s with params: %s | data: %s", 

68 http_method, 

69 endpoint, 

70 params, 

71 data, 

72 ) 

73 

74 response = self._session.request( 

75 method=http_method, 

76 url=endpoint, 

77 verify=self.ssl_verify, 

78 headers=self.header, 

79 params=params, 

80 json=data, 

81 timeout=timeout if timeout is not None else self.timeout, 

82 ) 

83 

84 # successful FIO response 

85 if response.status_code == 200: 

86 return 200, response.json() 

87 # FIO response in provided codes to catch in endpoint 

88 elif isinstance(err_codes, List) and response.status_code in err_codes: 

89 return response.status_code, None 

90 # other FIO response, not accounted for 

91 else: 

92 raise UnknownFIOResponse() 

93 

94 except requests.exceptions.Timeout as errt: 

95 raise requests.exceptions.Timeout() from errt 

96 except requests.exceptions.TooManyRedirects as errr: 

97 raise requests.exceptions.TooManyRedirects from errr 

98 except requests.exceptions.RequestException as e: 

99 raise SystemExit(e) 

100 

101 def get( 

102 self, 

103 endpoint: str, 

104 params: Dict = None, 

105 err_codes: Optional[List[int]] = None, 

106 timeout: Optional[float] = None, 

107 ) -> Tuple[int, any]: 

108 """Performs a GET request towards endpoint 

109 

110 Args: 

111 endpoint (str): URL 

112 params (Dict, optional): GET parameters. Defaults to None. 

113 err_codes (List[int], optional): List of error codes to handle in calling function. Defaults to None. 

114 timeout (float, optional): Request timeout in seconds. Defaults to None. 

115 

116 Returns: 

117 Tuple[int, any]: Request status code and request data 

118 """ 

119 return self._do( 

120 http_method="GET", 

121 endpoint=endpoint, 

122 params=params, 

123 err_codes=err_codes, 

124 timeout=timeout, 

125 ) 

126 

127 def post( 

128 self, 

129 endpoint: str, 

130 params: Dict = None, 

131 data: Dict = None, 

132 err_codes: Optional[List[int]] = None, 

133 timeout: Optional[float] = None, 

134 ) -> Tuple[int, any]: 

135 """Performs a POST request towards endpoint 

136 

137 Args: 

138 endpoint (str): URL 

139 params (Dict, optional): POST parameters. Defaults to None. 

140 data (Dict, optional): POST data. Defaults to None. 

141 err_codes (List[int], optional): List of error codes to handle in calling function. Defaults to None. 

142 timeout (float): Request timeout in seconds. Defaults to None. 

143 

144 Returns: 

145 Tuple[int, any]: Request status code and request data 

146 """ 

147 return self._do( 

148 http_method="POST", 

149 endpoint=endpoint, 

150 params=params, 

151 data=data, 

152 err_codes=err_codes, 

153 timeout=timeout, 

154 )