Coverage for src/hush_hush/_generated/client.py: 54%

1# Code generated by openapi-python-client from hush-hush/api/openapi.yaml; DO NOT EDIT. 

2 

3import ssl 

4from typing import Any, Self 

5 

6import httpx 

7from attrs import define, evolve, field 

8 

9 

10@define 

11class Client: 

12 """A class for keeping track of data related to the API 

13 

14 The following are accepted as keyword arguments and will be used to construct httpx Clients internally: 

15 

16 ``base_url``: The base URL for the API, all requests are made to a relative path to this URL 

17 

18 ``cookies``: A dictionary of cookies to be sent with every request 

19 

20 ``headers``: A dictionary of headers to be sent with every request 

21 

22 ``timeout``: The maximum amount of a time a request can take. API functions will raise 

23 httpx.TimeoutException if this is exceeded. 

24 

25 ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production, 

26 but can be set to False for testing purposes. 

27 

28 ``follow_redirects``: Whether or not to follow redirects. Default value is False. 

29 

30 ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor. 

31 

32 

33 Attributes: 

34 raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a 

35 status code that was not documented in the source OpenAPI document. Can also be provided as a keyword 

36 argument to the constructor. 

37 """ 

38 

39 raise_on_unexpected_status: bool = field(default=False, kw_only=True) 

40 _base_url: str = field(alias="base_url") 

41 _cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") 

42 _headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers") 

43 _timeout: httpx.Timeout | None = field(default=None, kw_only=True, alias="timeout") 

44 _verify_ssl: str | bool | ssl.SSLContext = field(default=True, kw_only=True, alias="verify_ssl") 

45 _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") 

46 _httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") 

47 _client: httpx.Client | None = field(default=None, init=False) 

48 _async_client: httpx.AsyncClient | None = field(default=None, init=False) 

49 

50 def with_headers(self, headers: dict[str, str]) -> "Client": 

51 """Get a new client matching this one with additional headers""" 

52 if self._client is not None: 

53 self._client.headers.update(headers) 

54 if self._async_client is not None: 

55 self._async_client.headers.update(headers) 

56 return evolve(self, headers={**self._headers, **headers}) 

57 

58 def with_cookies(self, cookies: dict[str, str]) -> "Client": 

59 """Get a new client matching this one with additional cookies""" 

60 if self._client is not None: 

61 self._client.cookies.update(cookies) 

62 if self._async_client is not None: 

63 self._async_client.cookies.update(cookies) 

64 return evolve(self, cookies={**self._cookies, **cookies}) 

65 

66 def with_timeout(self, timeout: httpx.Timeout) -> "Client": 

67 """Get a new client matching this one with a new timeout configuration""" 

68 if self._client is not None: 

69 self._client.timeout = timeout 

70 if self._async_client is not None: 

71 self._async_client.timeout = timeout 

72 return evolve(self, timeout=timeout) 

73 

74 def set_httpx_client(self, client: httpx.Client) -> Self: 

75 """Manually set the underlying httpx.Client 

76 

77 **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout. 

78 """ 

79 self._client = client 

80 return self 

81 

82 def get_httpx_client(self) -> httpx.Client: 

83 """Get the underlying httpx.Client, constructing a new one if not previously set""" 

84 if self._client is None: 

85 self._client = httpx.Client( 

86 base_url=self._base_url, 

87 cookies=self._cookies, 

88 headers=self._headers, 

89 timeout=self._timeout, 

90 verify=self._verify_ssl, 

91 follow_redirects=self._follow_redirects, 

92 **self._httpx_args, 

93 ) 

94 return self._client 

95 

96 def __enter__(self) -> Self: 

97 """Enter a context manager for self.client—you cannot enter twice (see httpx docs)""" 

98 self.get_httpx_client().__enter__() 

99 return self 

100 

101 def __exit__(self, *args: Any, **kwargs: Any) -> None: 

102 """Exit a context manager for internal httpx.Client (see httpx docs)""" 

103 self.get_httpx_client().__exit__(*args, **kwargs) 

104 

105 def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> Self: 

106 """Manually set the underlying httpx.AsyncClient 

107 

108 **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout. 

109 """ 

110 self._async_client = async_client 

111 return self 

112 

113 def get_async_httpx_client(self) -> httpx.AsyncClient: 

114 """Get the underlying httpx.AsyncClient, constructing a new one if not previously set""" 

115 if self._async_client is None: 

116 self._async_client = httpx.AsyncClient( 

117 base_url=self._base_url, 

118 cookies=self._cookies, 

119 headers=self._headers, 

120 timeout=self._timeout, 

121 verify=self._verify_ssl, 

122 follow_redirects=self._follow_redirects, 

123 **self._httpx_args, 

124 ) 

125 return self._async_client 

126 

127 async def __aenter__(self) -> Self: 

128 """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)""" 

129 await self.get_async_httpx_client().__aenter__() 

130 return self 

131 

132 async def __aexit__(self, *args: Any, **kwargs: Any) -> None: 

133 """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)""" 

134 await self.get_async_httpx_client().__aexit__(*args, **kwargs) 

135 

136 

137@define 

138class AuthenticatedClient: 

