Skip to content

typerdrive Client modules

typerdrive.client.attach

Provide a decorator that attaches TyperdriveClient instances to a typer command function.

Attributes

Classes

Functions

attach_client 

attach_client(
    **client_urls_or_settings_keys: str,
) -> Callable[
    [ContextFunction[P, T]], ContextFunction[P, T]
]

Attach TyperdriveClient instances to the decorated typer command function.

Parameters:

Name Type Description Default
client_urls_or_settings_keys str

A key/value mapping for base urls to use in the clients. The key will be the name of the client. The value will be either the base url to be used by the client or a key to the settings where the base url can be found.

 {}
Source code in src/typerdrive/client/attach.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def attach_client(**client_urls_or_settings_keys: str) -> Callable[[ContextFunction[P, T]], ContextFunction[P, T]]:
    """
    Attach `TyperdriveClient` instances to the decorated `typer` command function.

    Parameters:
        client_urls_or_settings_keys: A key/value mapping for base urls to use in the clients.
                                      The key will be the name of the client.
                                      The value will be either the base url to be used by the client or a key to the
                                      settings where the base url can be found.
    """

    def _decorate(func: ContextFunction[P, T]) -> ContextFunction[P, T]:
        manager_param_key: str | None = None
        client_param_keys: list[str] = []
        for key in func.__annotations__.keys():
            if func.__annotations__[key] is TyperdriveClient:
                if key in client_urls_or_settings_keys:
                    func.__annotations__[key] = Annotated[TyperdriveClient | None, CloakingDevice]
                    client_param_keys.append(key)

            elif func.__annotations__[key] is ClientManager:
                func.__annotations__[key] = Annotated[ClientManager | None, CloakingDevice]
                manager_param_key = key

        @wraps(func)
        def wrapper(ctx: typer.Context, *args: P.args, **kwargs: P.kwargs) -> T:
            manager = ClientManager()

            settings: BaseModel | None
            try:
                settings = get_settings_manager(ctx).settings_instance
            except SettingsError:
                settings = None

            for name, url_or_settings_key in client_urls_or_settings_keys.items():
                base_url: str = getattr(settings, url_or_settings_key, url_or_settings_key)
                with ClientError.handle_errors(
                    f"Couldn't use {base_url=} for client. If using a settings key, make sure settings are attached."
                ):
                    client_spec = ClientSpec(base_url=base_url)
                manager.add_client(name, client_spec)

            to_context(ctx, "client_manager", manager)

            for key in client_param_keys:
                kwargs[key] = manager.get_client(key)

            if manager_param_key:
                kwargs[manager_param_key] = manager

            return func(ctx, *args, **kwargs)

        return wrapper

    return _decorate

get_client 

get_client(ctx: Context, name: str) -> TyperdriveClient

Retrieve a specific TyperdriveClient from the TyperdriveContext.

Source code in src/typerdrive/client/attach.py 
34
35
36
37
38
def get_client(ctx: typer.Context, name: str) -> TyperdriveClient:
    """
    Retrieve a specific `TyperdriveClient` from the `TyperdriveContext`.
    """
    return get_client_manager(ctx).get_client(name)

get_client_manager 

get_client_manager(ctx: Context) -> ClientManager

Retrieve the ClientManager from the TyperdriveContext.

Source code in src/typerdrive/client/attach.py 
21
22
23
24
25
26
27
28
29
30
31
def get_client_manager(ctx: typer.Context) -> ClientManager:
    """
    Retrieve the `ClientManager` from the `TyperdriveContext`.
    """
    with ClientError.handle_errors("Client(s) are not bound to the context. Use the @attach_client() decorator"):
        mgr: Any = from_context(ctx, "client_manager")
    return ClientError.ensure_type(
        mgr,
        ClientManager,
        "Item in user context at `client_manager` was not a ClientManager",
    )

typerdrive.client.base

Provide a specialized HTTP client for making requests to APIs.

Classes

TyperdriveClient

Bases: Client

Extend the http.Client with *_x() methods that provide useful features for processing requests.

