← Kursa Dön
📄 Text · 15 min

Context Manager ve with Protokolü

Giriş: Otomatik Temizlik

Hayatında hiç bir odaya girip ışığı açtın, işini yaptın ve çıkarken ışığı kapatmayı unuttun mu? Muhtemelen evet. Yazılımda da aynı durum var: dosya açarsın ama kapatmayı unutursun, veritabanı bağlantısı kurarsın ama serbest bırakmayı atlarsın, kilit alırsın ama kilit açmayı ihmal edersin.

Context manager tam olarak bu sorunu çözer. "Odaya gir, işini yap, çıkarken ne olursa olsun ışığı kapat" garantisi verir. Python'daki with statement'ı bunu sağlar.

Bu derste with statement'ın nasıl çalıştığını, kendi context manager'larını nasıl yazacağını ve contextlib kütüphanesinin sunduğu kolaylıkları öğreneceksin.

Context Manager Nedir?

Analoji: Otel Odası

Bir otele gittiğini düşün. Resepsiyonda check-in yaparsın (kaynak edinme), odayı kullanırsın (işini yapma), check-out yaparken oda temizlenir ve anahtar geri alınır (temizlik). Oda temizliğini sen yapmazsın — otel sistemi otomatik halleder.

Context manager de böyle çalışır:

  1. Giriş (__enter__): Kaynağı edin (dosya aç, bağlantı kur)

  2. Kullanım: İşini yap

  3. Çıkış (__exit__): Ne olursa olsun kaynağı temizle (dosyayı kapat, bağlantıyı kes)

with Statement Olmadan

# Tehlikeli yol — dosya kapanmayabilir!
f = open("data.txt", "r")
content = f.read()
# Burada bir hata olursa? f.close() çalışmaz!
f.close()

# Daha güvenli ama çirkin
f = open("data.txt", "r")
try:
    content = f.read()
finally:
    f.close()

with Statement ile

# Temiz ve güvenli yol
with open("data.txt", "r") as f:
    content = f.read()
# Bloktan çıkınca dosya otomatik kapanır
# Hata olsa bile kapanır!

with statement, try/finally bloğunun güzel ve okunaklı hali. Ama arka planda neler oluyor?

with Statement Nasıl Çalışır: __enter__ ve __exit__

Bir nesne with statement ile kullanılabilmek için iki özel metod implemente etmelidir:

__enter__(self)

  • with bloğuna girerken çağrılır

  • Döndürdüğü değer as keyword'ünden sonraki değişkene atanır

  • Genellikle kaynağı hazırlar ve döndürür

__exit__(self, exc_type, exc_val, exc_tb)

  • with bloğundan çıkarken çağrılır (her durumda!)

  • Hata olsa da çağrılır

  • Üç parametre alır: exception tipi, değeri ve traceback

  • True döndürürse exception yutulur, False/None döndürürse exception yeniden fırlatılır

Akış Şeması

# with expression as variable:
#     body

# Arka planda olan:
manager = expression           # Context manager nesnesini al
value = manager.__enter__()    # __enter__ çağır
variable = value               # as değişkenine ata

try:
    body                       # with bloğunun gövdesi
except Exception as e:
    if not manager.__exit__(type(e), e, e.__traceback__):
        raise                  # __exit__ True dönmediyse hatayı fırlat
else:
    manager.__exit__(None, None, None)  # Hata yoksa da çağrılır

Bu mekanizma sayesinde __exit__ her koşulda çağrılır. İşte context manager'ın güvenlik garantisi bu.

Kendi Context Manager Sınıfını Yazma

Örnek 1: Basit Timer

import time

class Timer:
    """Kod bloğunun çalışma süresini ölçer."""
    
    def __enter__(self):
        self.start = time.perf_counter()
        return self  # as değişkenine atanacak
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end = time.perf_counter()
        self.elapsed = self.end - self.start
        print(f"⏱️ Geçen süre: {self.elapsed:.4f} saniye")
        return False  # Exception'ları yutma

# Kullanım
with Timer():
    total = sum(range(1_000_000))
    print(f"Toplam: {total}")
# Çıktı:
# Toplam: 499999500000
# ⏱️ Geçen süre: 0.0312 saniye

# as ile süreye sonra erişme
with Timer() as t:
    data = [x ** 2 for x in range(100_000)]

print(f"Sonra da erişebilirim: {t.elapsed:.4f}s")

