Pemrograman Async Python: Panduan Lengkap

Saat skrip Python Anda dengan sabar menunggu respons API, kueri database, atau operasi berkas selesai, waktu itu sering kali tidak terpakai. Dengan pemrograman async Python, kode Anda dapat menangani beberapa tugas sekaligus. Jadi ketika satu operasi menunggu, yang lain tetap berjalan, mengubah momen menganggur menjadi pekerjaan produktif dan sering kali memangkas waktu tunggu dari menit menjadi hitungan detik.

Dalam panduan ini, saya akan mengajarkan dasar-dasar pemrograman async di Python melalui mini-proyek. Anda akan melihat bagaimana coroutine, event loop, dan async I/O dapat membuat kode Anda jauh lebih responsif.

Jika Anda ingin mempelajari cara membangun web API asinkron, pastikan untuk melihat kursus tentang FastAPI.

Apa itu Pemrograman Async Python?

Dalam Python sinkron tradisional, kode Anda dieksekusi satu baris pada satu waktu. Misalnya, saat Anda memanggil sebuah API, program Anda berhenti dan menunggu respons. Jika itu memakan waktu dua detik, seluruh program Anda menganggur selama dua detik. Pemrograman asinkron memungkinkan kode Anda memulai panggilan API lalu melanjutkan tugas lain.

Saat respons tiba, kode Anda melanjutkan dari tempat semula. Alih-alih menunggu tiap operasi selesai, Anda dapat menjalankan beberapa operasi secara bersamaan. Hal ini paling penting ketika kode Anda menghabiskan waktu menunggu sistem eksternal seperti database, API, atau sistem berkas merespons. 

Agar ini berhasil, sistem async Python menggunakan beberapa konsep inti:

• Coroutine: Fungsi yang didefinisikan dengan async def alih-alih def. Fungsi ini dapat jeda dan dilanjutkan kembali, sehingga ideal untuk operasi yang melibatkan penantian.

• await: Kata kunci yang memberi tahu Python, "jeda coroutine ini sampai operasi ini selesai, tetapi biarkan kode lain berjalan sementara itu."

• Event loop: Mesin yang mengelola semua coroutine Anda, memutuskan mana yang dijalankan dan kapan beralih di antaranya.

• Task: Coroutine yang dibungkus untuk eksekusi bersamaan. Anda membuatnya dengan asyncio.create_task() untuk menjalankan beberapa operasi sekaligus.

Agar tidak bingung tentang apa yang bisa (dan tidak bisa) dilakukan async programming, ingat hal ini:

• Async paling efektif untuk pekerjaan I/O-bound seperti permintaan HTTP, kueri database, dan operasi berkas, saat kode Anda menunggu sistem eksternal.

• Async tidak membantu untuk pekerjaan CPU-bound seperti perhitungan kompleks atau pemrosesan data, saat kode Anda aktif menghitung alih-alih menunggu.

Cara terbaik untuk memahaminya adalah dengan menulis kode async sebenarnya. Pada bagian berikutnya, Anda akan membuat fungsi async pertama Anda dan melihat persis bagaimana coroutine dan event loop bekerja bersama.

Fungsi Async Python Pertama Anda

Sebelum menulis kode async, mari lihat fungsi sinkron biasa yang menunggu sebelum melakukan sesuatu:

import time

def greet_after_delay():
    print("Starting...")
    time.sleep(2)  # Blocks for 2 seconds
    print("Hello!")

greet_after_delay()Didukung oleh

Starting...
Hello!Didukung oleh

Fungsinya bekerja, tetapi time.sleep(2) memblokir seluruh program Anda. Tidak ada yang bisa berjalan selama dua detik itu.

Sekarang berikut versi async-nya:

import asyncio

async def greet_after_delay():
    print("Starting...")
    await asyncio.sleep(2)  # Pauses, but doesn't block
    print("Hello!")

asyncio.run(greet_after_delay())Didukung oleh

Starting...
Hello!Didukung oleh

Keluaran terlihat identik, tetapi ada hal berbeda yang terjadi di balik layar. Tiga perubahan membuatnya async:

1. async def alih-alih def mendeklarasikan ini sebagai coroutine.

2. await asyncio.sleep(2) alih-alih time.sleep(2) menjeda tanpa memblokir.

3. asyncio.run() memulai event loop dan menjalankan coroutine.

