如何保留@property特性并按需实现异步数据加载？

先说说你遇到的错误原因：

• asyncio.run()不能在已经运行的事件循环里调用，你的test()函数已经通过asyncio.run()启动了事件循环，在里面再调用asyncio.run()就会触发RuntimeError。
• asyncio.create_task()只是把协程加入事件队列，但不会等待它执行完成，所以show_list返回时mylist还是空的，未被等待的协程会抛出RuntimeWarning。

下面是两种可行的解决方案：

方案一：基于原生API改造同步@property #

直接修改show_list属性，让它在异步上下文里等待加载完成，同时兼容无事件循环的场景：

import asyncio

class A:
    def __init__(self):
        self.mylist = []

    @property
    def show_list(self):
        if not self.mylist:
            try:
                # 获取当前正在运行的事件循环
                loop = asyncio.get_running_loop()
                # 等待加载任务完成
                loop.run_until_complete(self.load_list())
            except RuntimeError:
                # 没有运行中的事件循环，直接启动新循环执行
                asyncio.run(self.load_list())
        return self.mylist

    async def load_list(self):
        await asyncio.sleep(1)
        self.mylist = [1]


async def test():
    f = A()
    print(f.show_list)


asyncio.run(test())

运行这段代码会等待1秒后输出[1]，没有报错和警告。

方案二：改用异步属性（自定义装饰器实现） #

如果你希望属性本身适配异步编程习惯，可以自定义一个异步属性装饰器，通过await获取属性值：

import asyncio

def async_property(func):
    def wrapper(self):
        return func(self)
    return property(wrapper)

class A:
    def __init__(self):
        self.mylist = []

    @async_property
    async def show_list(self):
        if not self.mylist:
            await self.load_list()
        return self.mylist

    async def load_list(self):
        await asyncio.sleep(1)
        self.mylist = [1]


async def test():
    f = A()
    print(await f.show_list)


asyncio.run(test())

这种方式更贴合异步代码的写法，但需要用await来获取属性值，本质是把属性包装成了异步的getter逻辑。

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