Örnek 2: Veritabanı Bağlantısı (Simülasyon)

class DatabaseConnection:
    """Veritabanı bağlantısı context manager'ı."""
    
    def __init__(self, host, database):
        self.host = host
        self.database = database
        self.connection = None
    
    def __enter__(self):
        print(f"🔌 {self.host}/{self.database} bağlantısı kuruluyor...")
        self.connection = f"Connection({self.host}/{self.database})"
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"🔌 Bağlantı kapatılıyor: {self.connection}")
        self.connection = None
        if exc_type is not None:
            print(f"⚠️ Hata oluştu: {exc_val}")
        return False  # Hataları yutma

    def query(self, sql):
        if self.connection is None:
            raise RuntimeError("Bağlantı yok!")
        print(f"📊 SQL çalıştırılıyor: {sql}")
        return [{"id": 1, "name": "Ali"}, {"id": 2, "name": "Veli"}]

# Kullanım
with DatabaseConnection("localhost", "mydb") as db:
    results = db.query("SELECT * FROM users")
    for row in results:
        print(row)
# Çıktı:
# 🔌 localhost/mydb bağlantısı kuruluyor...
# 📊 SQL çalıştırılıyor: SELECT * FROM users
# {'id': 1, 'name': 'Ali'}
# {'id': 2, 'name': 'Veli'}
# 🔌 Bağlantı kapatılıyor: Connection(localhost/mydb)

Örnek 3: Geçici Dizin

import os
import shutil
import tempfile

class TempDirectory:
    """Geçici dizin oluşturur, işlem bitince siler."""
    
    def __init__(self, prefix="tmp_"):
        self.prefix = prefix
        self.path = None
    
    def __enter__(self):
        self.path = tempfile.mkdtemp(prefix=self.prefix)
        print(f"📁 Geçici dizin oluşturuldu: {self.path}")
        return self.path
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.path and os.path.exists(self.path):
            shutil.rmtree(self.path)
            print(f"🗑️ Geçici dizin silindi: {self.path}")
        return False

# Kullanım
with TempDirectory(prefix="myapp_") as tmpdir:
    filepath = os.path.join(tmpdir, "temp_data.txt")
    with open(filepath, "w") as f:
        f.write("Geçici veri")
    print(f"Dosya oluşturuldu: {filepath}")
# Bloktan çıkınca dizin ve içindekiler silinir

Exception Handling in __exit__

__exit__ metodu üç parametre alır:

  • exc_type: Exception'ın tipi (örn: ValueError)

  • exc_val: Exception'ın değeri (hata mesajı)

  • exc_tb: Traceback nesnesi

Eğer with bloğunda hata olmadıysa üçü de None olur.

Hataları Yutmak (Suppress)

class SuppressErrors:
    """Belirli hataları yutar."""
    
    def __init__(self, *exception_types):
        self.exception_types = exception_types
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            if issubclass(exc_type, self.exception_types):
                print(f"🔇 Hata yutuldu: {exc_val}")
                return True  # True = exception'ı yut
        return False  # False = exception'ı fırlat

# ValueError yutulur
with SuppressErrors(ValueError, TypeError):
    int("abc")  # ValueError oluşur ama yutulur
    print("Bu satır çalışmaz")

print("Program devam ediyor!")  # Bu çalışır

# ZeroDivisionError yutulmaz
# with SuppressErrors(ValueError):
#     1 / 0  # ZeroDivisionError — yutulmaz, fırlatılır

⚠️ Dikkat: __exit__'ten True döndürmek exception'ı yok eder. Bunu dikkatli kullan! Hataları yutmak genellikle kötü bir pratik — sadece gerçekten istediğin durumlarda yap.

Temizlik + Hata Raporlama

class ManagedResource:
    """Temizlik yapan ve hataları raporlayan context manager."""
    
    def __init__(self, name):
        self.name = name
    
    def __enter__(self):
        print(f"✅ {self.name} açıldı")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Temizlik HER DURUMDA yapılır
        print(f"🧹 {self.name} temizleniyor...")
        
        if exc_type is not None:
            print(f"❌ Hata tespit edildi: {exc_type.__name__}: {exc_val}")
            # False döndür = hata yeniden fırlatılır
        
        print(f"✅ {self.name} kapatıldı")
        return False