Perhatikan bahwa asyncio.sleep() sendiri adalah fungsi async, itulah sebabnya perlu await. Ini aturan kunci: setiap fungsi async harus dipanggil dengan await. Baik itu bawaan seperti asyncio.sleep() atau buatan Anda sendiri, lupa await berarti fungsi itu tidak benar-benar dieksekusi.

Saat ini, versi async belum terlihat lebih cepat. Itu karena kita hanya punya satu tugas. Manfaat sebenarnya muncul saat Anda menjalankan beberapa coroutine sekaligus, yang akan kita bahas di bagian berikut.

Satu hal penting lainnya: Anda tidak bisa memanggil fungsi async langsung seperti fungsi biasa. Mari kita coba:

result = greet_after_delay()
print(result)
print(type(result))Didukung oleh

<coroutine object greet_after_delay at 0x...>
<class 'coroutine'>Didukung oleh

Memanggil greet_after_delay() mengembalikan objek coroutine, bukan hasilnya. Fungsinya tidak benar-benar berjalan. Anda memerlukan asyncio.run() atau await untuk mengeksekusinya di dalam fungsi lain.

Cara kerja event loop

Event loop adalah mesin di balik pemrograman async. Ia mengelola coroutine Anda dan memutuskan apa yang dijalankan kapan. Berikut yang terjadi langkah demi langkah saat Anda menjalankan fungsi async greet_after_delay():

1. asyncio.run() membuat event loop.

2. Event loop memulai greet_after_delay().

3. "Starting..." tercetak.

4. Mencapai await asyncio.sleep(2) → coroutine dijeda.

5. Event loop memeriksa: "Ada tugas lain untuk dijalankan?" (belum ada).

6. 2 detik berlalu, sleep selesai.

7. Event loop melanjutkan greet_after_delay().

8. "Hello!" tercetak.

9. Fungsi selesai → event loop keluar.

Langkah 5 adalah tempat async menjadi menarik. Dengan satu coroutine, tidak ada yang lain untuk dilakukan. Namun saat Anda memiliki banyak coroutine, event loop beralih ke pekerjaan lain sementara yang satu menunggu. Alih-alih menganggur selama tidur dua detik, ia dapat menjalankan kode lain.

Anggap event loop seperti pengatur lalu lintas. Ia tidak membuat mobil individual lebih cepat. Ia menjaga lalu lintas tetap bergerak dengan membiarkan mobil lain jalan saat satu mobil berhenti.

Kesalahan async yang umum: Lupa await

Kesalahan pemula yang umum adalah lupa await saat memanggil coroutine di dalam fungsi async lain:

import asyncio

async def get_message():
    await asyncio.sleep(1)
    return "Hello!"

async def main():
    message = get_message()  # Missing await!
    print(message)

asyncio.run(main())Didukung oleh

<coroutine object get_message at 0x...>
RuntimeWarning: coroutine 'get_message' was never awaitedDidukung oleh

Tanpa await, Anda mendapatkan objek coroutine alih-alih nilai kembalian. Python juga memperingatkan bahwa coroutine tidak pernah dieksekusi.

Perbaikannya sederhana:

async def main():
    message = await get_message()  # Added await
    print(message)

asyncio.run(main())Didukung oleh

Hello!Didukung oleh

Saat Anda melihat RuntimeWarning tentang coroutine yang tidak di-await, periksa bahwa Anda sudah menggunakan await pada setiap pemanggilan fungsi async.

Tugas Async Python yang Berjalan Bersamaan 

Pada bagian sebelumnya, kita mengonversi fungsi sinkron menjadi async. Namun itu tidak lebih cepat. Itu karena kita hanya menjalankan satu coroutine. Kekuatan nyata async muncul saat Anda menjalankan beberapa coroutine pada saat yang sama.

Mengapa await berurutan tetap berurutan

Anda mungkin berpikir memanggil beberapa fungsi async akan otomatis menjalankannya secara bersamaan. Namun lihat apa yang terjadi saat kita memanggil greet_after_delay() tiga kali:

import asyncio
import time

async def greet_after_delay(name):
    print(f"Starting {name}...")
    await asyncio.sleep(2)
    print(f"Hello, {name}!")

async def main():
    start = time.perf_counter()
    
    await greet_after_delay("Alice")
    await greet_after_delay("Bob")
    await greet_after_delay("Charlie")
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())Didukung oleh

