如何通过__await__魔术方法封装线程/进程以创建可等待类？

我希望创建**可等待（awaitable）**类实例，这在使用没有asyncio包装器的数据库连接器时会很有用，同时也有助于加深对asyncio和生成器（generators）的理解。我提出了以下两种解决方案：

方案1 #

import asyncio
import threading
import time


async def ticker():
    """Just some kind of async ticker function."""
    while True:
        await asyncio.sleep(1)
        print("tick tack sleep...")


class ThreadAwaitableWithToThread:
    """Illustrates how to wrap Thread in awaitable object."""
    def do_work(self):
        """Some target function for another Thread."""
        cur_th_name = threading.current_thread().getName()
        for i in range(5):
            print(f"Do work-{i} in {cur_th_name}")
            time.sleep(2)

        return "well done result!"

    def __await__(self):
        """Magic method to create awaitable object."""
        # to_thread is wrapper around run_in_executor
        result = yield from asyncio.to_thread(self.do_work).__await__()
        # The same without asyncio.to_thread:
        # result = yield from loop.run_in_executor(None, self.do_work).__await__()
        return result


async def amain():
    """Main asyncio entrypoint of the app."""
    t = asyncio.create_task(ticker())  # "daemon" task
    res = await ThreadAwaitableWithToThread()
    t.cancel()  # cancel "daemon" task
    print(f"Result: {res}")

if __name__ == '__main__':
    asyncio.run(amain())

该方案基于loop.run_in_executor，可能是最优解，但我希望找到基于基础生成器而非asyncio库功能的实现方式。

方案2 #

import asyncio
import threading
import time


async def ticker():
    """Just some kind of async ticker function."""
    while True:
        await asyncio.sleep(1)
        print("tick tack sleep...")


class ThreadAwaitable:
    """Illustrates how to wrap Thread in awaitable object."""

    def __init__(self):
        self.done = threading.Event()  # sync between this and another thread

    def do_work(self):
        """Some target function for another Thread."""
        cur_th_name = threading.current_thread().getName()
        for i in range(5):
            print(f"Do work-{i} in {cur_th_name}")
            time.sleep(2)

        self.done.set()

    def __await__(self):
        """Magic method to create awaitable object."""
        th = threading.Thread(target=self.do_work)
        th.start()  # start another tread
        while True:
            if self.done.is_set():  # sync with another thread
                print(f"Work in {self.__class__.__name__} is done!")
                return "well done result!"

            # make loop switch between tasks while waiting another Thread to finish work.
            yield from asyncio.sleep(0.1).__await__()


async def amain():
    """Main asyncio entrypoint of the app."""
    t = asyncio.create_task(ticker())  # "daemon" task
    res = await ThreadAwaitable()
    t.cancel()  # cancel "daemon" task
    print(f"Result: {res}")


if __name__ == '__main__':
    asyncio.run(amain())

该方案基于借助asyncio.sleep实现的长轮询，会消耗额外资源。

我希望找到更优雅的解决方案，了解到select、selectors库可监听套接字，若将线程/进程与套接字绑定，可在内部通过套接字切换协程/生成器，但相关库文档不够详细，且这相当于自行实现asyncio（类似David Beazley在Pycon 2015的分享）。请问是否有更好的解决方案？

内容的提问来源于stack exchange，提问作者Artiom Kozyrev