with ManagedResource("Dosya") as r:
    print("İşlem yapılıyor...")
    # raise ValueError("Bir şeyler ters gitti!")

@contextmanager Decorator: Daha Kolay Yol

Sınıf yazmak bazen fazla zahmetli olabiliyor. contextlib modülündeki @contextmanager decorator'ı, generator fonksiyonla context manager yazmana izin verir.

Temel Kullanım

from contextlib import contextmanager

@contextmanager
def timer():
    """Generator tabanlı timer context manager."""
    import time
    start = time.perf_counter()
    yield  # Burası with bloğunun çalıştığı yer
    end = time.perf_counter()
    print(f"⏱️ Geçen süre: {end - start:.4f} saniye")

with timer():
    total = sum(range(1_000_000))

Nasıl Çalışır?

@contextmanager ile dekore edilen generator fonksiyon:

  1. yield'den önce: __enter__ gibi çalışır

  2. yield edilen değer: as değişkenine atanır

  3. yield'den sonra: __exit__ gibi çalışır

@contextmanager
def managed_file(filename, mode="r"):
    """Dosya context manager'ı — generator ile."""
    print(f"📂 Dosya açılıyor: {filename}")
    f = open(filename, mode)
    try:
        yield f  # f, as değişkenine atanır
    finally:
        f.close()
        print(f"📂 Dosya kapatıldı: {filename}")

with managed_file("data.txt", "w") as f:
    f.write("Merhaba Dünya!")

Hata Yönetimi

try/finally kullanmak önemli — yoksa hata durumunda yield'den sonraki kod çalışmaz:

@contextmanager
def safe_resource(name):
    """Hata durumunda da temizlik yapan context manager."""
    print(f"✅ {name} açıldı")
    try:
        yield name
    except Exception as e:
        print(f"❌ Hata yakalandı: {e}")
        raise  # Hatayı yeniden fırlat (yutmak istemiyorsak)
    finally:
        print(f"🧹 {name} temizlendi")

with safe_resource("Veritabanı") as db:
    print(f"Kullanıyorum: {db}")
    # raise RuntimeError("Bağlantı koptu!")

Timer — Gelişmiş Versiyon

@contextmanager
def timer(label="İşlem"):
    """Etiketli timer."""
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"⏱️ {label}: {elapsed:.4f}s")

with timer("Liste oluşturma"):
    data = [x ** 2 for x in range(1_000_000)]

with timer("Generator toplama"):
    total = sum(x ** 2 for x in range(1_000_000))

contextlib Utilities

contextlib modülü çok faydalı araçlar sunar. Hepsini bilmen gerekmez ama en önemlilerini görelim.

suppress: Belirli Hataları Yut

from contextlib import suppress

# Bu:
with suppress(FileNotFoundError):
    os.remove("olmayan_dosya.txt")

# Şuna eşdeğer:
try:
    os.remove("olmayan_dosya.txt")
except FileNotFoundError:
    pass

suppress çok temiz ve okunaklı. "Bu hata olabilir, önemli değil" demenin şık yolu.

redirect_stdout: Çıktıyı Yönlendir

from contextlib import redirect_stdout
import io

# Çıktıyı string'e yönlendir
f = io.StringIO()
with redirect_stdout(f):
    print("Bu ekrana değil, string'e yazılır")
    print("Bu da")

output = f.getvalue()
print(f"Yakalanan çıktı: {output!r}")

# Çıktıyı dosyaya yönlendir
with open("output.txt", "w") as f:
    with redirect_stdout(f):
        print("Bu dosyaya yazılır")
        help(list)  # help() çıktısı da dosyaya gider

redirect_stderr: Hata Çıktısını Yönlendir

from contextlib import redirect_stderr
import io

f = io.StringIO()
with redirect_stderr(f):
    import warnings
    warnings.warn("Test uyarısı")

print(f"Yakalanan: {f.getvalue()}")

closing: close() Metodu Olan Nesneler İçin

from contextlib import closing
from urllib.request import urlopen

# urlopen context manager değil (eski Python'larda)
# closing() ile sarmalayabiliriz
with closing(urlopen("https://example.com")) as page:
    content = page.read()

ExitStack: Dinamik Context Manager Yığını

from contextlib import ExitStack

# Dinamik sayıda dosya açma
filenames = ["file1.txt", "file2.txt", "file3.txt"]

