使用装饰器封装asyncio.create_task的代码能否正常运行？求建议

核心问题与修正点 #

• 函数名拼写错误：原代码中装饰的函数是printe_task，但调用时使用printer_task()，会触发NameError，需统一函数名。
• 缺失必要导入：代码中用到的wraps（来自functools）、create_task/sleep（来自asyncio）都需要显式导入。
• 变量名冲突：原代码用task作为Task对象的变量名，与装饰器同名，会覆盖装饰器导致后续无法复用，需修改变量名避免冲突。

修正后的完整代码 #

from functools import wraps
import asyncio
from typing import Any, Task

def task(func: Any):
    @wraps(func)
    async def wrapped(*args: Any) -> Task[Any]:
        return asyncio.create_task(func(*args))
    return wrapped

# 修正函数名拼写，与调用处保持一致
@task
async def printer_task():
    await asyncio.sleep(1)
    print("Hello")

async def printer():
    await asyncio.sleep(1)
    print("Hello")

async def main():
    # 原逻辑：创建Task后await
    task_obj = asyncio.create_task(printer())
    await task_obj

    # 与上述逻辑完全等价的调用
    await printer_task()

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

效果说明 #

修正后，await printer_task()的运行逻辑和create_task(printer())后await完全一致：两者都会把异步函数提交到事件循环中作为独立Task执行，等待任务完成后输出结果。

内容的提问来源于stack exchange，提问作者yEmreAk.com