Starting Alice...
Hello, Alice!
Starting Bob...
Hello, Bob!
Starting Charlie...
Hello, Charlie!
Total time: 6.01 secondsDidukung oleh

Enam detik untuk tiga tugas berdurasi dua detik. Setiap await menunggu coroutine-nya selesai sebelum pindah ke baris berikutnya. Kodenya async, tetapi berjalan berurutan.

Menjalankan tugas async secara bersamaan dengan asyncio.gather() 

Untuk menjalankan coroutine pada saat yang sama, gunakan asyncio.gather(). Fungsi ini menerima beberapa coroutine dan menjalankannya secara bersamaan:

async def main():
    start = time.perf_counter()
    
    await asyncio.gather(
        greet_after_delay("Alice"),
        greet_after_delay("Bob"),
        greet_after_delay("Charlie"),
    )
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())Didukung oleh

Starting Alice...
Starting Bob...
Starting Charlie...
Hello, Alice!
Hello, Bob!
Hello, Charlie!
Total time: 2.00 secondsDidukung oleh

Dua detik alih-alih enam. Ketiga coroutine dimulai segera, tidur secara bersamaan, dan selesai bersama. Itu peningkatan 3x hanya dengan satu perubahan.

Perhatikan urutan keluaran: ketiga pesan "Starting..." tercetak sebelum pesan "Hello..." mana pun. Ini menunjukkan semua coroutine berjalan dalam jendela dua detik yang sama alih-alih saling menunggu.

asyncio.gather() mengembalikan daftar hasil dalam urutan yang sama dengan urutan Anda meneruskan coroutine. Jika coroutine Anda mengembalikan nilai, Anda dapat menampungnya:

async def fetch_number(n):
    await asyncio.sleep(1)
    return n * 10

async def main():
    results = await asyncio.gather(
        fetch_number(1),
        fetch_number(2),
        fetch_number(3),
    )
    print(results)

asyncio.run(main())Didukung oleh

[10, 20, 30]Didukung oleh

Hasil kembali dalam urutan [10, 20, 30], sesuai urutan coroutine yang diteruskan ke gather().

Permintaan HTTP Async Python dengan aiohttp

Sejauh ini, kita menggunakan asyncio.sleep() untuk mensimulasikan jeda. Sekarang mari buat permintaan HTTP nyata. Anda mungkin ingin menggunakan pustaka requests, tetapi itu tidak cocok di sini. requests bersifat sinkron dan memblokir event loop, menggagalkan tujuan async.

Sebagai gantinya, gunakan aiohttp, klien HTTP async yang dibuat untuk tujuan ini.

Pengenalan aiohttp

Berikut cara mengambil sebuah URL dengan aiohttp:

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    html = await fetch("https://example.com")
    print(f"Fetched {len(html)} characters")

asyncio.run(main())Didukung oleh

Fetched 513 charactersDidukung oleh

Perhatikan dua async with bertingkat. Masing-masing mengelola sumber daya berbeda, dan memahami perannya adalah kunci menggunakan aiohttp dengan benar.

Cara kerja ClientSession di aiohttp

Berikut yang terjadi langkah demi langkah saat Anda membuat permintaan dengan aiohttp:

1. aiohttp.ClientSession() membuat connection pool (awal kosong).

2. session.get(url) memeriksa pool: "Ada koneksi terbuka ke host ini (server situs)?"

3. Jika belum ada, dibuat koneksi TCP baru (protokol dasar untuk mengirim data melalui internet) dan jabat tangan SSL (penyiapan enkripsi untuk HTTPS).

4. Permintaan HTTP dikirim, dan kita menunggu header respons.

5. Objek respons memegang koneksi.

6. await response.text() membaca data body dari jaringan.

7. Keluar dari async with bagian dalam: Koneksi kembali ke pool (tetap terbuka!).

8. Permintaan berikutnya ke host yang sama dibuat, menggunakan ulang koneksi dari pool (melewati langkah 3).

9. Keluar dari async with luar: Semua koneksi dalam pool ditutup.

Langkah 7 dan 8 adalah wawasan kunci. Connection pool menjaga koneksi tetap hidup di antara permintaan. Saat Anda membuat permintaan lain ke host yang sama, ia melewati proses TCP dan jabat tangan SSL sepenuhnya.

