Telethon异步批量发送消息如何将普通列表转为异步可迭代对象

Python 中要让对象支持 async for 语法，只需让对象满足异步可迭代协议即可，下面是两种常用的实现方式：

方案1：异步生成器封装（最简实现） #

异步生成器原生支持异步迭代协议，只需少量代码就能完成封装：

import asyncio

async def async_list_wrapper(lst: list):
    for item in lst:
        # 可选：加该行让出事件循环，避免同步遍历阻塞事件循环
        # await asyncio.sleep(0)
        yield item

使用示例：

messages = [(id_1, msg_1), (id_2, msg_2), (id_3, msg_3)]

async def send_task():
    async for msg in async_list_wrapper(messages):
        receiver_id, msg_content = msg
        # 执行你的异步发送逻辑，示例：
        # await send_message(receiver_id, msg_content)
        print(f"发送消息给{receiver_id}: {msg_content}")

asyncio.run(send_task())

方案2：自定义异步迭代类 #

如果需要更灵活的控制逻辑，可以手动实现异步迭代协议的两个魔法方法：

• __aiter__：返回异步迭代器对象本身
• __anext__：异步返回下一个元素，遍历结束抛出StopAsyncIteration

class AsyncList:
    def __init__(self, origin_list: list):
        self._data = origin_list
        self._index = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self._index >= len(self._data):
            raise StopAsyncIteration
        current_item = self._data[self._index]
        self._index += 1
        return current_item

使用示例：

messages = [(id_1, msg_1), (id_2, msg_2), (id_3, msg_3)]

async def send_task():
    async for msg in AsyncList(messages):
        # 执行发送逻辑
        pass

SQLite 场景优化建议 #

如果你是从 SQLite 读取数据，不需要先把全量数据读到普通列表再封装，直接使用异步 SQLite 驱动aiosqlite即可，它的查询结果原生支持异步迭代，还能避免全量数据加载占用内存：

import aiosqlite

async def iter_messages_from_db():
    async with aiosqlite.connect("你的数据库路径.db") as conn:
        async with conn.execute("SELECT receiver_id, content FROM messages") as cursor:
            async for receiver_id, content in cursor:
                yield (receiver_id, content)

# 直接遍历数据库查询结果
async for msg in iter_messages_from_db():
    # 发送逻辑
    pass

内容的提问来源于stack exchange，提问作者Joao Pedro Lourenco Affonso