File tetap menjadi sarana tukar data paling umum di dunia nyata: sistem POS toko online mengekspor CSV tiap malam, mitra pengiriman menaruh laporan harian di folder bersama, dan aplikasi lama hanya bisa berbicara lewat file. Sayangnya file datang tanpa kontrak yang ketat — nama bisa berubah-ubah, kolom bisa tertukar, dan file bisa terbaca saat penulisannya belum selesai. Karena itu ingestion dari file tidak berdiri di satu skrip baca-tulis saja, melainkan pada disiplin folder, pola nama, dan validasi yang menjaga pipeline tetap aman.

Pola Folder Landing #

Pola dasarnya memisahkan peran produsen dan konsumen lewat folder. Produsen — sistem eksternal atau job ekspor — hanya menulis ke folder landing/. Konsumen, yaitu job ingestion kita, memindai landing/, memproses file yang sah, lalu memindahkannya ke processed/ agar tidak dibaca dua kali. File yang gagal validasi dibiarkan di landing/ supaya operator bisa memeriksa dan mengirim ulang.

Pemisahan folder sederhana ini sekaligus menjadi catatan keadaan: isi landing/ adalah antrean yang belum selesai, isi processed/ adalah riwayat yang sudah aman.

Mengenali File dengan Pola Nama #

Konsumen tidak boleh membaca sembarang file. Modul pathlib menyediakan glob() untuk menyaring berdasarkan pola nama: "transaksi_*.csv" hanya menangkap file CSV yang namanya diawali transaksi_. File laporan lain — katalog_produk.csv, catatan sementara, apapun — otomatis diabaikan. Menambahkan sorted() memastikan urutan pemrosesan stabil dari nama terkecil, yang pada nama berformat tanggal berarti dari tanggal terlama.

Validasi Dasar #

Sebelum dibaca penuh, setiap file diperiksa header-nya: baris pertama harus memuat kolom wajib id, pelanggan, dan total. Pemeriksaan ini murah — hanya satu baris dibaca — tetapi menangkap dua kejadian paling sering: file kosong hasil ekspor gagal, dan file dengan format kolom berubah tanpa pemberitahuan. Validasi yang lebih kaya (tipe data, rentang nilai) kita bahas tuntas di chapter Validasi Data saat Ingestion.

Menangani File Setengah Jadi #

Bahaya terbesar folder bersama: konsumen membaca file yang sedang ditulis produsen. Solusinya adalah pola tulis atomik: produsen menulis ke nama sementara berakhiran .tmp, lalu mengganti nama ke nama final hanya setelah seluruh isi selesai ditulis. Fungsi os.replace() menjamin penggantian nama terjadi atomik — pembaca tidak pernah melihat keadaan setengah jadi.

python 
import os
from pathlib import Path


def tulis_atomik(tujuan, isi):
    sementara = tujuan.with_name(tujuan.name + ".tmp")
    sementara.write_text(isi)
    os.replace(sementara, tujuan)


landing = Path("landing")
print("Sebelum:", sorted(p.name for p in landing.glob("transaksi_*")))
tulis_atomik(
    landing / "transaksi_2026-10-08.csv",
    "id,pelanggan,total\n1,Budi Santoso,75000\n",
)
print("Sesudah:", sorted(p.name for p in landing.glob("transaksi_*")))
print("File .tmp tersisa:", sorted(p.name for p in landing.glob("*.tmp")))

Output yang Dihasilkan:

text 
Sebelum: []
Sesudah: ['transaksi_2026-10-08.csv']
File .tmp tersisa: []

Pola .tmp pada sisi produsen berpasangan dengan pola glob "transaksi_*.csv" pada sisi konsumen: file sementara tidak pernah cocok dengan pola konsumen, sehingga hanya nama final yang bisa masuk pipeline. Rincian os.replace() ada di dokumentasi os .

Implementasi: Konsumen Folder Landing #

Kita rakit seluruh pola di atas menjadi satu konsumen. Folder landing/ berisi empat file: dua file transaksi sah, satu file .tmp yang sedang ditulis produsen lain, dan satu file ber-header salah (id,pelanggan,harga — kolom total hilang):

python 
import csv
import shutil
from pathlib import Path

FOLDER_LANDING = Path("landing")
FOLDER_PROCESSED = Path("processed")
KOLOM_WAJIB = {"id", "pelanggan", "total"}


def validasi_header(berkas):
    with berkas.open(newline="") as f:
        kolom = set(csv.DictReader(f).fieldnames or [])
    return kolom >= KOLOM_WAJIB


def proses_berkas():
    FOLDER_PROCESSED.mkdir(exist_ok=True)
    total_baris = 0
    for berkas in sorted(FOLDER_LANDING.glob("transaksi_*.csv")):
        if not validasi_header(berkas):
            print(f"{berkas.name}: DITOLAK (header tidak lengkap)")
            continue
        with berkas.open(newline="") as f:
            baris = list(csv.DictReader(f))
        total_baris += len(baris)
        shutil.move(berkas, FOLDER_PROCESSED / berkas.name)
        print(f"{berkas.name}: {len(baris)} baris -> processed/")
    print(f"Total baris diproses: {total_baris}")


proses_berkas()

Output yang Dihasilkan:

text 
transaksi_2026-10-04.csv: 3 baris -> processed/
transaksi_2026-10-05.csv: 2 baris -> processed/
transaksi_2026-10-07.csv: DITOLAK (header tidak lengkap)
Total baris diproses: 5

Mari kita telaah baris-baris kode di atas:

  1. Baris 7 (KOLOM_WAJIB): kontrak kolom dideklarasikan sebagai himpunan — satu tempat yang perlu diperbarui ketika format file berubah.
  2. Baris 10-13 (validasi_header): DictReader membaca header tanpa memuat isi file; fieldnames or [] menjaga file kosong tidak memicu galat, dan pembanding himpunan >= lolos meski file punya kolom tambahan.
  3. Baris 19 (sorted(...glob(...))): pola transaksi_*.csv membuat transaksi_2026-10-06.csv.tmp terlewati tanpa perlu pengecekan khusus — file setengah jadi memang bukan urusan konsumen.
  4. Baris 20-22 (penolakan): file dengan header salah dilaporkan eksplisit dan ditinggal di landing/; penolakan yang terdengar (di log) lebih baik daripada kegagalan senyap.
  5. Baris 26 (shutil.move): pemindahan ke processed/ hanya terjadi setelah seluruh baris selesai dibaca — file yang gagal di tengah proses tetap utuh di tempatnya.

Dua file sah berpindah ke processed/ dengan lima baris total, satu file ditolak dengan alasan jelas, dan file .tmp tidak pernah tersentuh. Pola folder ini bekerja untuk produsen apa pun yang bisa menulis file — tetapi ketika sumbernya database relasional, kita bisa menarik data jauh lebih presisi lewat SELECT, seperti dibahas di chapter Ingestion dari Database.