Ini penting karena membangun koneksi baru itu lambat. Jabat tangan TCP membutuhkan satu perjalanan bolak-balik ke server. Jabat tangan SSL membutuhkan dua lagi. Bergantung pada latensi, itu 100–300 ms sebelum Anda bahkan mengirim byte data pertama.

Menggunakan sesi bersama untuk semua permintaan

Sekarang Anda bisa melihat mengapa membuat sesi baru untuk setiap permintaan adalah masalah:

# Wrong: new session for each request
async def fetch_bad(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    results = await asyncio.gather(*[fetch_bad(url) for url in urls])Didukung oleh

Setiap pemanggilan fetch_bad() membuat sesi baru dengan pool kosong. Setiap permintaan membayar biaya jabat tangan penuh, meskipun semuanya ke host yang sama.

Solusinya adalah membuat satu sesi dan meneruskannya ke fungsi fetch Anda:

# Right: reuse a single session
async def fetch_good(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[fetch_good(session, url) for url in urls])Didukung oleh

Dengan sesi bersama, permintaan pertama membangun koneksi, dan sembilan permintaan berikutnya menggunakannya kembali. Satu jabat tangan alih-alih sepuluh.

Contoh permintaan HTTP async: Mengambil data Hacker News

Mari praktikkan ini dengan API Hacker News. API ini sempurna untuk mendemonstrasikan perilaku asinkron karena mengambil cerita memerlukan beberapa permintaan. Jika Anda baru bekerja dengan REST API di Python, lihat Python APIs: A Guide to Building and Using APIs untuk konsep dasar.

Struktur API Hacker News:

• https://hacker-news.firebaseio.com/v0/topstories.json mengembalikan daftar ID cerita (hanya angka)

• https://hacker-news.firebaseio.com/v0/item/{id}.json mengembalikan detail untuk satu cerita

Untuk mendapatkan 10 cerita, Anda memerlukan 11 permintaan: satu untuk daftar ID, lalu satu untuk tiap cerita. Di sinilah pemrograman async bersinar.

Pertama, mari lihat apa yang dikembalikan API jika kita mencoba mengambil cerita pertama:

import aiohttp
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def main():
    async with aiohttp.ClientSession() as session:
        # Get top story IDs
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        print(f"Found {len(story_ids)} stories")
        print(f"First 5 IDs: {story_ids[:5]}")
        
        # Fetch first story details
        first_id = story_ids[0]
        async with session.get(f"{HN_API}/item/{first_id}.json") as response:
            story = await response.json()
        
        print(f"\nStory structure:")
        for key, value in story.items():
            print(f"  {key}: {repr(value)[:50]}")

asyncio.run(main())Didukung oleh

Found 500 stories
First 5 IDs: [46051449, 46055298, 46021577, 46053566, 45984864]

Story structure:
  by: 'mikeayles'
  descendants: 22
  id: 46051449
  kids: [46054027, 46053889, 46053275, 46053515, 46053002,
  score: 217
  text: 'I got DOOM running in KiCad by rendering it with 
  time: 1764108815
  title: 'Show HN: KiDoom – Running DOOM on PCB Traces'
  type: 'story'
  url: 'https://www.mikeayles.com/#kidoom'Didukung oleh

API mengembalikan 500 ID cerita, dan setiap cerita memiliki field seperti title, url, score, dan by (penulis). 

Mengambil banyak hasil secara berurutan vs. bersamaan

Sekarang mari ambil 10 cerita secara berurutan:

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        stories = []
        for story_id in story_ids[:10]:
            story = await fetch_story(session, story_id)
            stories.append(story)
        elapsed = time.perf_counter() - start
        
        print(f"Sequential: Fetched {len(stories)} stories in {elapsed:.2f} seconds")

asyncio.run(main())Didukung oleh

Sequential: Fetched 10 stories in 2.41 secondsDidukung oleh

Sekarang mari ambil cerita yang sama secara bersamaan:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        tasks = [fetch_story(session, story_id) for story_id in story_ids[:10]]
        stories = await asyncio.gather(*tasks)
        elapsed = time.perf_counter() - start
        
        print(f"Concurrent: Fetched {len(stories)} stories in {elapsed:.2f} seconds")
        print("\nTop 3 stories:")
        for story in stories[:3]:
            print(f"  - {story.get('title', 'No title')}")

asyncio.run(main())Didukung oleh

Concurrent: Fetched 10 stories in 0.69 seconds

Top 3 stories:
  - Show HN: KiDoom – Running DOOM on PCB Traces
  - AWS is 10x slower than a dedicated server for the same price [video]
  - Surprisingly, Emacs on Android is pretty goodDidukung oleh

Versi bersamaan 3,5x lebih cepat. Alih-alih menunggu tiap permintaan selesai sebelum memulai berikutnya, semua 10 permintaan berjalan pada saat yang sama. Di sinilah pemrograman async memberikan hasil nyata pada I/O jaringan.

Penanganan Error dan Pembatasan Laju pada Async Python

Saat mengambil data secara bersamaan, beberapa hal bisa salah. Anda bisa membebani server dengan terlalu banyak permintaan. Beberapa permintaan bisa menggantung selamanya. Yang lain mungkin langsung gagal. Dan ketika kegagalan terjadi, Anda memerlukan strategi pemulihan.

Bagian ini membahas tiap kekhawatiran sesuai urutan terjadinya: mengontrol berapa banyak permintaan yang keluar, menetapkan batas waktu, menangani kegagalan, dan mencoba ulang saat masuk akal. Jika Anda perlu penyegaran tentang dasar-dasar penanganan pengecualian Python, lihat Exception & Error Handling in Python. Kita akan menggunakan setelan dasar ini sepanjang pembahasan:

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()Didukung oleh

Pembatasan laju dengan semaphore

Pada bagian sebelumnya, kita menembakkan 10 permintaan sekaligus. Itu berjalan baik. Namun bagaimana saat Anda perlu mengambil 500 cerita? Atau meng-crawl 10.000 halaman?

Sebagian besar API menerapkan rate limit. Mereka mungkin mengizinkan 10 permintaan per detik, atau 100 koneksi bersamaan. Jika melampaui batas, Anda akan diblokir, diperlambat, atau dibanned. Bahkan jika API tidak menerapkan batas, menembakkan ribuan permintaan sekaligus bisa membebani sistem Anda sendiri atau server.

Anda butuh cara untuk mengontrol berapa banyak permintaan yang “sedang berlangsung” pada momen tertentu. Itulah peran semaphore.

Semaphore bekerja seperti sistem izin. Bayangkan Anda memiliki tiga izin. Tugas mana pun yang ingin membuat permintaan harus terlebih dahulu mendapatkan izin. Saat selesai, ia mengembalikan izin, sehingga permintaan baru bisa menggunakannya. Jika tidak ada izin yang tersedia, tugas menunggu sampai ada yang bebas.

Begini jalannya dengan 3 izin dan 4 atau lebih tugas:

1. Tersedia tiga izin.

2. Tugas A mengambil satu izin (tersisa 2), memulai permintaan.

3. Tugas B mengambil satu izin (tersisa 1), memulai permintaan.

4. Tugas C mengambil satu izin (tersisa 0), memulai permintaan.

5. Tugas D ingin izin, tetapi tidak ada yang tersedia—ia menunggu.

6. Tugas A selesai, mengembalikan izinnya (tersedia 1).

7. Tugas D mengambil izin tersebut dan memulai permintaan.

8. Ini berlanjut hingga semua tugas selesai.

Penungguan pada langkah 5 efisien. Tugas tidak berputar dalam loop memeriksa "apakah ada izin yang bebas?". Ia disuspensi dan membiarkan kode lain berjalan. Event loop membangunkannya hanya saat izin tersedia.

Sekarang lihat kodenya. Dalam asyncio, Anda membuat semaphore dengan asyncio.Semaphore(n), di mana n adalah jumlah izin. Untuk menggunakannya, bungkus kode Anda dalam async with semaphore:. Ini memperoleh izin saat memasuki blok dan otomatis melepaskannya saat keluar:

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:  # Acquire permit (or wait if none available)
        async with session.get(f"{HN_API}/item/{story_id}.json") as response:
            return await response.json()
    # Permit automatically released hereDidukung oleh

Mari bandingkan mengambil 30 cerita dengan dan tanpa semaphore:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = (await response.json())[:30]

        # Without rate limiting: all 30 at once
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story(session, sid) for sid in story_ids])
        print(f"No limit: {time.perf_counter() - start:.2f}s (30 concurrent)")

        # With Semaphore(5): max 5 at a time
        semaphore = asyncio.Semaphore(5)
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story_limited(session, sid, semaphore) for sid in story_ids])
        print(f"Semaphore(5): {time.perf_counter() - start:.2f}s (5 concurrent)")

