创建兼具同步与异步迭代器特性的Python类是否安全？

是否安全？ #

这取决于具体使用方式：

• 如果每次迭代都创建新实例（比如你示例里的async for elem in Iter()和for elem in Iter()），短期看是“安全”的——每个实例的状态self.i独立，同步和异步迭代不会互相干扰。
• 但如果复用同一个实例（比如把iter_obj = Iter()赋值给变量，同时用for和async for迭代它），绝对不安全。同步的__next__和异步的__anext__会共享self.i状态，两者的操作会互相覆盖计数，导致迭代顺序混乱、提前终止或重复输出元素，这类bug在并发场景下更难排查。

是否属于不良编程实践？ #

是的，这种设计属于不良实践，核心原因有三点：

1. 违反单一职责原则：一个类同时处理同步、异步两种迭代逻辑，职责模糊，后续维护时很难理清两种迭代的边界，代码可读性差。
2. 隐藏的bug风险：哪怕你现在确保每次迭代都用新实例，后续其他开发者（甚至未来的你）可能不小心复用实例，引入难以复现的状态冲突bug。
3. 不符合Python设计惯例：同步迭代器（__iter__+__next__）和异步迭代器（__aiter__+__anext__）是两种独立协议，Python社区惯例是分开实现，让使用者明确知道该用哪种方式迭代，避免混淆。

改进方案 #

最合理的做法是拆分同步和异步迭代器，或把核心数据逻辑抽离，让两类迭代器各自负责对应协议：

方案1：拆分独立类

import asyncio

class SyncIter:
    def __init__(self):
        self.i = 0
        self.elems = list(range(10))
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.i >= len(self.elems):
            raise StopIteration
        self.i += 1
        return self.elems[self.i - 1]

class AsyncIter:
    def __init__(self):
        self.i = 0
        self.elems = list(range(10))
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.i >= len(self.elems):
            raise StopAsyncIteration
        self.i += 1
        return self.elems[self.i - 1]

async def main():
    print("async usage:")
    async for elem in AsyncIter():
        print(elem)
    print("sync usage:")
    for elem in SyncIter():
        print(elem)

try:
    asyncio.run(main())
except RuntimeError:
    await main()

方案2：复用核心逻辑

如果同步和异步迭代的核心逻辑一致（比如从同一数据源取数），可以把数据逻辑抽成单独类，让迭代器只负责处理迭代协议：

import asyncio

class DataStore:
    def __init__(self):
        self.items = list(range(10))
    
    def get_item(self, index):
        if index >= len(self.items):
            return None
        return self.items[index]

class SyncIterator:
    def __init__(self, store):
        self.store = store
        self.current_idx = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        item = self.store.get_item(self.current_idx)
        if item is None:
            raise StopIteration
        self.current_idx += 1
        return item

class AsyncIterator:
    def __init__(self, store):
        self.store = store
        self.current_idx = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        # 这里可加入真正的异步操作，比如异步IO取数
        item = self.store.get_item(self.current_idx)
        if item is None:
            raise StopAsyncIteration
        self.current_idx += 1
        return item

async def main():
    data_store = DataStore()
    print("async usage:")
    async for elem in AsyncIterator(data_store):
        print(elem)
    print("sync usage:")
    for elem in SyncIterator(data_store):
        print(elem)

try:
    asyncio.run(main())
except RuntimeError:
    await main()

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