Coverage for bounciepy/async_rest_api_client.py: 100%

1from typing import Optional, Union 

2 

3from aiohttp import ClientResponse, ClientSession, ClientTimeout 

4 

5from bounciepy.const import ( 

6 API_DEFAULT_TIMEOUT_SECONDS, 

7 AUTH_GRANT_TYPE, 

8 AUTH_TOKEN_URL, 

9 REST_API_BASE_URL, 

10) 

11from bounciepy.exceptions import BadRequestError, UnauthorizedError 

12 

13 

14class AsyncRESTAPIClient: 

15 def __init__( 

16 self, 

17 client_id: str, 

18 client_secret: str, 

19 redirect_url: str, 

20 auth_code: str, 

21 state: str = "init_bouncie", 

22 session: Optional[ClientSession] = None, 

23 ) -> None: 

24 self._client_id: str = client_id 

25 self._client_secret: str = client_secret 

26 self._redirect_url: str = redirect_url 

27 self._auth_code: str = auth_code 

28 self._state: str = state 

29 self._session: Optional[ClientSession] = session 

30 self._access_token: Optional[str] = None 

31 self._access_token_valid: bool = False 

32 self._user_email: Optional[str] = None 

33 self._user_name: Optional[str] = None 

34 self._user_id: Optional[str] = None 

35 self._vehicles: list = [] 

36 self._headers: dict = {} 

37 

38 @property 

39 def client_session(self) -> Optional[ClientSession]: 

40 return self._session 

41 

42 @property 

43 def access_token(self) -> Optional[str]: 

44 return self._access_token 

45 

46 def _set_access_token(self, access_token: str) -> None: 

47 self._access_token = access_token 

48 self._headers = {"Authorization": access_token} 

49 self._access_token_valid = True 

50 

51 async def _handle_response( 

52 self, response: ClientResponse 

53 ) -> Union[dict, list, None]: 

54 data = None 

55 if response.status in (200, 201): 

56 data = await response.json() 

57 elif response.status == 400: 

58 data = await response.json() 

59 raise BadRequestError(data["errors"]) 

60 elif response.status == 401: 

61 self._access_token_valid = False 

62 raise UnauthorizedError("Error: Invalid or expired access token.") 

63 return data 

64 

65 async def _get_session(self) -> ClientSession: 

66 if not self._session or self._session.closed: 

67 self._session = ClientSession( 

68 timeout=ClientTimeout(total=API_DEFAULT_TIMEOUT_SECONDS) 

69 ) 

70 return self._session 

71 

72 async def get_access_token(self) -> bool: 

73 current_session = await self._get_session() 

74 data: dict = { 

75 "client_id": self._client_id, 

76 "client_secret": self._client_secret, 

77 "grant_type": AUTH_GRANT_TYPE, 

78 "code": self._auth_code, 

79 "redirect_uri": self._redirect_url, 

80 } 

81 async with current_session.post( 

82 url=AUTH_TOKEN_URL, 

83 data=data, 

84 ) as response: 

85 response_data = await self._handle_response(response) 

86 if not response_data: 

87 return False 

88 if isinstance(response_data, dict): 

89 self._set_access_token(access_token=response_data["access_token"]) 

90 return True 

91 

92 async def http_get(self, url: str, **kwargs: dict) -> Union[dict, list, None]: 

93 count = 0 

94 while count < 2: 

95 try: 

96 current_session = await self._get_session() 

97 response = await current_session.get( 

98 url=url, headers=self._headers, allow_redirects=True, **kwargs 

99 ) 

100 data = await self._handle_response(response) 

101 count = 2 

102 except UnauthorizedError: 

103 if await self.get_access_token(): 

104 count = 1 

105 return data 

106 

107 async def get_user(self) -> Optional[dict]: 

108 user_data = await self.http_get(f"{REST_API_BASE_URL}/user") 

109 if user_data and isinstance(user_data, dict): 

110 self._user_name = user_data["name"] if "name" in user_data else None 

111 self._user_email = user_data["email"] if "email" in user_data else None 

112 self._user_id = user_data["id"] if "id" in user_data else None 

113 return user_data 

114 return None 

115 

116 async def get_all_vehicles(self) -> Optional[list]: 

117 vehicles_data = await self.http_get(f"{REST_API_BASE_URL}/vehicles") 

118 if vehicles_data and isinstance(vehicles_data, list): 

119 self._vehicles = vehicles_data 

120 return vehicles_data 

121 return None 

122 

123 async def get_vehicle_by_imei(self, imei: str) -> Optional[dict]: 

124 vehicle_data = await self.http_get( 

125 f"{REST_API_BASE_URL}/vehicles", 

126 params={"imei": imei}, 

127 ) 

128 if isinstance(vehicle_data, list): 

129 return vehicle_data[0] 

130 return None 

131 

132 async def get_vehicle_by_vin(self, vin): 

133 vehicle_data = await self.http_get( 

134 url=f"{REST_API_BASE_URL}/vehicles", 

135 params={"vin": vin}, 

136 ) 

137 if isinstance(vehicle_data, list): 

138 return vehicle_data[0] 

139 return None 

140 

141 async def get_trips(self, imei: str, gps_format: str="geojson") -> Optional[list]: 

142 trip_data = await self.http_get( 

143 f"{REST_API_BASE_URL}/trips", 

144 params={"imei": imei, 

145 "gps-format": gps_format}, 

146 ) 

147 if trip_data and isinstance(trip_data, list): 

148 return trip_data 

149 return None