Source code in src/typerdrive/client/base.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class TyperdriveClient(Client):
    """
    Extend the `http.Client` with `*_x()` methods that provide useful features for processing requests.
    """

    def request_x[RM: pydantic.BaseModel](
        self,
        method: str,
        url: URL | str,
        *,
        param_obj: pydantic.BaseModel | None = None,
        body_obj: pydantic.BaseModel | None = None,
        expected_status: int | None = None,
        expect_response: bool = True,
        response_model: type[RM] | None = None,
        **request_kwargs: Any,
    ) -> RM | int | dict[str, Any]:
        """
        Make a request against an API.

        Provides functionality to take url params and request body from instances of `pydantic` models.
        Also, provides checks for the status code returned from the API.
        Will deserialize the response into a `pydantic` model if one is provided.

        Note that all the arguments of `httpx.Client` are also supported.

        Parameters:
            method:           The HTTP method to use in the request
            url:              The url to use for the request. Will be appended to `base_url` if one has been set.
            param_obj:        An optional `pydantic` model to use for url params. This will be serialized to JSON and
                              passed as the request URL parameters.
                              If set, and params are passed through another mechanism as well, an exception will be
                              raised.
            body_obj:         An optional `pydantic` model to use for request body. This will be serialized to JSON and
                              passed as the request body.
                              If set, and the body is passed through another mechanism as well, an exception will be
                              raised.
            expected_status:  If provided, check the response code from the API. If the code doesn't match, raise an
                              exception.
            expect_response:  If set, expect the response to have a JSON body that needs to be deserialized. If not set,
                              just return the status code.
            response_model:   If provided, deserialize the response into an instance of this model. If not provided,
                              the return value will just be a dictionary containing the response data.
        """
        logger.debug(f"Processing {method} request to {self.base_url.join(url)}")

        if param_obj is not None:
            logger.debug(f"Unpacking {param_obj=} to url params")

            ClientError.require_condition(
                "params" not in request_kwargs,
                "'params' not allowed when using param_obj",
            )
            with ClientError.handle_errors("Param data could not be deserialized for http request"):
                request_kwargs["params"] = param_obj.model_dump(mode="json")

        if body_obj is not None:
            logger.debug(f"Unpacking {body_obj=} to request body")

            ClientError.require_condition(
                all(k not in request_kwargs for k in ["data", "json", "content"]),
                "'data', 'json' and 'content' not allowed when using body_obj",
            )
            with ClientError.handle_errors("Request body data could not be deserialized for http request"):
                request_kwargs["content"] = body_obj.model_dump_json()
                request_kwargs["headers"] = {"Content-Type": "application/json"}

        with ClientError.handle_errors(
            "Communication with the API failed",
            handle_exc_class=RequestError,
        ):
            logger.debug("Issuing request")
            response = self.request(method, url, **request_kwargs)

        if expected_status is not None:
            logger.debug(f"Checking response for {expected_status=}")
            ClientError.require_condition(
                expected_status == response.status_code,
                "Got an unexpected status code: Expected {}, got {} -- {}".format(
                    expected_status, response.status_code, response.reason_phrase
                ),
                raise_kwargs=dict(details=response.text),
            )

        if not expect_response:
            logger.debug(f"Skipping response processing due to {expect_response=}")
            return response.status_code

        with ClientError.handle_errors("Failed to unpack JSON from response"):
            logger.debug("Parsing JSON from response")
            data: dict[str, Any] = response.json()

        if not response_model:
            logger.debug("Returning raw data due to no response model being supplied")
            return data

        with ClientError.handle_errors("Unexpected data in response"):
            logger.debug(f"Serializing response as {response_model.__name__}")
            return response_model(**data)

    def get_x[RM: pydantic.BaseModel](
        self,
        url: URL | str,
        *,
        param_obj: pydantic.BaseModel | None = None,
        body_obj: pydantic.BaseModel | None = None,
        expected_status: int | None = None,
        expect_response: bool = True,
        response_model: type[RM] | None = None,
        **request_kwargs: Any,
    ) -> RM | int | dict[str, Any]:
        """
        Make a GET request against an API using the `request_x()` method.
        """
        return self.request_x(
            "GET",
            url,
            param_obj=param_obj,
            body_obj=body_obj,
            expected_status=expected_status,
            expect_response=expect_response,
            response_model=response_model,
            **request_kwargs,
        )

    def post_x[RM: pydantic.BaseModel](
        self,
        url: URL | str,
        *,
        param_obj: pydantic.BaseModel | None = None,
        body_obj: pydantic.BaseModel | None = None,
        expected_status: int | None = None,
        expect_response: bool = True,
        response_model: type[RM] | None = None,
        **request_kwargs: Any,
    ) -> RM | int | dict[str, Any]:
        """
        Make a POST request against an API using the `request_x()` method.
        """
        return self.request_x(
            "POST",
            url,
            param_obj=param_obj,
            body_obj=body_obj,
            expected_status=expected_status,
            expect_response=expect_response,
            response_model=response_model,
            **request_kwargs,
        )

    def put_x[RM: pydantic.BaseModel](
        self,
        url: URL | str,
        *,
        param_obj: pydantic.BaseModel | None = None,
        body_obj: pydantic.BaseModel | None = None,
        expected_status: int | None = None,
        expect_response: bool = True,
        response_model: type[RM] | None = None,
        **request_kwargs: Any,
    ) -> RM | int | dict[str, Any]:
        """
        Make a PUT request against an API using the `request_x()` method.
        """
        return self.request_x(
            "PUT",
            url,
            param_obj=param_obj,
            body_obj=body_obj,
            expected_status=expected_status,
            expect_response=expect_response,
            response_model=response_model,
            **request_kwargs,
        )

    def patch_x[RM: pydantic.BaseModel](
        self,
        url: URL | str,
        *,
        param_obj: pydantic.BaseModel | None = None,
        body_obj: pydantic.BaseModel | None = None,
        expected_status: int | None = None,
        expect_response: bool = True,
        response_model: type[RM] | None = None,
        **request_kwargs: Any,
    ) -> RM | int | dict[str, Any]:
        """
        Make a PATCH request against an API using the `request_x()` method.
        """
        return self.request_x(
            "PATCH",
            url,
            param_obj=param_obj,
            body_obj=body_obj,
            expected_status=expected_status,
            expect_response=expect_response,
            response_model=response_model,
            **request_kwargs,
        )

    def delete_x[RM: pydantic.BaseModel](
        self,
        url: URL | str,
        *,
        param_obj: pydantic.BaseModel | None = None,
        body_obj: pydantic.BaseModel | None = None,
        expected_status: int | None = None,
        expect_response: bool = True,
        response_model: type[RM] | None = None,
        **request_kwargs: Any,
    ) -> RM | int | dict[str, Any]:
        """
        Make a DELETE request against an API using the `request_x()` method.
        """
        return self.request_x(
            "DELETE",
            url,
            param_obj=param_obj,
            body_obj=body_obj,
            expected_status=expected_status,
            expect_response=expect_response,
            response_model=response_model,
            **request_kwargs,
        )
