Python调用run_in_executor报coroutine was never awaited如何解决？

你遇到的RuntimeWarning: coroutine 'Generator.generate_token' was never awaited核心原因是协程函数被直接传给run_in_executor执行，没有被await调度。run_in_executor的设计用途是在独立线程/进程中运行同步阻塞函数，它不会自动识别、调度协程对象，传入协程函数只会生成一个未被执行的协程实例，既不会运行内部逻辑，还会因为协程从未被等待抛出警告。

除此之外你的代码还有4处显性错误：

• 协程滥用await：attributes.append()、Pillow的convert()/save()都是普通同步方法，不是可等待对象，前面加await会直接抛TypeError: object NoneType can't be used in 'await' expression错误。
• 迭代器使用错误：dict.items()返回的是普通同步迭代器，不支持async for遍历，直接用普通for即可。
• 语法缩进错误：generate_token方法里的return attributes缩进错位写到了函数外部，会直接触发语法错误。
• 任务等待逻辑缺失：原来的循环里逐个await run_in_executor的返回值，会导致线程池任务串行执行，完全丧失并发能力。

根据你的业务场景（核心是Pillow图片处理，属于CPU密集型任务），优先选第一种改法，性能最好。

方案1：全同步逻辑+线程池/进程池（推荐） #

协程模型适合IO密集型场景，CPU密集的图片处理用同步逻辑丢到线程池/进程池跑性能更高，完全没必要给图片处理方法硬加async/await：

import os
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from PIL import Image
from typing import List, Dict

class Generator:
    MEDIA_ROOT = "path/to/your/media_dir" # 替换为实际静态资源路径

    @classmethod
    def paste_image(cls, base_img: Image, overlay_img: Image) -> Image:
        # 这里放你原来的图片粘贴同步逻辑，不要加async
        base_img.paste(overlay_img, (0,0), mask=overlay_img)
        return base_img

    @classmethod
    def generate_token(cls, iter: int, combination: Dict) -> List[Dict]:
        # 去掉所有无效async/await，全同步执行
        combination_vals = list(combination.values())
        first_layer = Image.open(os.path.join(cls.MEDIA_ROOT, combination_vals[0]))
        attributes = []
        layer_items = list(combination.items())[1:]

        for layer, attribute in layer_items:
            overlay = Image.open(os.path.join(cls.MEDIA_ROOT, attribute)).convert('RGBA')
            result = cls.paste_image(first_layer, overlay)
            attributes.append({"trait_type": layer, "value": attribute})
        
        # 同步保存图片
        os.makedirs("output", exist_ok=True)
        result.convert("RGB").save(
            os.path.join("output", f"{iter}.jpg"),
            "JPEG",
            optimize=True
        )
        return attributes

    @classmethod
    async def generate_tokens(cls, loop, combinations: List[Dict], collection: Dict) -> Dict:
        # CPU密集任务建议替换为ProcessPoolExecutor规避GIL限制，性能更高
        with ThreadPoolExecutor(max_workers=os.cpu_count()*4) as executor:
            task_list = [
                loop.run_in_executor(executor, cls.generate_token, iter, comb)
                for iter, comb in enumerate(combinations, 1)
            ]
            # 并发等待所有任务完成
            await asyncio.gather(*task_list)
        return {"success": True, "message": "Generation completed"}

    @classmethod
    def generate_combinations(cls, traits: Dict, id: int) -> Dict:
        # 原有组合生成逻辑
        completed_combinations = [] # 替换为实际生成的组合列表
        collection = {} # 替换为实际集合配置

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            result = loop.run_until_complete(
                cls.generate_tokens(loop, completed_combinations, collection)
            )
        finally:
            loop.close()
        return result

方案2：保留协程逻辑（仅适合有异步IO需求的场景） #

如果你确实需要在generate_token里执行异步IO操作（比如异步读写文件、异步请求外部接口），不要把协程丢进线程池，直接用asyncio的任务调度实现并发即可：

@classmethod
    async def generate_tokens(cls, combinations: List[Dict], collection: Dict) -> Dict:
        task_list = [
            asyncio.create_task(cls.generate_token(iter, comb))
            for iter, comb in enumerate(combinations, 1)
        ]
        await asyncio.gather(*task_list)
        return {"success": True, "message": "Generation completed"}

注意：如果generate_token里全是Pillow的CPU计算逻辑，不要用这个方案——单线程协程跑CPU密集任务会阻塞整个事件循环，并发性能远低于多进程/多线程方案。

排查这类协程未等待警告的通用思路：

1. 所有async def定义的函数，调用时必须要么加await，要么包装成asyncio Task交给事件循环调度
2. 不要把协程对象/协程函数传给只支持同步逻辑的入口（比如线程池执行器、同步回调函数）
3. 不要给普通同步方法、普通同步迭代器乱加await、async for等异步语法，避免逻辑中途抛错导致协程无法被正常调度

内容的提问来源于stack exchange，提问作者Никита Опалько