Async Iterators dan Generators

Data dari jaringan jarang datang sekaligus: baris log mengalir satu per satu, respons API besar dikirim dalam potongan (chunk), dan kursor database mengembalikan hasil per batch. Perulangan for biasa memaksa kita memuat semuanya dulu, atau memblokir thread sambil menunggu elemen berikutnya. async for menggabungkan keduanya: memproses elemen begitu tersedia sambil menyerahkan kontrol di sela penungguan — konsep yang melanjutkan generator pada Chapter yield dan Generator Functions ke dunia async.

Protokol aiter dan anext #

Di balik async for ada pasangan special method: __aiter__() mengembalikan object iterator, dan __anext__() mengembalikan elemen berikutnya sebagai coroutine. Ketika elemen habis, __anext__ menaikkan StopAsyncIteration — versi async dari mekanisme __iter__/__next__ pada iterator biasa:

import asyncio

class PembacaBaris:
    def __init__(self, baris):
        self._baris = baris
        self._indeks = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self._indeks >= len(self._baris):
            raise StopAsyncIteration
        baris = self._baris[self._indeks]
        self._indeks += 1
        await asyncio.sleep(0.1)
        return baris

async def main():
    async for baris in PembacaBaris(["alpha", "beta", "gamma"]):
        print("menerima:", baris)

asyncio.run(main())

Output yang Dihasilkan:

menerima: alpha
menerima: beta
menerima: gamma

Mari kita telaah apa yang terjadi pada baris-baris kode di atas:

1. Baris 8 (def __aiter__(self):): method ini tidak async — tugasnya sekadar mengembalikan iterator (di sini self), tanpa perlu menunggu apa pun.
2. Baris 11 (async def __anext__(self):): method ini async karena mengambil elemen berikutnya bisa menunggu I/O. await asyncio.sleep(0.1) mensimulasikan jeda kedatangan data.
3. Baris 13 (raise StopAsyncIteration): sinyal resmi bahwa iterator habis; statement async for menangkapnya dan mengakhiri perulangan tanpa error.
4. Baris 20 (async for baris in ...): interpreter memanggil __aiter__ sekali, lalu berulang men-await __anext__ — di setiap await tersebut, task lain di event loop boleh maju.

Async Generator: yield di Dalam async def #

Menulis class lengkap hanya untuk mengalirkan data terasa panjang. Kabar baiknya, keyword yield juga sah di dalam async def — kombinasi ini disebut async generator: cukup yield satu per satu setelah setiap penungguan I/O:

import asyncio

async def unduh_perlahan(ukuran_chunk=4):
    data = "data-berkas-palsu-untuk-demo"
    for indeks in range(0, len(data), ukuran_chunk):
        await asyncio.sleep(0.1)
        yield data[indeks:indeks + ukuran_chunk]

async def main():
    diterima = 0
    async for chunk in unduh_perlahan():
        diterima += len(chunk)
        print(f"menerima {chunk!r} (total {diterima} byte)")
    print("unduhan selesai")

asyncio.run(main())

Output yang Dihasilkan:

menerima 'data' (total 4 byte)
menerima '-ber' (total 8 byte)
menerima 'kas-' (total 12 byte)
menerima 'pals' (total 16 byte)
menerima 'u-un' (total 20 byte)
menerima 'tuk-' (total 24 byte)
menerima 'demo' (total 28 byte)
unduhan selesai

Tidak ada __aiter__ atau __anext__ yang ditulis manual — sebuah async generator function otomatis mengimplementasikan keduanya. await asyncio.sleep(0.1) sebelum tiap yield mensimulasikan latensi jaringan antar-potongan, dan async for mencetak setiap chunk begitu tiba. Program tidak pernah menyimpan seluruh 28 byte sekaligus; memori yang dipakai hanya seukuran satu chunk — sifat penting saat “berkas” sesungguhnya berukuran ratusan megabyte.

Antara dua yield, generator boleh menunggu I/O sebanyak apa pun: membaca batch berikutnya dari database, menunggu socket, atau menunda pemrosesan ketika konsumen lambat. Karena itulah async generator menjadi bahan baku library streaming — kita akan bertemu lagi dengannya saat mengalirkan body respons HTTP pada Chapter Async Network I/O.

Async Comprehension #

List comprehension juga mengenal bentuk async for — berguna untuk mengumpulkan hasil stream menjadi list di satu tempat:

import asyncio

async def unduh_perlahan(ukuran_chunk=4):
    data = "data-berkas-palsu-untuk-demo"
    for indeks in range(0, len(data), ukuran_chunk):
        await asyncio.sleep(0.05)
        yield data[indeks:indeks + ukuran_chunk]

async def main():
    potongan = [chunk async for chunk in unduh_perlahan(ukuran_chunk=10)]
    print(potongan)

asyncio.run(main())

Output yang Dihasilkan:

['data-berka', 's-palsu-un', 'tuk-demo']

Perhatikan bahwa comprehension ini tetap berjalan di dalam coroutine main — async for hanya sah di dalam async def, sama seperti await. Perhatikan juga trade-off-nya: comprehension mengumpulkan seluruh hasil di memori dulu, jadi untuk memproses tiap elemen begitu tiba (misalnya menampilkan progress atau menulis ke disk per-chunk), async for biasa tetap lebih tepat.

Dengan data yang mengalir, satu masalah masih tertinggal: bagaimana menghentikan aliran atau task yang terlalu lama menunggu? Jawabannya ada pada mekanisme pembatalan di Chapter Cancellation dan Timeouts.