Coverage for odmpy/overdrive.py: 96.8%

1# Copyright (C) 2023 github.com/ping 

2# 

3# This file is part of odmpy. 

4# 

5# odmpy is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# (at your option) any later version. 

9# 

10# odmpy is distributed in the hope that it will be useful, 

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

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

13# GNU General Public License for more details. 

14# 

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

16# along with odmpy. If not, see <http://www.gnu.org/licenses/>. 

17# 

18 

19import logging 

20from typing import Optional, Dict, List 

21from urllib.parse import urljoin 

22 

23import requests 

24from requests.adapters import HTTPAdapter, Retry 

25 

26# 

27# Basic skeletal client for the OverDrive Thunder API 

28# 

29 

30USER_AGENT = ( 

31 "Mozilla/5.0 (Macintosh; Intel Mac OS X 11_1) AppleWebKit/605.1.15 (KHTML, like Gecko) " # noqa 

32 "Version/14.0.2 Safari/605.1.15" 

33) 

34SITE_URL = "https://libbyapp.com" 

35THUNDER_API_URL = "https://thunder.api.overdrive.com/v2/" 

36CLIENT_ID = "dewey" 

37 

38 

39class OverDriveClient(object): 

40 """ 

41 A really simplified OverDrive Thunder API client 

42 """ 

43 

44 def __init__(self, **kwargs) -> None: 

45 """ 

46 Constructor. 

47 

48 :param kwargs: 

49 - user_agent: User Agent string for requests 

50 - timeout: The timeout interval for a network request. Default 15 (seconds). 

51 - retries: The number of times to retry a network request on failure. Default 0. 

52 """ 

53 self.logger = logging.getLogger(__name__) 

54 self.user_agent = kwargs.pop("user_agent", USER_AGENT) 

55 self.timeout = int(kwargs.pop("timeout", 15)) 

56 self.retries = int(kwargs.pop("retry", 0)) 

57 

58 session = requests.Session() 

59 adapter = HTTPAdapter(max_retries=Retry(total=self.retries, backoff_factor=0.1)) 

60 # noinspection HttpUrlsUsage 

61 for prefix in ("http://", "https://"): 

62 session.mount(prefix, adapter) 

63 self.session = kwargs.pop("session", None) or session 

64 

65 def default_headers(self) -> Dict: 

66 """ 

67 Default http request headers. 

68 

69 :return: 

70 """ 

71 headers = { 

72 "User-Agent": self.user_agent, 

73 "Referer": SITE_URL + "/", 

74 "Origin": SITE_URL, 

75 "Cache-Control": "no-cache", 

76 "Pragma": "no-cache", 

77 } 

78 return headers 

79 

80 def default_params(self) -> Dict: 

81 """ 

82 Default set of GET request parameters. 

83 

84 :return: 

85 """ 

86 params = {"x-client-id": CLIENT_ID} 

87 return params 

88 

89 def make_request( 

90 self, 

91 endpoint: str, 

92 params: Optional[Dict] = None, 

93 data: Optional[Dict] = None, 

94 headers: Optional[Dict] = None, 

95 method: Optional[str] = None, 

96 ): 

97 """ 

98 Sends an API request. 

99 

100 :param endpoint: Relative path to endpoint 

101 :param params: URL query parameters 

102 :param data: POST data parameters 

103 :param method: HTTP method, e.g. 'PUT' 

104 :param headers: Custom headers 

105 :return: Union[List, Dict, str] 

106 """ 

107 endpoint_url = urljoin(THUNDER_API_URL, endpoint) 

108 headers = headers or self.default_headers() 

109 if not method: 

110 # try to set an HTTP method 

111 if data is not None: 

112 method = "POST" 

113 else: 

114 method = "GET" 

115 

116 req = requests.Request( 

117 method, 

118 endpoint_url, 

119 headers=headers, 

120 params=params, 

121 data=data, 

122 ) 

123 res = self.session.send(self.session.prepare_request(req), timeout=self.timeout) 

124 self.logger.debug("body: %s", res.text) 

125 res.raise_for_status() 

126 

127 if res.headers.get("content-type", "").startswith("application/json"): 

128 return res.json() 

129 return res.text 

130 

131 def media(self, title_id: str, **kwargs) -> Dict: 

132 """ 

133 Retrieve a title. 

134 Title id can also be a reserve id. 

135 

136 :param title_id: A unique id that identifies the content. 

137 :return: 

138 """ 

139 params = self.default_params() 

140 params.update(kwargs) 

141 return self.make_request(f"media/{title_id}", params=params) 

142 

143 def media_bulk(self, title_ids: List[str], **kwargs) -> List[Dict]: 

144 """ 

145 Retrieve a list of titles. 

146 

147 :param title_ids: The ids passed in this request can be titleIds or reserveIds. 

148 :return: 

149 """ 

150 params = self.default_params() 

151 params.update({"titleIds": ",".join(title_ids)}) 

152 params.update(kwargs) 

153 return self.make_request("media/bulk", params=params) 

154 

155 def library(self, library_key: str, **kwargs) -> Dict: 

156 """ 

157 Get a library's configuration data. 

158 

159 :param library_key: A unique key that identifies the library, e.g. lapl 

160 :param kwargs: 

161 :return: 

162 """ 

163 params = self.default_params() 

164 params.update(kwargs) 

165 return self.make_request(f"libraries/{library_key}", params=params) 

166 

167 def library_media(self, library_key: str, title_id: str, **kwargs) -> dict: 

168 """ 

169 Get title. 

170 

171 :param library_key: A unique key that identifies the library 

172 :param title_id: 

173 :return: 

174 """ 

175 params = self.default_params() 

176 params.update({"titleIds": title_id}) 

177 params.update(kwargs) 

178 return self.make_request( 

179 f"libraries/{library_key}/media/{title_id}", params=params 

180 ) 

181 

182 def library_media_availability( 

183 self, library_key: str, title_id: str, **kwargs 

184 ) -> dict: 

185 """ 

186 Get title availability. 

187 

188 :param library_key: A unique key that identifies the library, e.g. lapl 

189 :param title_id: 

190 :param kwargs: 

191 :return: 

192 """ 

193 params = self.default_params() 

194 params.update(kwargs) 

195 return self.make_request( 

196 f"libraries/{library_key}/media/{title_id}/availability", params=params 

197 )