Python 簡易自建迭代器

說明

大家都知道 for 迴圈，相信也很常看到下列這段程式

for i in range(5):
    print(i)

輸出
0
1
2
3
4

這次要分享如何實作類似 range() 的物件

基本的迭代器

A.py

class A:
    def __init__(self):
        self.x = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.x > 2:
            raise StopIteration
        else:
            self.x += 1

            return self.x

• __init__(): 這個是比較常見的初始化，像是建構子那樣
• __iter__(): 回傳可迭代的物件，這邊讓它回傳自己
• __next__(): 就像是每一次迭代條件式和運算式(例如常見的 i++)
• StopIteration: 代表迴圈結束

最後使用 for 迴圈跑 A 類別結果如下

for i in A():
    print(i)

輸出
1
2
3

非同步迭代器

在《Python 非同步設計使用 Asyncio》書中，他有舉例一個使用 redis 的非同迭代器，為了方便執行，我拿掉 redis 連線等一系列的動作改成 sleep 來模擬非同步迭代器

OneAtATime.py

import asyncio

class OneAtATime:
    def __init__(self, keys):
        self.keys = keys

    def __aiter__(self):
        self.ikeys = iter(self.keys)

        return self

    async def __anext__(self):
        try:
            key = next(self.ikeys)
        except StopIteration:
            raise StopAsyncIteration

        await asyncio.sleep(1)

        return key

• __aiter__: 與 __iter__() 一樣，但是它是用於非同步迭代器
• __anext__(): 與 __next__() 一樣，但是它是用於非同步迭代器
• iter(): 用來存放迭代的 key
• next(): 取得 iter() 存放的 key
• StopAsyncIteration: 用來通知非同步迭代器已結束，可以看到上面先捕捉 StopIteration，再拋出 StopAsyncIteration
• 需注意 def __aiter__() 沒有加 async，而 async def __anext__() 有加 async

使用範例

import asyncio

async def main():
    await asyncio.sleep(1)
    keys = ['1', '2', '3', '4', '5',]

    async for value in OneAtATime(keys):
        print(value)

asyncio.run(main())

輸出
1
2
3
4
5

非同步程式我自己還在學習當中，還不太清楚該如何活用，上面這個例子只能用來分享有非同步迭代器這個東西，在《Python 非同步設計使用 Asyncio》書中，作者有提到更簡單的寫法，但這並非迭代器的議題，以後有機會我再分享非同步程式設計給大家。