asyncio.run(main())Didukung oleh

No limit: 0.62s (30 concurrent)
Semaphore(5): 1.50s (5 concurrent)Didukung oleh

Versi semaphore lebih lambat karena memproses permintaan dalam batch berisi lima. Namun itulah konsekuensinya: Anda mengorbankan kecepatan demi perilaku yang lebih dapat diprediksi dan ramah server.

Satu hal yang perlu dicatat: semaphore membatasi permintaan yang bersamaan, bukan permintaan per satuan waktu. Semaphore(10) berarti "maksimal 10 permintaan sedang berlangsung sekaligus," bukan "10 permintaan per detik." Jika Anda membutuhkan pembatasan berbasis waktu yang ketat (seperti tepat 10 permintaan per detik), Anda bisa menggabungkan semaphore dengan jeda antar batch, atau gunakan pustaka seperti aiolimiter.

Timeout dengan asyncio.wait_for()

Bahkan dengan konkurensi terkontrol, permintaan individual bisa menggantung. Server mungkin menerima koneksi Anda tetapi tidak pernah merespons. Tanpa timeout, program Anda menunggu tanpa batas.

Fungsi asyncio.wait_for() membungkus coroutine apa pun dengan tenggat waktu. Anda memberinya coroutine dan timeout dalam detik. Jika operasi tidak selesai tepat waktu, fungsi ini melempar asyncio.TimeoutError:

