Sebuah pipeline membuka banyak resource sepanjang hidupnya: file staging, koneksi database, sesi API. Setiap resource yang terbuka dan lupa ditutup menahan memori, koneksi, atau deskriptor file — dan bila galat terjadi di tengah proses, pembersihan manual sering terlewat. Context manager melalui pernyataan with menjamin kode pembersihan selalu dijalankan, baik saat blok selesai normal maupun saat exception diterbangkan. Kita bahas tiga lapisannya: with bawaan, context manager buatan sendiri, dan decorator contextlib.contextmanager.

with untuk File #

Bentuk paling sering dipakai adalah membuka file dengan with open(...). Ketika blok berakhir, Python menutup file secara otomatis — termasuk memastikan penyangga (buffer) ditulis ke disk lebih dulu.

python 
with open("staging_transaksi.txt", "w") as f:
    f.write("TRX-001 Budi 150000\n")
    f.write("TRX-002 Siti 275000\n")
    f.write("TRX-003 Dewi 90000\n")

with open("staging_transaksi.txt") as f:
    for baris in f:
        id_trx, nama, total = baris.split()
        print(f"{id_trx} pelanggan={nama} total=Rp{total}")

print("status file setelah with:", f.closed)

Output yang Dihasilkan:

text 
TRX-001 pelanggan=Budi total=Rp150000
TRX-002 pelanggan=Siti total=Rp275000
TRX-003 pelanggan=Dewi total=Rp90000
status file setelah with: True

Mari kita telaah baris-baris kode di atas:

  1. Baris 1-4 (with open(..., "w")): variabel f hanya hidup di dalam blok. Saat blok keluar, file ditutup dan disiram ke disk — bahkan bila f.write() di tengah blok menimbulkan exception.
  2. Baris 6-9 (with open(...)): mode baca. Perulangan for baris in f membaca baris demi baris tanpa memuat seluruh isi ke memori — pola lazy dari Chapter Iterators dan Generators.
  3. Baris 11 (f.closed): setelah blok berakhir, atribut closed bernilai True — file sudah tertutup meski variabel f masih bisa diakses.

Context Manager Custom dengan enter dan exit #

Untuk resource yang tidak punya context manager bawaan, kita membuatnya sendiri dengan dua method: __enter__ dipanggil saat masuk blok, __exit__ dipanggil saat keluar blok. Contoh berikut membungkus koneksi database tiruan.

python 
import time


class KoneksiDatabase:
    def __init__(self, dsn):
        self.dsn = dsn
        self.terbuka = False

    def __enter__(self):
        print(f"[OPEN ] koneksi ke {self.dsn}")
        self.terbuka = True
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.terbuka = False
        print(f"[CLOSE] koneksi ke {self.dsn} ditutup")
        return False

    def query(self, sql):
        if not self.terbuka:
            raise RuntimeError("koneksi sudah ditutup")
        return f"hasil untuk: {sql}"


with KoneksiDatabase("postgresql://warehouse.example.com/laporan") as db:
    print(db.query("SELECT kota, SUM(total) FROM transaksi GROUP BY kota"))

print("terbuka setelah blok:", db.terbuka)

# Exception di dalam blok: __exit__ tetap dijalankan
try:
    with KoneksiDatabase("postgresql://warehouse.example.com/laporan") as db2:
        print(db2.query("SELECT 1"))
        raise ValueError("schema tidak cocok")
except ValueError as e:
    print("tertangkap di luar blok:", e)

Output yang Dihasilkan:

text 
[OPEN ] koneksi ke postgresql://warehouse.example.com/laporan
hasil untuk: SELECT kota, SUM(total) FROM transaksi GROUP BY kota
[CLOSE] koneksi ke postgresql://warehouse.example.com/laporan ditutup
terbuka setelah blok: False
[OPEN ] koneksi ke postgresql://warehouse.example.com/laporan
hasil untuk: SELECT 1
[CLOSE] koneksi ke postgresql://warehouse.example.com/laporan ditutup
tertangkap di luar blok: schema tidak cocok

