threads.py

class _BlockingAsyncContextManager(AbstractContextManager):
    _enter_future: Future
    _exit_future: Future
    _exit_event: Event
    _exit_exc_info: Tuple[Optional[Type[BaseException]], Optional[BaseException],
                          Optional[TracebackType]]

    def __init__(self, async_cm: AsyncContextManager[T_co], portal: 'BlockingPortal'):
        self._async_cm = async_cm
        self._portal = portal

    async def run_async_cm(self):
        try:
            self._exit_event = create_event()
            value = await self._async_cm.__aenter__()
        except BaseException as exc:
            self._enter_future.set_exception(exc)
            raise
        else:
            self._enter_future.set_result(value)

        await self._exit_event.wait()
        return await self._async_cm.__aexit__(*self._exit_exc_info)

    def __enter__(self) -> T_co:
        self._enter_future = Future()
        self._exit_future = self._portal.spawn_task(self.run_async_cm)
        cm = self._enter_future.result()
        return cast(T_co, cm)

    def __exit__(self, __exc_type: Optional[Type[BaseException]],
                 __exc_value: Optional[BaseException],
                 __traceback: Optional[TracebackType]) -> Optional[bool]:
        self._exit_exc_info = __exc_type, __exc_value, __traceback
        self._portal.call(self._exit_event.set)
        return self._exit_future.result()