Functions
delete_x 
delete_x(
    url: URL | str,
    *,
    param_obj: BaseModel | None = None,
    body_obj: BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]

Make a DELETE request against an API using the request_x() method.

Source code in src/typerdrive/client/base.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def delete_x[RM: pydantic.BaseModel](
    self,
    url: URL | str,
    *,
    param_obj: pydantic.BaseModel | None = None,
    body_obj: pydantic.BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]:
    """
    Make a DELETE request against an API using the `request_x()` method.
    """
    return self.request_x(
        "DELETE",
        url,
        param_obj=param_obj,
        body_obj=body_obj,
        expected_status=expected_status,
        expect_response=expect_response,
        response_model=response_model,
        **request_kwargs,
    )
get_x 
get_x(
    url: URL | str,
    *,
    param_obj: BaseModel | None = None,
    body_obj: BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]

Make a GET request against an API using the request_x() method.

Source code in src/typerdrive/client/base.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_x[RM: pydantic.BaseModel](
    self,
    url: URL | str,
    *,
    param_obj: pydantic.BaseModel | None = None,
    body_obj: pydantic.BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]:
    """
    Make a GET request against an API using the `request_x()` method.
    """
    return self.request_x(
        "GET",
        url,
        param_obj=param_obj,
        body_obj=body_obj,
        expected_status=expected_status,
        expect_response=expect_response,
        response_model=response_model,
        **request_kwargs,
    )
patch_x 
patch_x(
    url: URL | str,
    *,
    param_obj: BaseModel | None = None,
    body_obj: BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]

Make a PATCH request against an API using the request_x() method.

Source code in src/typerdrive/client/base.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def patch_x[RM: pydantic.BaseModel](
    self,
    url: URL | str,
    *,
    param_obj: pydantic.BaseModel | None = None,
    body_obj: pydantic.BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]:
    """
    Make a PATCH request against an API using the `request_x()` method.
    """
    return self.request_x(
        "PATCH",
        url,
        param_obj=param_obj,
        body_obj=body_obj,
        expected_status=expected_status,
        expect_response=expect_response,
        response_model=response_model,
        **request_kwargs,
    )
post_x 
post_x(
    url: URL | str,
    *,
    param_obj: BaseModel | None = None,
    body_obj: BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]

Make a POST request against an API using the request_x() method.

Source code in src/typerdrive/client/base.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def post_x[RM: pydantic.BaseModel](
    self,
    url: URL | str,
    *,
    param_obj: pydantic.BaseModel | None = None,
    body_obj: pydantic.BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]:
    """
    Make a POST request against an API using the `request_x()` method.
    """
    return self.request_x(
        "POST",
        url,
        param_obj=param_obj,
        body_obj=body_obj,
        expected_status=expected_status,
        expect_response=expect_response,
        response_model=response_model,
        **request_kwargs,
    )
