MCPcopy Create free account
hub / github.com/LaunchPlatform/beanhub-cli / Client

Class Client

beanhub_cli/internal_api/client.py:13–142  ·  view source on GitHub ↗

A class for keeping track of data related to the API The following are accepted as keyword arguments and will be used to construct httpx Clients internally: ``base_url``: The base URL for the API, all requests are made to a relative path to this URL ``cookies``: A dictionary o

Source from the content-addressed store, hash-verified

11
12@define
13class Client:
14 """A class for keeping track of data related to the API
15
16 The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
17
18 ``base_url``: The base URL for the API, all requests are made to a relative path to this URL
19
20 ``cookies``: A dictionary of cookies to be sent with every request
21
22 ``headers``: A dictionary of headers to be sent with every request
23
24 ``timeout``: The maximum amount of a time a request can take. API functions will raise
25 httpx.TimeoutException if this is exceeded.
26
27 ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
28 but can be set to False for testing purposes.
29
30 ``follow_redirects``: Whether or not to follow redirects. Default value is False.
31
32 ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
33
34
35 Attributes:
36 raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
37 status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
38 argument to the constructor.
39 """
40
41 raise_on_unexpected_status: bool = field(default=False, kw_only=True)
42 _base_url: str = field(alias="base_url")
43 _cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
44 _headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
45 _timeout: Optional[httpx.Timeout] = field(
46 default=None, kw_only=True, alias="timeout"
47 )
48 _verify_ssl: Union[str, bool, ssl.SSLContext] = field(
49 default=True, kw_only=True, alias="verify_ssl"
50 )
51 _follow_redirects: bool = field(
52 default=False, kw_only=True, alias="follow_redirects"
53 )
54 _httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
55 _client: Optional[httpx.Client] = field(default=None, init=False)
56 _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)
57
58 def with_headers(self, headers: dict[str, str]) -> "Client":
59 """Get a new client matching this one with additional headers"""
60 if self._client is not None:
61 self._client.headers.update(headers)
62 if self._async_client is not None:
63 self._async_client.headers.update(headers)
64 return evolve(self, headers={**self._headers, **headers})
65
66 def with_cookies(self, cookies: dict[str, str]) -> "Client":
67 """Get a new client matching this one with additional cookies"""
68 if self._client is not None:
69 self._client.cookies.update(cookies)
70 if self._async_client is not None:

Callers 1

make_clientFunction · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected