Dekoratörler (decorators) ve bağlam yöneticileri (context managers)

Neden önemli?

Ilgaz Bank'ın işlem servisinde on farklı fonksiyon var: para transferi, bakiye sorgusu, kredi başvurusu... Hepsinde aynı üç şey isteniyor: çağrı loglansın, süresi ölçülsün, geçici hata olursa yeniden denensin. Bu üç davranışı on fonksiyona tek tek kopyalamak hem tekrar hem bakım kâbusudur. Dekoratör (decorator), bir fonksiyonu alıp etrafına davranış ekleyen ve yeni fonksiyon döndüren bir araçtır; @zamanla yazarsın, iş biter. Bağlam yöneticisi (context manager) ise aynı fikrin "bir blok öncesi ve sonrası" sürümüdür: dosyayı aç ve ne olursa olsun kapat, bağlantıyı kur ve bırak, işlemi başlat ve hata olursa geri al. İkisi de önceki dersteki closure ve "fonksiyon bir değerdir" fikrinin üstüne kuruludur.

Fonksiyonu saran fonksiyon

Dekoratör mekanizmasını önce elle kur. Bir fonksiyon alan, içinde onu çağıran yeni bir fonksiyon tanımlayan ve bu yeni fonksiyonu döndüren bir fonksiyon yaz.

def logla(fonk):
    def sarici(*args, **kwargs):
        print(f"-> {fonk.__name__} çağrıldı, args={args}")
        sonuc = fonk(*args, **kwargs)
        print(f"<- {fonk.__name__} bitti, sonuç={sonuc}")
        return sonuc
    return sarici

def transfer(kaynak, hedef, tutar):
    return f"{kaynak}->{hedef}: {tutar}"

transfer = logla(transfer)      # elle sarmalama
print(transfer("TR1", "TR2", 500))

Üç parça: sarici her türlü argümanı kabul etsin diye *args, **kwargs alır; asıl fonksiyonu çağırıp sonucunu döndürür (unutulursa dekore edilen fonksiyon None vermeye başlar); logla sarıcıyı döndürür. transfer = logla(transfer) satırı, ismi yeni fonksiyona bağlar. Python bu satırı yazmanın kısa yolunu sunar:

def logla(fonk):
    def sarici(*args, **kwargs):
        print(f"-> {fonk.__name__}")
        return fonk(*args, **kwargs)
    return sarici

@logla
def bakiye_sorgu(hesap):
    return 12500

print(bakiye_sorgu("TR1"))
print(bakiye_sorgu.__name__)   # sarici! asıl adı kayboldu

@logla, tam olarak bakiye_sorgu = logla(bakiye_sorgu) demektir; def biter bitmez uygulanır. Son satır bir sorunu gösteriyor: dekore edilen fonksiyonun adı ve docstring'i sarıcınınkiyle değişti. Hata izlerinde, yardım metinlerinde ve testlerde bu kafa karıştırır.

functools.wraps: kimliği koru

from functools import wraps

def logla(fonk):
    @wraps(fonk)
    def sarici(*args, **kwargs):
        print(f"-> {fonk.__name__}")
        return fonk(*args, **kwargs)
    return sarici

@logla
def bakiye_sorgu(hesap):
    """Hesabın güncel bakiyesini döndürür."""
    return 12500

print(bakiye_sorgu.__name__)
print(bakiye_sorgu.__doc__)
print(bakiye_sorgu.__wrapped__("TR1"))   # dekoratörsüz asıl fonksiyon

Parametreli dekoratör: üç katman

"3 kez yeniden dene" gibi ayar alan dekoratör için bir katman daha gerekir: dıştaki fonksiyon ayarı alır ve asıl dekoratörü döndürür. @yeniden_dene(3) önce yeniden_dene(3)'ü çağırır, dönen dekoratör fonksiyona uygulanır.

from functools import wraps

def yeniden_dene(deneme):
    def dekorator(fonk):
        @wraps(fonk)
        def sarici(*args, **kwargs):
            son_hata = None
            for i in range(1, deneme + 1):
                try:
                    return fonk(*args, **kwargs)
                except ConnectionError as e:
                    son_hata = e
                    print(f"  deneme {i} başarısız: {e}")
            raise son_hata
        return sarici
    return dekorator

sayac = {"cagri": 0}

@yeniden_dene(3)
def banka_servisi():
    sayac["cagri"] += 1
    if sayac["cagri"] < 3:
        raise ConnectionError("zaman aşımı")
    return "OK"

print(banka_servisi())
print("toplam çağrı:", sayac["cagri"])

Katmanları ezberleme, oku: ayar → dekoratör → sarıcı. Sarıcı asıl işi yapar; dekoratör sarıcıyı üretir; en dıştaki katman ayarı closure'a hapseder. Gerçek sistemlerde denemeler arasına time.sleep ile artan bekleme konur; burada çıktıyı sade tutmak için atlandı.

İş hayatından dört dekoratör

Zamanlama

import time
from functools import wraps

def zamanla(fonk):
    @wraps(fonk)
    def sarici(*args, **kwargs):
        t0 = time.perf_counter()
        try:
            return fonk(*args, **kwargs)
        finally:
            sure = time.perf_counter() - t0
            print(f"{fonk.__name__}: {sure:.4f} sn")
    return sarici

@zamanla
def toplam_ciro(siparisler):
    return sum(s["tutar"] for s in siparisler)

veri = [{"tutar": i % 500} for i in range(200_000)]
print(toplam_ciro(veri))

Ölçülen süre her çalıştırmada farklı çıkar; bu doğaldır. finally sayesinde fonksiyon hata fırlatsa da süre basılır. time.perf_counter, süre ölçümü için time.time'dan daha hassastır.

Önbellek: functools.lru_cache

Aynı argümanla tekrar tekrar çağrılan pahalı bir fonksiyonun sonucunu hatırlamak için dekoratör yazmana gerek yok; standart kütüphanede hazır.

from functools import lru_cache

hesap_sayisi = {"n": 0}

@lru_cache(maxsize=128)
def kur_sorgula(para_birimi):
    hesap_sayisi["n"] += 1       # "pahalı" servis çağrısını temsil eder
    return {"USD": 32.5, "EUR": 35.1}[para_birimi]

for pb in ["USD", "EUR", "USD", "USD", "EUR"]:
    kur_sorgula(pb)
print("gerçek hesap:", hesap_sayisi["n"])
print(kur_sorgula.cache_info())
kur_sorgula.cache_clear()

@lru_cache(maxsize=None)
def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)
print(fib(80))

Loglama ve yetki kontrolü

Dekoratörler üst üste konabilir; en yakındaki önce uygulanır. Aşağıda yetki_gerekli önce transfer'i sarar, sonra logla o sarılmış hali sarar.

from functools import wraps

def logla(fonk):
    @wraps(fonk)
    def sarici(*args, **kwargs):
        print(f"[LOG] {fonk.__name__} {kwargs}")
        return fonk(*args, **kwargs)
    return sarici

def yetki_gerekli(fonk):
    @wraps(fonk)
    def sarici(*args, kullanici, **kwargs):
        if kullanici.get("rol") != "yonetici":
            raise PermissionError(f"{kullanici['ad']} yetkisiz")
        return fonk(*args, kullanici=kullanici, **kwargs)
    return sarici

@logla
@yetki_gerekli
def transfer(tutar, *, kullanici):
    return f"{tutar} TL gönderildi"

print(transfer(500, kullanici={"ad": "Ayşe", "rol": "yonetici"}))
try:
    transfer(500, kullanici={"ad": "Ali", "rol": "stajyer"})
except PermissionError as e:
    print("Hata:", e)

with ve bağlam yöneticileri

with open(...) as f: yazdığında Python iki şey garanti eder: bloğa girerken __enter__ çağrılır (dosya açılır, dönen değer f olur), bloktan çıkarken, hata olsa bile, __exit__ çağrılır (dosya kapanır). Bu "kur-kullan-temizle" kalıbı dosyadan çok daha geniş: veritabanı işlemi, kilit, geçici ayar, zaman ölçümü.

contextlib.contextmanager: üreteçten bağlam yöneticisi

En kısa yol, tek yield içeren bir üreteç (generator) fonksiyonunu @contextmanager ile süslemek. yield'den öncesi girişte, sonrası çıkışta çalışır; yield ettiğin değer as ile alınır.

from contextlib import contextmanager

@contextmanager
def banka_islemi(hesap):
    print("işlem başladı, bakiye:", hesap["bakiye"])
    yedek = hesap["bakiye"]
    try:
        yield hesap
        print("işlem onaylandı, bakiye:", hesap["bakiye"])
    except Exception as e:
        hesap["bakiye"] = yedek
        print(f"hata ({e}); geri alındı, bakiye: {hesap['bakiye']}")

hesap = {"bakiye": 1000}

with banka_islemi(hesap) as h:
    h["bakiye"] -= 300

with banka_islemi(hesap) as h:
    h["bakiye"] -= 900
    if h["bakiye"] < 0:
        raise ValueError("yetersiz bakiye")