with ExitStack() as stack:
    files = [
        stack.enter_context(open(fname, "w"))
        for fname in filenames
    ]
    for f in files:
        f.write("Merhaba!\n")
# Tüm dosyalar otomatik kapanır

ExitStack özellikle dosya sayısının runtime'da belirlenmesi gereken durumlarda kullanışlı.

Nested Context Managers

Tek Satırda Birden Fazla Context Manager

Python 3.1'den itibaren birden fazla context manager'ı tek with statement'ında kullanabilirsin:

# Dosya kopyalama — iki dosya aynı anda açık
with open("kaynak.txt", "r") as src, open("hedef.txt", "w") as dst:
    for line in src:
        dst.write(line.upper())

Python 3.10'dan itibaren parantezli syntax de var:

# Python 3.10+ parantezli syntax
with (
    open("kaynak.txt", "r") as src,
    open("hedef.txt", "w") as dst,
    Timer() as t,
):
    for line in src:
        dst.write(line.upper())

İç İçe with Blokları

# Bazen iç içe yazmak daha okunaklı
with DatabaseConnection("localhost", "users_db") as db:
    users = db.query("SELECT * FROM users")
    
    with open("users_export.csv", "w") as f:
        f.write("id,name\n")
        for user in users:
            f.write(f"{user['id']},{user['name']}\n")

💡 İpucu: Eğer context manager'lar birbirine bağımlı değilse tek satırda yaz. Bağımlıysa (biri diğerinin sonucunu kullanıyorsa) iç içe yaz.

Pratik: İleri Örnekler

Çalışma Dizini Değiştirici

import os
from contextlib import contextmanager

@contextmanager
def change_directory(path):
    """Geçici olarak çalışma dizinini değiştirir."""
    original = os.getcwd()
    try:
        os.chdir(path)
        print(f"📁 Dizin değiştirildi: {path}")
        yield path
    finally:
        os.chdir(original)
        print(f"📁 Dizin geri alındı: {original}")

with change_directory("/tmp"):
    print(f"Şu an: {os.getcwd()}")
    # /tmp içinde işlemler yap

print(f"Geri döndük: {os.getcwd()}")

Ortam Değişkeni Geçici Değiştirici

@contextmanager
def temp_env(**kwargs):
    """Geçici olarak ortam değişkenlerini değiştirir."""
    old_values = {}
    for key, value in kwargs.items():
        old_values[key] = os.environ.get(key)
        os.environ[key] = value
    
    try:
        yield
    finally:
        for key, old_value in old_values.items():
            if old_value is None:
                del os.environ[key]
            else:
                os.environ[key] = old_value

with temp_env(DEBUG="1", DATABASE_URL="sqlite:///test.db"):
    print(os.environ["DEBUG"])        # 1
    print(os.environ["DATABASE_URL"]) # sqlite:///test.db

# Blok dışında eski değerler geri gelir

Transaction Manager

@contextmanager
def transaction(connection):
    """Veritabanı transaction yönetimi."""
    cursor = connection.cursor()
    try:
        yield cursor
        connection.commit()
        print("✅ Transaction commit edildi")
    except Exception as e:
        connection.rollback()
        print(f"❌ Transaction rollback yapıldı: {e}")
        raise

# Kullanım (pseudo-code)
# with transaction(db_connection) as cursor:
#     cursor.execute("INSERT INTO users VALUES (1, 'Ali')")
#     cursor.execute("INSERT INTO orders VALUES (1, 1, 100)")
# Her iki INSERT ya birlikte başarılı olur, ya birlikte geri alınır

İzinleri Geçici Değiştirici

@contextmanager
def temp_permissions(filepath, new_mode):
    """Dosya izinlerini geçici olarak değiştirir."""
    import stat
    original_mode = os.stat(filepath).st_mode
    try:
        os.chmod(filepath, new_mode)
        yield
    finally:
        os.chmod(filepath, original_mode)

# Dosyayı geçici olarak yazılabilir yap
# with temp_permissions("config.ini", 0o644):
#     with open("config.ini", "w") as f:
#         f.write("new_config=value")

Context Manager Tasarım Desenleri

1. Setup/Teardown Pattern

En yaygın kullanım: bir kaynağı aç, kullan, kapat.

@contextmanager
def smtp_connection(host, port):
    """Email bağlantısı aç/kapat."""
    import smtplib
    server = smtplib.SMTP(host, port)
    try:
        server.starttls()
        yield server
    finally:
        server.quit()

