asyncio.wait执行后如何识别已完成协程对应的自定义RandomLife实例

该需求完全可以实现，核心是手动建立Task和RandomLife实例的映射关系。

你直接将RandomLife实例传入asyncio.wait时，asyncio会自动将这些可等待对象包装为独立的Task实例，默认不会和原RandomLife对象建立关联，因此无法直接从返回的Task反查对应的实例，需要我们手动维护映射。

• 提前为每个RandomLife实例手动创建Task，不要交给asyncio.wait自动生成
• 建立Task对象到RandomLife实例的映射关系，可以用字典存储，也可以给Task加自定义属性
• 拿到完成的Task后，通过映射关系取出对应的RandomLife实例即可

import random
import asyncio

class RandomLife(object):
    def __init__(self, name: str):
        self.name = name
        self.coro = asyncio.sleep(random.randrange(0, 5))

    def __await__(self):
        return self.coro.__await__()


async def main():
    objects = [RandomLife("one"), RandomLife("two"), RandomLife("three")]
    # 手动创建Task并建立映射字典
    task_map = {}
    tasks = []
    for obj in objects:
        task = asyncio.create_task(obj)
        task_map[task] = obj
        tasks.append(task)

    finished, unfinished = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    # 直接通过映射获取对应实例
    for task in finished:
        finished_obj = task_map[task]
        print(f"首个完成的实例名称：{finished_obj.name}")

    await asyncio.wait(unfinished)

if __name__ == "__main__":
    asyncio.run(main())

如果不想额外维护映射字典，也可以直接给Task对象添加自定义属性存储对应实例，注意不要和asyncio.Task的内置属性重名即可：

for obj in objects:
    task = asyncio.create_task(obj)
    task.bound_life_obj = obj
    tasks.append(task)

# 后续直接从task取属性即可
for task in finished:
    print(task.bound_life_obj.name)

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