async def slow_operation():
    print("Starting slow operation...")
    await asyncio.sleep(5)
    return "Done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"Success: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out after 2 seconds")

asyncio.run(main())Didukung oleh

Starting slow operation...
Operation timed out after 2 secondsDidukung oleh

Saat timeout habis, wait_for() membatalkan coroutine. Anda bisa menangkap TimeoutError dan memutuskan apa yang harus dilakukan: lewati permintaan, kembalikan nilai bawaan, atau coba lagi.

Untuk permintaan bersamaan, bungkus masing-masing secara individual. Berikut helper yang mengembalikan kamus error alih-alih melempar error:

async def fetch_story_with_timeout(session, story_id, timeout=5.0):
    try:
        coro = fetch_story(session, story_id)
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": f"Story {story_id} timed out"}Didukung oleh

Saat coroutine dibatalkan (oleh timeout atau alasan lain), Python melempar asyncio.CancelledError di dalamnya. Jika coroutine Anda memegang sumber daya seperti file handle atau koneksi, gunakan try/finally untuk memastikan pembersihan tetap terjadi bahkan saat pembatalan:

async def fetch_with_cleanup(session, url):
    print("Starting fetch...")
    try:
        async with session.get(url) as response:
            return await response.text()
    finally:
        print("Cleanup complete")  # Runs even on cancellationDidukung oleh

Penanganan error dengan asyncio.gather()

Timeout menangkap permintaan yang lambat. Namun beberapa permintaan gagal seketika dengan error. Mari lihat apa yang terjadi saat satu permintaan dalam satu batch gagal.

Pertama, kita butuh versi fetch_story() yang melempar pengecualian pada ID tidak valid:

async def fetch_story_strict(session, story_id):
    story = await fetch_story(session, story_id)
    if story is None:
        raise ValueError(f"Story not found: {story_id}")
    return storyDidukung oleh

Sekarang mari ambil empat cerita valid plus satu ID tidak valid:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]  # 4 valid + 1 invalid

        try:
            stories = await asyncio.gather(
                *[fetch_story_strict(session, sid) for sid in ids_to_fetch]
            )
            print(f"Got {len(stories)} stories")
        except ValueError as e:
            print(f"ERROR: {e}")

asyncio.run(main())Didukung oleh

ERROR: Story not found: 99999999999Didukung oleh

Dengan satu ID tidak valid, kita kehilangan keempat hasil yang berhasil. Secara bawaan, gather() menggunakan perilaku fail-fast: satu pengecualian membatalkan semuanya dan meneruskan ke atas.