2. State Değiştirme Pattern

Bir durumu geçici olarak değiştir, sonra geri al.

@contextmanager  
def precision(decimal_places):
    """Geçici olarak ondalık hassasiyeti değiştir."""
    import decimal
    old_context = decimal.getcontext().copy()
    decimal.getcontext().prec = decimal_places
    try:
        yield
    finally:
        decimal.setcontext(old_context)

3. Loglama/Ölçüm Pattern

Bir işlemin başlangıç ve bitişini logla.

@contextmanager
def log_operation(operation_name):
    """İşlemi logla."""
    import logging
    logger = logging.getLogger(__name__)
    logger.info(f"🚀 {operation_name} başladı")
    try:
        yield
        logger.info(f"✅ {operation_name} tamamlandı")
    except Exception as e:
        logger.error(f"❌ {operation_name} başarısız: {e}")
        raise

4. Lock Pattern

Thread güvenliği için kilit al/bırak.

import threading

@contextmanager
def timed_lock(lock, timeout=5.0):
    """Zaman aşımlı kilit."""
    acquired = lock.acquire(timeout=timeout)
    if not acquired:
        raise TimeoutError(f"Kilit {timeout}s içinde alınamadı")
    try:
        yield
    finally:
        lock.release()

lock = threading.Lock()
with timed_lock(lock, timeout=3.0):
    # Kritik bölge
    pass

Async Context Manager

Python 3.5+ ile async context manager'lar da var. __aenter__ ve __aexit__ kullanır:

class AsyncTimer:
    async def __aenter__(self):
        import time
        self.start = time.perf_counter()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        import time
        self.elapsed = time.perf_counter() - self.start
        print(f"⏱️ Async süre: {self.elapsed:.4f}s")
        return False

# Kullanım:
# async with AsyncTimer():
#     await some_async_operation()

contextlib ile de yazılabilir:

from contextlib import asynccontextmanager

@asynccontextmanager
async def async_db_connection(url):
    """Async veritabanı bağlantısı."""
    print(f"🔌 Bağlanıyor: {url}")
    connection = await create_connection(url)  # async bağlantı
    try:
        yield connection
    finally:
        await connection.close()
        print("🔌 Bağlantı kapatıldı")

Yaygın Hatalar

1. __exit__'te Temizliği Unutmak

# YANLIŞ: Hata olursa temizlik yapılmaz
class BadManager:
    def __exit__(self, *args):
        if args[0] is None:  # Sadece hata yoksa temizle
            self.cleanup()
        # Hata varsa temizlik YOK!

# DOĞRU: Her durumda temizle
class GoodManager:
    def __exit__(self, *args):
        self.cleanup()  # Her durumda temizle
        return False

2. yield Sonrasında finally Kullanmamak

# YANLIŞ: Hata olursa temizlik yapılmaz
@contextmanager
def bad_manager():
    resource = acquire()
    yield resource
    release(resource)  # Hata olursa buraya ulaşılmaz!

# DOĞRU
@contextmanager
def good_manager():
    resource = acquire()
    try:
        yield resource
    finally:
        release(resource)  # Her durumda çalışır

3. as Değişkenini Bloğun Dışında Kullanmak

with open("data.txt") as f:
    content = f.read()

# f hala erişilebilir ama KAPALI
# f.read()  # ValueError: I/O operation on closed file
# content kullanılabilir — o bir string, dosya değil
print(content)  # Bu çalışır

Özet

  • Context manager, with statement ile kaynak yönetimini otomatikleştiren bir protokoldür. __enter__ giriş, __exit__ çıkış işlemlerini tanımlar.

  • `with` statement, try/finally bloğunun okunaklı alternatifidir. Hata olsa bile __exit__ her zaman çağrılır.

  • Sınıf tabanlı context manager __enter__ ve __exit__ metodları ile, generator tabanlı context manager @contextmanager decorator'ı ile yazılır.

  • `__exit__` metodu True döndürürse exception yutulur. Dikkatli kullan — genellikle False döndürmek (hatayı fırlatmak) daha güvenlidir.

  • `contextlib` modülü suppress, redirect_stdout, closing, ExitStack gibi kullanışlı araçlar sunar.

  • Context manager'lar dosya, veritabanı, kilit, geçici durum değişikliği ve ölçüm gibi kaynak yönetimi senaryolarında vazgeçilmezdir.