put_x 
put_x(
    url: URL | str,
    *,
    param_obj: BaseModel | None = None,
    body_obj: BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]

Make a PUT request against an API using the request_x() method.

Source code in src/typerdrive/client/base.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def put_x[RM: pydantic.BaseModel](
    self,
    url: URL | str,
    *,
    param_obj: pydantic.BaseModel | None = None,
    body_obj: pydantic.BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]:
    """
    Make a PUT request against an API using the `request_x()` method.
    """
    return self.request_x(
        "PUT",
        url,
        param_obj=param_obj,
        body_obj=body_obj,
        expected_status=expected_status,
        expect_response=expect_response,
        response_model=response_model,
        **request_kwargs,
    )
request_x 
request_x(
    method: str,
    url: URL | str,
    *,
    param_obj: BaseModel | None = None,
    body_obj: BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]

Make a request against an API.

Provides functionality to take url params and request body from instances of pydantic models. Also, provides checks for the status code returned from the API. Will deserialize the response into a pydantic model if one is provided.

Note that all the arguments of httpx.Client are also supported.

Parameters:

Name Type Description Default
method str

The HTTP method to use in the request

 required
url URL | str

The url to use for the request. Will be appended to base_url if one has been set.

 required
param_obj BaseModel | None

An optional pydantic model to use for url params. This will be serialized to JSON and passed as the request URL parameters. If set, and params are passed through another mechanism as well, an exception will be raised.

 None
body_obj BaseModel | None

An optional pydantic model to use for request body. This will be serialized to JSON and passed as the request body. If set, and the body is passed through another mechanism as well, an exception will be raised.

 None
expected_status int | None

If provided, check the response code from the API. If the code doesn't match, raise an exception.

 None
expect_response bool

If set, expect the response to have a JSON body that needs to be deserialized. If not set, just return the status code.

 True
response_model type[RM] | None

If provided, deserialize the response into an instance of this model. If not provided, the return value will just be a dictionary containing the response data.

 None
Source code in src/typerdrive/client/base.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def request_x[RM: pydantic.BaseModel](
    self,
    method: str,
    url: URL | str,
    *,
    param_obj: pydantic.BaseModel | None = None,
    body_obj: pydantic.BaseModel | None = None,
    expected_status: int | None = None,
    expect_response: bool = True,
    response_model: type[RM] | None = None,
    **request_kwargs: Any,
) -> RM | int | dict[str, Any]:
    """
    Make a request against an API.

    Provides functionality to take url params and request body from instances of `pydantic` models.
    Also, provides checks for the status code returned from the API.
    Will deserialize the response into a `pydantic` model if one is provided.

    Note that all the arguments of `httpx.Client` are also supported.

    Parameters:
        method:           The HTTP method to use in the request
        url:              The url to use for the request. Will be appended to `base_url` if one has been set.
        param_obj:        An optional `pydantic` model to use for url params. This will be serialized to JSON and
                          passed as the request URL parameters.
                          If set, and params are passed through another mechanism as well, an exception will be
                          raised.
        body_obj:         An optional `pydantic` model to use for request body. This will be serialized to JSON and
                          passed as the request body.
                          If set, and the body is passed through another mechanism as well, an exception will be
                          raised.
        expected_status:  If provided, check the response code from the API. If the code doesn't match, raise an
                          exception.
        expect_response:  If set, expect the response to have a JSON body that needs to be deserialized. If not set,
                          just return the status code.
        response_model:   If provided, deserialize the response into an instance of this model. If not provided,
                          the return value will just be a dictionary containing the response data.
    """
    logger.debug(f"Processing {method} request to {self.base_url.join(url)}")

    if param_obj is not None:
        logger.debug(f"Unpacking {param_obj=} to url params")

        ClientError.require_condition(
            "params" not in request_kwargs,
            "'params' not allowed when using param_obj",
        )
        with ClientError.handle_errors("Param data could not be deserialized for http request"):
            request_kwargs["params"] = param_obj.model_dump(mode="json")

    if body_obj is not None:
        logger.debug(f"Unpacking {body_obj=} to request body")

        ClientError.require_condition(
            all(k not in request_kwargs for k in ["data", "json", "content"]),
            "'data', 'json' and 'content' not allowed when using body_obj",
        )
        with ClientError.handle_errors("Request body data could not be deserialized for http request"):
            request_kwargs["content"] = body_obj.model_dump_json()
            request_kwargs["headers"] = {"Content-Type": "application/json"}

    with ClientError.handle_errors(
        "Communication with the API failed",
        handle_exc_class=RequestError,
    ):
        logger.debug("Issuing request")
        response = self.request(method, url, **request_kwargs)

    if expected_status is not None:
        logger.debug(f"Checking response for {expected_status=}")
        ClientError.require_condition(
            expected_status == response.status_code,
            "Got an unexpected status code: Expected {}, got {} -- {}".format(
                expected_status, response.status_code, response.reason_phrase
            ),
            raise_kwargs=dict(details=response.text),
        )

    if not expect_response:
        logger.debug(f"Skipping response processing due to {expect_response=}")
        return response.status_code

    with ClientError.handle_errors("Failed to unpack JSON from response"):
        logger.debug("Parsing JSON from response")
        data: dict[str, Any] = response.json()

    if not response_model:
        logger.debug("Returning raw data due to no response model being supplied")
        return data

    with ClientError.handle_errors("Unexpected data in response"):
        logger.debug(f"Serializing response as {response_model.__name__}")
        return response_model(**data)