Untuk mempertahankan hasil parsial, tambahkan return_exceptions=True. Ini mengubah perilaku gather(): alih-alih melempar pengecualian, ia mengembalikannya sebagai item dalam daftar hasil bersama nilai yang berhasil:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]

        results = await asyncio.gather(
            *[fetch_story_strict(session, sid) for sid in ids_to_fetch],
            return_exceptions=True  # Don't raise, return exceptions in list
        )

        # Separate successes from failures using isinstance()
        stories = [r for r in results if not isinstance(r, Exception)]
        errors = [r for r in results if isinstance(r, Exception)]

        print(f"Got {len(stories)} stories, {len(errors)} failed")

asyncio.run(main())Didukung oleh

Got 4 stories, 1 failedDidukung oleh

Pemeriksaan isinstance(result, Exception) memungkinkan Anda memisahkan hasil yang berhasil dari error. Anda kemudian dapat memproses yang berhasil dan mencatat atau mencoba lagi yang gagal.

Logika retry dengan exponential backoff

Beberapa kegagalan bersifat sementara. Server mungkin kelebihan beban sesaat, atau gangguan jaringan menjatuhkan koneksi Anda. Untuk kasus ini, mencoba ulang masuk akal.

Namun mencoba ulang segera dapat memperburuk keadaan. Jika server sedang kesulitan, membombardirnya dengan retry akan menambah masalah. Exponential backoff menyelesaikan ini dengan menunggu lebih lama di antara tiap percobaan.

Pola ini menggunakan 2 ** attempt untuk menghitung waktu tunggu: percobaan 0 menunggu satu detik (2⁰), percobaan 1 menunggu dua detik (2¹), percobaan 2 menunggu empat detik (2²), dan seterusnya. Ini memberi server waktu pemulihan yang makin panjang:

async def fetch_with_retry(session, story_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            story = await fetch_story(session, story_id)
            if story is None:
                raise ValueError(f"Story {story_id} not found")
            return story
        except (aiohttp.ClientError, ValueError):  # Catch specific exceptions
            if attempt == max_retries - 1:
                print(f"Story {story_id}: Failed after {max_retries} attempts")
                return None

            backoff = 2 ** attempt  # 1s, 2s, 4s...
            print(f"Story {story_id}: Attempt {attempt + 1} failed, retrying in {backoff}s...")
            await asyncio.sleep(backoff)Didukung oleh

Perhatikan kita menangkap pengecualian spesifik (aiohttp.ClientError, ValueError) alih-alih except kosong. Ini memastikan kita hanya mencoba ulang pada error yang mungkin sementara. KeyError dari kode yang buruk tidak boleh memicu retry.

Mari uji dengan campuran ID valid dan tidak valid:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        test_ids = [story_ids[0], 99999999999, story_ids[1], 88888888888, story_ids[2]]

        results = await asyncio.gather(*[fetch_with_retry(session, sid) for sid in test_ids])

        successful = [r for r in results if r is not None]
        print(f"\nSuccessful: {len(successful)}, Failed: {len(test_ids) - len(successful)}")

asyncio.run(main())Didukung oleh

Story 99999999999: Attempt 1 failed, retrying in 1s...
Story 88888888888: Attempt 1 failed, retrying in 1s...
Story 99999999999: Attempt 2 failed, retrying in 2s...
Story 88888888888: Attempt 2 failed, retrying in 2s...
Story 99999999999: Failed after 3 attempts
Story 88888888888: Failed after 3 attempts

Successful: 3, Failed: 2Didukung oleh

Di produksi, Anda juga akan menambahkan jitter (penundaan acak kecil) untuk mencegah beberapa permintaan gagal melakukan retry pada saat yang persis sama. Selain itu, Anda hanya akan mencoba ulang error sementara (masalah jaringan sisi server, seperti 503) sambil langsung menyerah pada error permanen (misalnya, 404 atau 401).

Penyimpanan Database Async Python dengan aiosqlite

Kita telah mengambil cerita Hacker News dengan pembatasan laju yang tepat, timeout, dan penanganan error. Sekarang mari simpan ke database.

Menggunakan pustaka database sinkron biasa seperti sqlite3 akan memblokir event loop saat kueri, menggagalkan tujuan pemrograman asinkron. Saat kode Anda menunggu database, tidak ada coroutine lain yang bisa berjalan. Untuk aplikasi async, Anda memerlukan pustaka database async.

aiosqlite membungkus sqlite3 bawaan Python dalam antarmuka async. Ia menjalankan operasi database di thread pool sehingga tidak memblokir event loop. SQLite tidak memerlukan penyiapan server—hanya sebuah berkas—jadi Anda bisa langsung menjalankan kode ini. Jika Anda baru bekerja dengan database di Python, kursus Introduction to Databases in Python membahas dasar sinkron yang menjadi landasan aiosqlite.

Menyiapkan database

Polanya seharusnya terlihat familier. Layaknya aiohttp.ClientSession, Anda menggunakan async with untuk mengelola koneksi:

import aiosqlite

async def init_db(db_path):
    async with aiosqlite.connect(db_path) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.commit()

asyncio.run(init_db("stories.db"))Didukung oleh

Fungsi-fungsi kunci:

• aiosqlite.connect(path) membuka (atau membuat) berkas database.

• await db.execute(sql) menjalankan pernyataan SQL.

• await db.commit() menyimpan perubahan ke disk.

Menyimpan cerita

Berikut fungsi untuk menyimpan satu cerita:

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )Didukung oleh

