qtsurfer.api.client

136@define
137class AuthenticatedClient:
138    """A Client which has been authenticated for use on secured endpoints
139
140    The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
141
142        ``base_url``: The base URL for the API, all requests are made to a relative path to this URL
143
144        ``cookies``: A dictionary of cookies to be sent with every request
145
146        ``headers``: A dictionary of headers to be sent with every request
147
148        ``timeout``: The maximum amount of a time a request can take. API functions will raise
149        httpx.TimeoutException if this is exceeded.
150
151        ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
152        but can be set to False for testing purposes.
153
154        ``follow_redirects``: Whether or not to follow redirects. Default value is False.
155
156        ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
157
158
159    Attributes:
160        raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
161            status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
162            argument to the constructor.
163        token: The token to use for authentication
164        prefix: The prefix to use for the Authorization header
165        auth_header_name: The name of the Authorization header
166    """
167
168    raise_on_unexpected_status: bool = field(default=False, kw_only=True)
169    _base_url: str = field(alias="base_url")
170    _cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
171    _headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
172    _timeout: httpx.Timeout | None = field(default=None, kw_only=True, alias="timeout")
173    _verify_ssl: str | bool | ssl.SSLContext = field(default=True, kw_only=True, alias="verify_ssl")
174    _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
175    _httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
176    _client: httpx.Client | None = field(default=None, init=False)
177    _async_client: httpx.AsyncClient | None = field(default=None, init=False)
178
179    token: str
180    prefix: str = "Bearer"
181    auth_header_name: str = "Authorization"
182
183    def with_headers(self, headers: dict[str, str]) -> "AuthenticatedClient":
184        """Get a new client matching this one with additional headers"""
185        if self._client is not None:
186            self._client.headers.update(headers)
187        if self._async_client is not None:
188            self._async_client.headers.update(headers)
189        return evolve(self, headers={**self._headers, **headers})
190
191    def with_cookies(self, cookies: dict[str, str]) -> "AuthenticatedClient":
192        """Get a new client matching this one with additional cookies"""
193        if self._client is not None:
194            self._client.cookies.update(cookies)
195        if self._async_client is not None:
196            self._async_client.cookies.update(cookies)
197        return evolve(self, cookies={**self._cookies, **cookies})
198
199    def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient":
200        """Get a new client matching this one with a new timeout configuration"""
201        if self._client is not None:
202            self._client.timeout = timeout
203        if self._async_client is not None:
204            self._async_client.timeout = timeout
205        return evolve(self, timeout=timeout)
206
207    def set_httpx_client(self, client: httpx.Client) -> "AuthenticatedClient":
208        """Manually set the underlying httpx.Client
209
210        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
211        """
212        self._client = client
213        return self
214
215    def get_httpx_client(self) -> httpx.Client:
216        """Get the underlying httpx.Client, constructing a new one if not previously set"""
217        if self._client is None:
218            self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
219            self._client = httpx.Client(
220                base_url=self._base_url,
221                cookies=self._cookies,
222                headers=self._headers,
223                timeout=self._timeout,
224                verify=self._verify_ssl,
225                follow_redirects=self._follow_redirects,
226                **self._httpx_args,
227            )
228        return self._client
229
230    def __enter__(self) -> "AuthenticatedClient":
231        """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
232        self.get_httpx_client().__enter__()
233        return self
234
235    def __exit__(self, *args: Any, **kwargs: Any) -> None:
236        """Exit a context manager for internal httpx.Client (see httpx docs)"""
237        self.get_httpx_client().__exit__(*args, **kwargs)
238
239    def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "AuthenticatedClient":
240        """Manually set the underlying httpx.AsyncClient
241
242        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
243        """
244        self._async_client = async_client
245        return self
246
247    def get_async_httpx_client(self) -> httpx.AsyncClient:
248        """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
249        if self._async_client is None:
250            self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
251            self._async_client = httpx.AsyncClient(
252                base_url=self._base_url,
253                cookies=self._cookies,
254                headers=self._headers,
255                timeout=self._timeout,
256                verify=self._verify_ssl,
257                follow_redirects=self._follow_redirects,
258                **self._httpx_args,
259            )
260        return self._async_client
261
262    async def __aenter__(self) -> "AuthenticatedClient":
263        """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
264        await self.get_async_httpx_client().__aenter__()
265        return self
266
267    async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
268        """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
269        await self.get_async_httpx_client().__aexit__(*args, **kwargs)

  9@define
 10class Client:
 11    """A class for keeping track of data related to the API
 12
 13    The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
 14
 15        ``base_url``: The base URL for the API, all requests are made to a relative path to this URL
 16
 17        ``cookies``: A dictionary of cookies to be sent with every request
 18
 19        ``headers``: A dictionary of headers to be sent with every request
 20
 21        ``timeout``: The maximum amount of a time a request can take. API functions will raise
 22        httpx.TimeoutException if this is exceeded.
 23
 24        ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
 25        but can be set to False for testing purposes.
 26
 27        ``follow_redirects``: Whether or not to follow redirects. Default value is False.
 28
 29        ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
 30
 31
 32    Attributes:
 33        raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
 34            status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
 35            argument to the constructor.
 36    """
 37
 38    raise_on_unexpected_status: bool = field(default=False, kw_only=True)
 39    _base_url: str = field(alias="base_url")
 40    _cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
 41    _headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
 42    _timeout: httpx.Timeout | None = field(default=None, kw_only=True, alias="timeout")
 43    _verify_ssl: str | bool | ssl.SSLContext = field(default=True, kw_only=True, alias="verify_ssl")
 44    _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
 45    _httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
 46    _client: httpx.Client | None = field(default=None, init=False)
 47    _async_client: httpx.AsyncClient | None = field(default=None, init=False)
 48
 49    def with_headers(self, headers: dict[str, str]) -> "Client":
 50        """Get a new client matching this one with additional headers"""
 51        if self._client is not None:
 52            self._client.headers.update(headers)
 53        if self._async_client is not None:
 54            self._async_client.headers.update(headers)
 55        return evolve(self, headers={**self._headers, **headers})
 56
 57    def with_cookies(self, cookies: dict[str, str]) -> "Client":
 58        """Get a new client matching this one with additional cookies"""
 59        if self._client is not None:
 60            self._client.cookies.update(cookies)
 61        if self._async_client is not None:
 62            self._async_client.cookies.update(cookies)
 63        return evolve(self, cookies={**self._cookies, **cookies})
 64
 65    def with_timeout(self, timeout: httpx.Timeout) -> "Client":
 66        """Get a new client matching this one with a new timeout configuration"""
 67        if self._client is not None:
 68            self._client.timeout = timeout
 69        if self._async_client is not None:
 70            self._async_client.timeout = timeout
 71        return evolve(self, timeout=timeout)
 72
 73    def set_httpx_client(self, client: httpx.Client) -> "Client":
 74        """Manually set the underlying httpx.Client
 75
 76        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
 77        """
 78        self._client = client
 79        return self
 80
 81    def get_httpx_client(self) -> httpx.Client:
 82        """Get the underlying httpx.Client, constructing a new one if not previously set"""
 83        if self._client is None:
 84            self._client = httpx.Client(
 85                base_url=self._base_url,
 86                cookies=self._cookies,
 87                headers=self._headers,
 88                timeout=self._timeout,
 89                verify=self._verify_ssl,
 90                follow_redirects=self._follow_redirects,
 91                **self._httpx_args,
 92            )
 93        return self._client
 94
 95    def __enter__(self) -> "Client":
 96        """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
 97        self.get_httpx_client().__enter__()
 98        return self
 99
100    def __exit__(self, *args: Any, **kwargs: Any) -> None:
101        """Exit a context manager for internal httpx.Client (see httpx docs)"""
102        self.get_httpx_client().__exit__(*args, **kwargs)
103
104    def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
105        """Manually set the underlying httpx.AsyncClient
106
107        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
108        """
109        self._async_client = async_client
110        return self
111
112    def get_async_httpx_client(self) -> httpx.AsyncClient:
113        """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
114        if self._async_client is None:
115            self._async_client = httpx.AsyncClient(
116                base_url=self._base_url,
117                cookies=self._cookies,
118                headers=self._headers,
119                timeout=self._timeout,
120                verify=self._verify_ssl,
121                follow_redirects=self._follow_redirects,
122                **self._httpx_args,
123            )
124        return self._async_client
125
126    async def __aenter__(self) -> "Client":
127        """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
128        await self.get_async_httpx_client().__aenter__()
129        return self
130
131    async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
132        """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
133        await self.get_async_httpx_client().__aexit__(*args, **kwargs)