Mari kita telaah baris-baris kode di atas:

  1. Baris 9-12 (__enter__): method ini menandai koneksi terbuka, mencetak log, lalu mengembalikan self sehingga variabel db menerima objek koneksi.
  2. Baris 14-17 (__exit__): method ini menutup koneksi apa pun yang terjadi di dalam blok. Nilai balik False berarti exception (bila ada) tetap diteruskan ke pemanggil.
  3. Baris 25-28: blok with menjalankan __enter__, isi blok, lalu __exit__. Baris db.terbuka bernilai False — koneksi ditutup rapi.
  4. Baris 31-36 (try/except): raise ValueError di dalam blok tetap memicu [CLOSE] lebih dulu, baru kemudian exception naik dan ditangkap di luar blok. Urutan pada output membuktikan pembersihan tidak pernah terlewat.

contextlib.contextmanager untuk Kode Ringkas #

Bila resource-nya sederhana, decorator @contextmanager dari modul contextlib memadatkan dua method tadi menjadi satu generator. Segalanya sebelum yield berperan sebagai __enter__; segalanya setelah yield berperan sebagai __exit__. Contoh berikut memasang timer pada setiap tahap pipeline ETL.

python 
import time
from contextlib import contextmanager


@contextmanager
def ukur_waktu(nama_tahap):
    mulai = time.perf_counter()
    try:
        yield
    finally:
        durasi = time.perf_counter() - mulai
        print(f"[TIMER] {nama_tahap} selesai dalam {durasi:.3f} detik")


def validasi(records):
    time.sleep(0.05)  # simulasi proses I/O
    return [r for r in records if r["total"] > 0]


def transform(records):
    time.sleep(0.05)  # simulasi kalkulasi
    return [{"kota": r["kota"], "total": r["total"] * 11 // 10} for r in records]


def muat(records):
    time.sleep(0.05)  # simulasi INSERT ke warehouse
    return len(records)


pipeline = [
    ("validasi", validasi),
    ("transform", transform),
    ("load", muat),
]

data_mentah = [
    {"kota": "Jakarta", "total": 150000},
    {"kota": "Bandung", "total": 275000},
    {"kota": "Surabaya", "total": -1},  # baris kotor
]

jumlah_baris = 0
for nama, tahap in pipeline:
    with ukur_waktu(nama):
        hasil = tahap(data_mentah)
    if nama == "load":
        jumlah_baris = hasil
    else:
        data_mentah = hasil

print("jumlah baris termuat ke warehouse:", jumlah_baris)

Output yang Dihasilkan:

text 
[TIMER] validasi selesai dalam 0.050 detik
[TIMER] transform selesai dalam 0.050 detik
[TIMER] load selesai dalam 0.050 detik
jumlah baris termuat ke warehouse: 2

Mari kita telaah baris-baris kode di atas:

  1. Baris 5-12 (@contextmanager): yield memisahkan dua fase. Blok finally menjamin pesan timer tercetak meski tahap pipeline menimbulkan exception — pola yang sama dengan __exit__ pada versi class.
  2. Baris 15-27 (validasi, transform, muat): tiga tahap tiruan, masing-masing menunggu 0,05 detik. Tahap validasi membuang baris kotor bernilai negatif.
  3. Baris 42-49 (for nama, tahap in pipeline): setiap tahap dibungkus with ukur_waktu(nama). Timer menyala saat blok dibuka dan melaporkan durasi saat blok ditutup; hasil tahap dialirkan ke tahap berikutnya.

Rujukan resmi untuk pendalaman: dokumentasi contextlib dan spesifikasi pernyataan with . Dengan resource kini selalu dibersihkan rapi, pertanyaan berikutnya adalah kecepatan: bagaimana memproses banyak file atau banyak endpoint sekaligus — pokok bahasan Chapter Concurrent Processing.