Placeholder ? mencegah injeksi SQL—jangan pernah menggunakan f-string untuk memasukkan nilai ke SQL. INSERT OR REPLACE memperbarui cerita yang sudah ada jika kita mengambilnya lagi.

Pipeline Async Lengkap di Python: Ambil dan Simpan

Sekarang mari gabungkan semua dari tutorial ini menjadi pipeline lengkap. Kita akan mengambil 20 cerita Hacker News dengan pembatasan laju dan menyimpannya ke database:

import aiohttp
import aiosqlite
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:
        story = await fetch_story(session, story_id)
        if story:
            return story
        return None

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )

async def main():
    # Initialize database
    async with aiosqlite.connect("hn_stories.db") as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # Fetch stories
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{HN_API}/topstories.json") as response:
                story_ids = await response.json()
            
            semaphore = asyncio.Semaphore(5)
            tasks = [fetch_story_limited(session, sid, semaphore) for sid in story_ids[:20]]
            stories = await asyncio.gather(*tasks)
        
        # Save to database
        for story in stories:
            if story:
                await save_story(db, story)
        await db.commit()
        
        # Query and display
        cursor = await db.execute("SELECT id, title, score FROM stories ORDER BY score DESC LIMIT 5")
        rows = await cursor.fetchall()
        
        print(f"Saved {len([s for s in stories if s])} stories. Top 5 by score:")
        for row in rows:
            print(f"  [{row[2]}] {row[1][:50]}")

asyncio.run(main())Didukung oleh

Saved 20 stories. Top 5 by score:
  [671] Google Antigravity exfiltrates data via indirect p
  [453] Trillions spent and big software projects are stil
  [319] Ilya Sutskever: We're moving from the age of scali
  [311] Show HN: We built an open source, zero webhooks pa
  [306] FLUX.2: Frontier Visual IntelligenceDidukung oleh

Pipeline ini menggunakan pola dari setiap bagian: ClientSession untuk connection pooling, Semaphore(5) untuk pembatasan laju, gather() untuk pengambilan bersamaan, dan sekarang aiosqlite untuk penyimpanan async. Tiap komponen menangani bagiannya tanpa memblokir yang lain.

Setiap kali Anda menjalankan alur kerja ini, Anda akan menerima cerita teratas hari ini.

Kesimpulan

Tutorial ini membawa Anda dari sintaks async/await dasar hingga pipeline data lengkap. Anda mempelajari bagaimana coroutine jeda dan dilanjutkan, bagaimana event loop mengelola tugas bersamaan, dan bagaimana asyncio.gather() menjalankan beberapa operasi sekaligus. Anda menambahkan permintaan HTTP nyata dengan aiohttp, mengontrol konkurensi dengan semaphore, menangani kegagalan dengan timeout dan retry, serta menyimpan hasil ke database dengan aiosqlite.

Gunakan async saat kode Anda menunggu sistem eksternal: HTTP API, database, I/O berkas, atau soket jaringan. Untuk pekerjaan berat CPU seperti pemrosesan data atau perhitungan angka, async tidak akan membantu—lihat multiprocessing atau concurrent.futures sebagai gantinya. Untuk melangkah lebih jauh, Anda dapat menjelajahi dokumentasi asyncio dan mempertimbangkan FastAPI untuk membangun web API async. 

Jika Anda ingin membangun di atas pengetahuan ini dan belajar merancang aplikasi cerdas, pastikan untuk melihat jalur karier Associate AI Engineer for Developers.