Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import typing
5from .._models import Request, Response
6from .base import AsyncBaseTransport, BaseTransport
8SyncHandler = typing.Callable[[Request], Response]
9AsyncHandler = typing.Callable[[Request], typing.Coroutine[None, None, Response]]
12class MockTransport(AsyncBaseTransport, BaseTransport):
13 def __init__(self, handler: SyncHandler | AsyncHandler) -> None:
14 self.handler = handler
16 def handle_request(
17 self,
18 request: Request,
19 ) -> Response:
20 request.read()
21 response = self.handler(request)
22 if not isinstance(response, Response): # pragma: no cover
23 raise TypeError("Cannot use an async handler in a sync Client")
24 return response
26 async def handle_async_request(
27 self,
28 request: Request,
29 ) -> Response:
30 await request.aread()
31 response = self.handler(request)
33 # Allow handler to *optionally* be an `async` function.
34 # If it is, then the `response` variable need to be awaited to actually
35 # return the result.
37 if not isinstance(response, Response):
38 response = await response
40 return response