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
| 597 | |
| 598 | |
| 599 | class ResponseContextManager(Generic[_APIResponseT]): |
| 600 | """Context manager for ensuring that a request is not made |
| 601 | until it is entered and that the response will always be closed |
| 602 | when the context manager exits |
| 603 | """ |
| 604 | |
| 605 | def __init__(self, request_func: Callable[[], _APIResponseT]) -> None: |
| 606 | self._request_func = request_func |
| 607 | self.__response: _APIResponseT | None = None |
| 608 | |
| 609 | def __enter__(self) -> _APIResponseT: |
| 610 | self.__response = self._request_func() |
| 611 | return self.__response |
| 612 | |
| 613 | def __exit__( |
| 614 | self, |
| 615 | exc_type: type[BaseException] | None, |
| 616 | exc: BaseException | None, |
| 617 | exc_tb: TracebackType | None, |
| 618 | ) -> None: |
| 619 | if self.__response is not None: |
| 620 | self.__response.close() |
| 621 | |
| 622 | |
| 623 | class AsyncResponseContextManager(Generic[_AsyncAPIResponseT]): |