print("son bakiye:", hesap["bakiye"])

try/except yield'i sarar; blok içinde fırlayan hata, üreteçte tam yield satırında ortaya çıkar. Burada hatayı yutup geri alma yaptık; hatanın yukarı devam etmesini istiyorsan except bloğunun sonunda raise yaz ya da try/finally kullan.

Sınıf tabanlı bağlam yöneticisi

Durum karmaşıksa ya da nesnenin başka metotları da olacaksa __enter__ ve __exit__ metotlu bir sınıf yazarsın. __exit__ üç argüman alır (hata türü, hata nesnesi, iz); hata yoksa üçü de None'dır. True döndürürse hata bastırılır.

import time

class Kronometre:
    def __init__(self, etiket):
        self.etiket = etiket
        self.sure = None

    def __enter__(self):
        self.t0 = time.perf_counter()
        return self

    def __exit__(self, hata_turu, hata, iz):
        self.sure = time.perf_counter() - self.t0
        durum = "hata ile" if hata_turu else "başarıyla"
        print(f"{self.etiket} {durum} bitti")
        return False   # hatayı bastırma

with Kronometre("rapor") as k:
    toplam = sum(range(300_000))
print("süre ölçüldü mü:", k.sure is not None, "| toplam:", toplam)

try:
    with Kronometre("hatalı iş"):
        raise ValueError("beklenmedik")
except ValueError as e:
    print("dışarıda yakalandı:", e)

RAPORLAR = {}

def rapor(ad):
    def kaydet(fonk):
        print(f"kayıt: {ad} → {fonk.__name__}")
        RAPORLAR[ad] = fonk
        return fonk                 # sarmadan aynen döndür
    return kaydet

@rapor("ciro")
def ciro_raporu(veri):
    return sum(veri)

@rapor("adet")
def adet_raporu(veri):
    return len(veri)

print("Program başladı, kayıtlı raporlar:", list(RAPORLAR))
for ad in ["adet", "ciro"]:
    print(ad, "=", RAPORLAR[ad]([120, 80, 300]))

from functools import wraps

def etiket(ad):
    def dekorator(fonk):
        @wraps(fonk)
        def sarici(*args, **kwargs):
            print(f"{ad} girdi")
            sonuc = fonk(*args, **kwargs)
            print(f"{ad} çıktı")
            return sonuc
        return sarici
    return dekorator

@etiket("DIŞ")
@etiket("İÇ")
def hesapla():
    print("  asıl iş")
    return 42

print(hesapla())

from functools import wraps

def logla(fonk=None, *,):
    def dekorator(f):
        @wraps(f)
        def sarici(*args, **kwargs):
            print(f"[{onek}] {f.__name__}{args}")
            return f(*args, **kwargs)
        return sarici
    if fonk is None:            # @logla(onek=...) biçimi
        return dekorator
    return dekorator(fonk)      # @logla biçimi

@logla
def topla(a, b):
    return a + b

@logla(onek="BANKA")
def faiz(anapara, oran):
    return round(anapara * oran, 2)

print(topla(2, 3), faiz(10000, 0.035))

from functools import lru_cache

@lru_cache(maxsize=2)
def kur(birim):
    print("  servisten çekildi:", birim)
    return {"USD": 34.2, "EUR": 37.1, "GBP": 44.0}[birim]

for b in ["USD", "USD", "EUR", "GBP", "USD"]:
    kur(b)
print(kur.cache_info())
kur.cache_clear()
print("Temizlendi:", kur.cache_info().currsize)

import io
from contextlib import suppress, ExitStack, redirect_stdout

class HataSay:
    def __init__(self, *turler):
        self.turler, self.sayi = turler, 0
    def __enter__(self):
        return self
    def __exit__(self, tur, deger, iz):
        if tur is not None and issubclass(tur, self.turler):
            self.sayi += 1
            return True           # hatayı yut
        return False

sayac = HataSay(ValueError)
for ham in ["10", "x", "7", "?"]:
    with sayac:
        print("geçerli:", int(ham))
print("yutulan hata:", sayac.sayi)

with suppress(KeyError):
    {}["yok"]
print("suppress sonrası devam")

with ExitStack() as yigin:
    dosyalar = [yigin.enter_context(io.StringIO(f"şube {i}")) for i in range(3)]
    print([f.read() for f in dosyalar])

tampon = io.StringIO()
with redirect_stdout(tampon):
    print("bu ekrana değil tampona")
print("yakalandı:", tampon.getvalue().strip())