139 """A Client which has been authenticated for use on secured endpoints 

140 

141 The following are accepted as keyword arguments and will be used to construct httpx Clients internally: 

142 

143 ``base_url``: The base URL for the API, all requests are made to a relative path to this URL 

144 

145 ``cookies``: A dictionary of cookies to be sent with every request 

146 

147 ``headers``: A dictionary of headers to be sent with every request 

148 

149 ``timeout``: The maximum amount of a time a request can take. API functions will raise 

150 httpx.TimeoutException if this is exceeded. 

151 

152 ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production, 

153 but can be set to False for testing purposes. 

154 

155 ``follow_redirects``: Whether or not to follow redirects. Default value is False. 

156 

157 ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor. 

158 

159 

160 Attributes: 

161 raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a 

162 status code that was not documented in the source OpenAPI document. Can also be provided as a keyword 

163 argument to the constructor. 

164 token: The token to use for authentication 

165 prefix: The prefix to use for the Authorization header 

166 auth_header_name: The name of the Authorization header 

167 """ 

168 

169 raise_on_unexpected_status: bool = field(default=False, kw_only=True) 

170 _base_url: str = field(alias="base_url") 

171 _cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") 

172 _headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers") 

173 _timeout: httpx.Timeout | None = field(default=None, kw_only=True, alias="timeout") 

174 _verify_ssl: str | bool | ssl.SSLContext = field(default=True, kw_only=True, alias="verify_ssl") 

175 _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") 

176 _httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") 

177 _client: httpx.Client | None = field(default=None, init=False) 

178 _async_client: httpx.AsyncClient | None = field(default=None, init=False) 

179 

180 token: str 

181 prefix: str = "Bearer" 

182 auth_header_name: str = "Authorization" 

183 

184 def with_headers(self, headers: dict[str, str]) -> "AuthenticatedClient": 

185 """Get a new client matching this one with additional headers""" 

186 if self._client is not None: 

187 self._client.headers.update(headers) 

188 if self._async_client is not None: 

189 self._async_client.headers.update(headers) 

190 return evolve(self, headers={**self._headers, **headers}) 

191 

192 def with_cookies(self, cookies: dict[str, str]) -> "AuthenticatedClient": 

193 """Get a new client matching this one with additional cookies""" 

194 if self._client is not None: 

195 self._client.cookies.update(cookies) 

196 if self._async_client is not None: 

197 self._async_client.cookies.update(cookies) 

198 return evolve(self, cookies={**self._cookies, **cookies}) 

199 

200 def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient": 

201 """Get a new client matching this one with a new timeout configuration""" 

202 if self._client is not None: 

203 self._client.timeout = timeout 

204 if self._async_client is not None: 

205 self._async_client.timeout = timeout 

206 return evolve(self, timeout=timeout) 

207 

208 def set_httpx_client(self, client: httpx.Client) -> Self: 

209 """Manually set the underlying httpx.Client 

210 

211 **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout. 

212 """ 

213 self._client = client 

214 return self 

215 

216 def get_httpx_client(self) -> httpx.Client: 

217 """Get the underlying httpx.Client, constructing a new one if not previously set""" 

218 if self._client is None: 

219 self._headers[self.auth_header_name] = ( 

220 f"{self.prefix} {self.token}" if self.prefix else self.token 

221 ) 

222 self._client = httpx.Client( 

223 base_url=self._base_url, 

224 cookies=self._cookies, 

225 headers=self._headers, 

226 timeout=self._timeout, 

227 verify=self._verify_ssl, 

228 follow_redirects=self._follow_redirects, 

229 **self._httpx_args, 

230 ) 

231 return self._client 

232 

233 def __enter__(self) -> Self: 

234 """Enter a context manager for self.client—you cannot enter twice (see httpx docs)""" 

235 self.get_httpx_client().__enter__() 

236 return self 

237 

238 def __exit__(self, *args: Any, **kwargs: Any) -> None: 

239 """Exit a context manager for internal httpx.Client (see httpx docs)""" 

240 self.get_httpx_client().__exit__(*args, **kwargs) 

241 

242 def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> Self: 

243 """Manually set the underlying httpx.AsyncClient 

244 

245 **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout. 

246 """ 

247 self._async_client = async_client 

248 return self 

249 

250 def get_async_httpx_client(self) -> httpx.AsyncClient: 

251 """Get the underlying httpx.AsyncClient, constructing a new one if not previously set""" 

252 if self._async_client is None: 

253 self._headers[self.auth_header_name] = ( 

254 f"{self.prefix} {self.token}" if self.prefix else self.token 

255 ) 

256 self._async_client = httpx.AsyncClient( 

257 base_url=self._base_url, 

258 cookies=self._cookies, 

259 headers=self._headers, 

260 timeout=self._timeout, 

261 verify=self._verify_ssl, 

262 follow_redirects=self._follow_redirects, 

263 **self._httpx_args, 

264 ) 

265 return self._async_client 

266 

267 async def __aenter__(self) -> Self: 

268 """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)""" 

269 await self.get_async_httpx_client().__aenter__() 

270 return self 

271 

272 async def __aexit__(self, *args: Any, **kwargs: Any) -> None: 

273 """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)""" 

274 await self.get_async_httpx_client().__aexit__(*args, **kwargs)