typerdrive.client.exceptions

Provide exceptions specific to the client feature of typerdrive.

Classes

ClientError

Bases: TyperdriveError

The base exception for client errors.

Source code in src/typerdrive/client/exceptions.py 
 9
10
11
12
13
class ClientError(TyperdriveError):
    """
    The base exception for client errors.
    """
    exit_code: ExitCode = ExitCode.GENERAL_ERROR
Attributes
exit_code class-attribute instance-attribute 
exit_code: ExitCode = GENERAL_ERROR

typerdrive.client.manager

Provide a class for managing the typerdrive client feature.

Classes

ClientManager

Manage instances of TyperdriveClient.

Source code in src/typerdrive/client/manager.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class ClientManager:
    """
    Manage instances of `TyperdriveClient`.
    """

    clients: dict[str, TyperdriveClient]
    """ The `TyperdriveClient instances to manage. """

    def __init__(self):
        self.clients = {}

    def add_client(self, name: str, spec: ClientSpec) -> None:
        """
        Add a `TyperdriveClient` under the given name.

        Parameters:
            name: The name of the client
            spec: A `ClientSpec` describing the client. Used to validate the specs.
        """
        ClientError.require_condition(
            name not in self.clients,
            f"Client with name {name} already exists in context",
        )
        self.clients[name] = TyperdriveClient(base_url=str(spec.base_url))

    def get_client(self, name: str) -> TyperdriveClient:
        """
        Fetch a client from the manager matching the given name.
        """
        return ClientError.enforce_defined(self.clients.get(name), f"No client named {name} found in context")
Attributes
clients instance-attribute 
clients: dict[str, TyperdriveClient] = {}

The `TyperdriveClient instances to manage.

Functions
__init__ 
__init__()
Source code in src/typerdrive/client/manager.py 
30
31
def __init__(self):
    self.clients = {}
add_client 
add_client(name: str, spec: ClientSpec) -> None

Add a TyperdriveClient under the given name.

Parameters:

Name Type Description Default
name str

The name of the client

 required
spec ClientSpec

A ClientSpec describing the client. Used to validate the specs.

 required
Source code in src/typerdrive/client/manager.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def add_client(self, name: str, spec: ClientSpec) -> None:
    """
    Add a `TyperdriveClient` under the given name.

    Parameters:
        name: The name of the client
        spec: A `ClientSpec` describing the client. Used to validate the specs.
    """
    ClientError.require_condition(
        name not in self.clients,
        f"Client with name {name} already exists in context",
    )
    self.clients[name] = TyperdriveClient(base_url=str(spec.base_url))
get_client 
get_client(name: str) -> TyperdriveClient

Fetch a client from the manager matching the given name.

Source code in src/typerdrive/client/manager.py 
47
48
49
50
51
def get_client(self, name: str) -> TyperdriveClient:
    """
    Fetch a client from the manager matching the given name.
    """
    return ClientError.enforce_defined(self.clients.get(name), f"No client named {name} found in context")

ClientSpec

Bases: BaseModel

Source code in src/typerdrive/client/manager.py 
18
19
class ClientSpec(BaseModel):
    base_url: Annotated[str, BeforeValidator(pyright_safe_validator)]
Attributes
base_url instance-attribute 
base_url: Annotated[
    str, BeforeValidator(pyright_safe_validator)
]

Functions

pyright_safe_validator 

pyright_safe_validator(value: str) -> str
Source code in src/typerdrive/client/manager.py 
13
14
15
def pyright_safe_validator(value: str) -> str:
    AnyHttpUrl(value)
    return value