Python'da özel context manager yazmak

Selamlar, bu yazımda Python'ın en zarif özelliklerinden biri olan context manager'lara, özellikle de kendi context manager'larınızı nasıl yazacağınıza bakacağız. with open(...) kalıbını hepimiz biliyoruz; ama aynı mantığı veritabanı bağlantıları, kilitler, zamanlayıcılar ve genel olarak setup-teardown isteyen her kaynak için kullanabilirsiniz. Hadi dalalım.

Context manager neden lazım?#

Açıkça söyleyeyim, eğer kodunuzda her yerde try/finally tekrar eden satırlar varsa, orada büyük ihtimalle gizli bir context manager bekliyor. with bloğu üç şeyi bir arada veriyor: garantili temizlik (exception olsa bile), tekrar kullanılabilir kalıp ve kaynak kapsamını net görsel olarak işaretleyen bir blok. Bu üçü olmadan kod hızla 'try'lı, dağınık' bir hale geliyor.

Buradan ne çıkarmalıyız? Aynı setup-teardown desenini iki yerde gördüğünüzde durup düşünün; muhtemelen bir context manager'a kavuşturmanın zamanı gelmiştir.

Sınıf tabanlı yaklaşım#

En esnek yol, __enter__ ve __exit__ metotlarını içeren bir sınıf yazmak. Şablonu şöyle:

class OrnekContext:
    def __enter__(self):
        # Hazirlik kodu burada
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Temizlik kodu burada, her durumda calisir
        # True donerseniz exception bastirilir, False ile yukari iletilir
        return False

__enter__'ın döndürdüğü değer as ile bağlanan değişken olur. __exit__ ise bloktan çıkarken çalışır; eğer içeride exception fırladıysa parametreler dolu gelir, fırlamadıysa hepsi None olur.

Şimdi bir veritabanı bağlantısı örneği görelim. Aşağıdaki sınıf, blok normal biterse commit, exception olursa rollback çağırıyor:

import sqlite3
from typing import Optional


class DbConnection:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[sqlite3.Connection] = None
        self.cursor: Optional[sqlite3.Cursor] = None

    def __enter__(self) -> sqlite3.Cursor:
        self.connection = sqlite3.connect(self.db_path)
        self.cursor = self.connection.cursor()
        return self.cursor

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        if exc_type is not None:
            self.connection.rollback()
        else:
            self.connection.commit()
        self.cursor.close()
        self.connection.close()
        return False


with DbConnection('app.db') as cursor:
    cursor.execute('INSERT INTO users VALUES (1, ?)', ('Ayse',))

Kullanırken sadece with yazıyoruz; commit, rollback ve close'u tek tek hatırlamak zorunda değiliz. Bence bu yaklaşımın en güzel tarafı bu: tek bir yerde doğru yaptıktan sonra her yerde doğru.

contextlib ile daha kısa yol#

Her seferinde sınıf yazmak istemeyebilirsiniz. Bu durumda contextlib.contextmanager dekoratörü işinizi görür. Generator fonksiyonu yazıyorsunuz; yield öncesi setup, sonrası teardown:

from contextlib import contextmanager
import os


@contextmanager
def working_directory(path: str):
    original_dir = os.getcwd()
    try:
        os.chdir(path)
        yield path
    finally:
        os.chdir(original_dir)


with working_directory('/tmp'):
    # Burada gecici olarak /tmp icindeyiz
    pass

Buradaki kritik nokta try/finally. yield'i çıplak bırakırsanız, blok içinde exception fırladığında temizlik kodunuz çalışmaz. Şahsi kanaatim, dekoratör yaklaşımı sadece basit setup-teardown için; durum tutmanız gerekiyorsa sınıfa geçin.

Birden fazla context'i birlikte kullanmak#

Python 3.10 ile parantez sözdizimi geldi:

with (
    open('input.txt') as infile,
    open('output.txt', 'w') as outfile,
):
    outfile.write(infile.read())

Sayı dinamikse ExitStack daha pratik:

from contextlib import ExitStack


def process_files(filenames: list):
    with ExitStack() as stack:
        files = [stack.enter_context(open(fname)) for fname in filenames]
        for f in files:
            print(f.readline())

ExitStack her dosyayı otomatik kapatır, sıraya da kendisi karar verir.

Sık karşılaşılan hatalar#

• Generator'da try/finally unutmak: yield'den sonraki temizlik kodu, exception olursa çalışmaz. finally bloğuna almazsanız sızıntı kaçınılmazdır.
• __exit__'te True döndürmek: Bu, blok içindeki tüm exception'ları sessizce yutar. Hata ayıklamayı imkansız hale getirir. Sadece gerçekten 'bu exception'ı buradan öteye iletme' demek istiyorsanız kullanın, o da çoğu zaman istemediğiniz şeydir.
• __enter__'da exception fırlatmak ama temizliği yapmamak: Eğer __enter__ ortasında patlarsa __exit__ çağrılmaz. Bu yüzden __enter__ içinde de kendi try/except'inizle yarım kalmış kaynakları kapatın.
• Aynı context'i birden fazla kez girmek: Sınıfı yeniden girilebilir (reentrant) yazmadıysanız ikinci with durumu bozar. Gerekiyorsa threading.RLock benzeri bir yapıya bakın.

Kapanış#

Bu yazıda hem sınıf tabanlı hem de contextlib ile context manager yazmaya baktık. Bana sorarsanız, kodunuzdaki her tekrar eden try/finally deseni aslında 'beni bir context manager'a çevir' diye bağırıyor. Umarım faydalı olur, bir sonraki yazıda görüşmek üzere.