Tutorial Simple Python Batch Processing

Publish on Jun 29, 2025 by Alfin F
Subscribe dengan Account Google untuk mendapatkan News Letter terbaru dari Halovina !
SUBSCRIBE
Tutorial Simple Python Batch Processing

Bayangkan kamu ingin mengirim 12 paket ke kantor pos, tapi hanya bisa membawa 5 paket sekaligus. Kamu pergi beberapa kali, dan setiap kali, semua paket dalam satu perjalanan dikirim bersamaan.

Setelah semua selesai, kamu cek apakah semua paket sudah sampai dengan baik atau ada yang gagal.

Alur Sederhana



  1. Buat data yang ingin dikirim.

  2. Bagi data menjadi beberapa kelompok kecil (batch).

  3. Untuk setiap batch:

    • Kirim semua data dalam batch ke API secara bersamaan.

    • Simpan hasilnya.



  4. Setelah semua batch selesai, tampilkan hasil pengiriman setiap data.




Solusi Kode Python


(1) Import Library

  • asyncio dan httpx digunakan untuk membuat dan mengirim permintaan HTTP secara paralel/asinkron.

  • typing hanya untuk membantu penulisan tipe data (tidak mempengaruhi logika utama).

import asyncio
import httpx
from typing import List, Dict, Any

(2) Fungsi post_data


  • Fungsi ini bertugas mengirim satu data ke API menggunakan metode POST.

  • Jika berhasil, akan mencetak pesan sukses dan mengembalikan hasilnya.

  • Jika gagal (misal koneksi bermasalah atau server error), akan mencetak pesan error dan mengembalikan info error.

async def post_data(session: httpx.AsyncClient, url: str, data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Sends a single HTTP POST request.
    """
    try:
        response = await session.post(url, json=data, timeout=10)  # 10 seconds timeout
        response.raise_for_status()  # Will raise HTTPStatusError for 4xx/5xx responses
        return {"status": "success", "data": data, "response": response.json()}
    except httpx.RequestError as exc:
        return {"status": "error", "data": data, "error": str(exc)}
    except httpx.HTTPStatusError as exc:
        return {"status": "error", "data": data, "error": f"HTTP Error {exc.response.status_code}: {exc.response.text}"}
    except Exception as exc:
        return {"status": "error", "data": data, "error": str(exc)}

(3) Fungsi process_batch


  • Fungsi ini menerima satu kelompok (batch) data.

  • Semua data dalam batch dikirim ke API secara bersamaan (paralel).

  • Hasil dari semua pengiriman dikumpulkan dan dikembalikan.

async def process_batch(session: httpx.AsyncClient, batch_data: List[Dict[str, Any]], target_url: str) -> List[Dict[str, Any]]:
    """
    Processes a single batch of data asynchronously.
    """
    tasks = [post_data(session, target_url, item_data) for item_data in batch_data]
    results = await asyncio.gather(*tasks)
    return results

(4) Fungsi main


  • Membuat 12 data contoh (dummy) yang akan dikirim ke API.

  • Data dibagi menjadi beberapa batch, tiap batch berisi 5 data.

  • Untuk setiap batch, data dikirim ke API secara paralel.

  • Setelah semua batch selesai, hasilnya ditampilkan satu per satu.

async def main():
    target_url = "https://jsonplaceholder.typicode.com/posts"  # Example API URL
    all_data_to_post = []



    # Create 12 dummy data entries to POST
    for i in range(1, 13):
        all_data_to_post.append({"title": f"Data Title {i}", "body": f"This is the body for data {i}", "userId": 1})



    batch_size = 5
    results_collector = []



    async with httpx.AsyncClient() as client:
        for i in range(0, len(all_data_to_post), batch_size):
            batch = all_data_to_post[i:i + batch_size]
            print(f"\n--- Processing Batch {int(i/batch_size) + 1} (Size: {len(batch)}) ---")
            batch_results = await process_batch(client, batch, target_url)
            results_collector.extend(batch_results)
            print(f"--- Batch {int(i/batch_size) + 1} Finish ---")



    print("\n=== All Batches Have Been Processed ===")
    # You can process results_collector here
    # Example: Display the status of each request
    for res in results_collector:
        print(f"Final Result: {res['status']} for data: {res['data'].get('title', 'N/A')}")

(5) Bagian Akhir

Jika file dijalankan langsung, maka fungsi main akan dieksekusi.

if __name__ == "__main__":
    asyncio.run(main())

Baca artikel lainya :


 

Copyright © 2024 halovina.com

Develop by Halovina Tech