如何为asyncio.create_task附加额外数据？附Python示例

针对你提出的问题，这里提供几种无需修改原download_file函数职责的优化方案：

1. 用包装协程绑定关联数据 #

创建一个外层包装协程，负责调用原下载函数并同时返回关联的Foo对象，既保持原函数的单一职责，又能关联任务与数据：

import asyncio
from dataclasses import dataclass

@dataclass
class Foo:
    name: str
    url_to_download: str
    size: int

async def download_file(url: str):
    # 模拟下载逻辑
    return f"content from {url}"

# 新增包装协程，绑定Foo对象与下载任务
async def wrapped_download(foo_obj: Foo):
    download_result = await download_file(foo_obj.url_to_download)
    return foo_obj, download_result

objs = [Foo("file1", "http://example.com/file1", 1024), Foo("file2", "http://example.com/file2", 2048)]

tasks = [asyncio.create_task(wrapped_download(obj)) for obj in objs]

for task in asyncio.as_completed(tasks):
    foo, data = await task
    print(f"处理文件: {foo.name}, 下载内容: {data}")

2. 直接为Task对象附加自定义属性 #

asyncio的Task对象允许添加自定义属性，只需在创建任务后直接绑定Foo对象即可，无需修改任何协程函数：

import asyncio
from dataclasses import dataclass

@dataclass
class Foo:
    name: str
    url_to_download: str
    size: int

async def download_file(url: str):
    return f"content from {url}"

objs = [Foo("file1", "http://example.com/file1", 1024), Foo("file2", "http://example.com/file2", 2048)]

tasks = []
for obj in objs:
    task = asyncio.create_task(download_file(obj.url_to_download))
    # 直接为任务附加Foo对象属性
    task.associated_foo = obj
    tasks.append(task)

for task in asyncio.as_completed(tasks):
    data = await task
    foo = task.associated_foo
    print(f"处理文件: {foo.name}, 下载内容: {data}")

3. 用字典映射任务与Foo对象 #

通过外部字典建立任务到Foo对象的映射关系，完全解耦任务与关联数据：

import asyncio
from dataclasses import dataclass

@dataclass
class Foo:
    name: str
    url_to_download: str
    size: int

async def download_file(url: str):
    return f"content from {url}"

objs = [Foo("file1", "http://example.com/file1", 1024), Foo("file2", "http://example.com/file2", 2048)]

task_foo_map = {}
tasks = []
for obj in objs:
    task = asyncio.create_task(download_file(obj.url_to_download))
    task_foo_map[task] = obj
    tasks.append(task)

for task in asyncio.as_completed(tasks):
    data = await task
    foo = task_foo_map[task]
    print(f"处理文件: {foo.name}, 下载内容: {data}")

方案对比 #

• 包装协程：符合单一职责原则，原下载函数无需改动，适合需要严格分离职责的场景。
• 自定义任务属性：代码改动最小，实现最简洁，适合快速迭代的场景。
• 字典映射：完全解耦任务与数据，避免修改任务对象本身，适合对任务对象有严格约束的场景。

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