Context manager for ensuring that a request is not made until it is entered and that the response will always be closed when the context manager exits
| 621 | |
| 622 | |
| 623 | class AsyncResponseContextManager(Generic[_AsyncAPIResponseT]): |
| 624 | """Context manager for ensuring that a request is not made |
| 625 | until it is entered and that the response will always be closed |
| 626 | when the context manager exits |
| 627 | """ |
| 628 | |
| 629 | def __init__(self, api_request: Awaitable[_AsyncAPIResponseT]) -> None: |
| 630 | self._api_request = api_request |
| 631 | self.__response: _AsyncAPIResponseT | None = None |
| 632 | |
| 633 | async def __aenter__(self) -> _AsyncAPIResponseT: |
| 634 | self.__response = await self._api_request |
| 635 | return self.__response |
| 636 | |
| 637 | async def __aexit__( |
| 638 | self, |
| 639 | exc_type: type[BaseException] | None, |
| 640 | exc: BaseException | None, |
| 641 | exc_tb: TracebackType | None, |
| 642 | ) -> None: |
| 643 | if self.__response is not None: |
| 644 | await self.__response.close() |
| 645 | |
| 646 | |
| 647 | def to_streamed_response_wrapper(func: Callable[P, R]) -> Callable[P, ResponseContextManager[APIResponse[R]]]: |