H.Ozgan
a.k.a netologist

Python'da Veritabanı ve Transaction Yönetimi — Derinlemesine Rehber

13 Mart 2026 · netologist · 22 dakika, 4552 kelime · #connection-pooling #database #note #python #sql #sqlalchemy #transactions

İçindekiler

  1. Temeller: DB-API 2.0 (PEP 249)
  2. Bağlantılar, Cursor’lar ve Round-Trip Maliyeti
  3. ACID ve Byte Seviyesinde Gerçek Anlamı
  4. Transaction İzolasyon Seviyeleri — Derinlemesine
  5. Context Manager Olarak Transaction’lar
  6. Savepoint’ler ve İç İçe Transaction’lar
  7. Connection Pooling (Bağlantı Havuzu)
  8. SQLAlchemy Core: Açık (Explicit) Transaction Yönetimi
  9. SQLAlchemy ORM: Unit of Work ve Session Semantiği
  10. Django ORM Transaction Yönetimi
  11. Asenkron Veritabanı Erişimi
  12. Kilitleme Stratejileri: Optimistic vs Pessimistic
  13. Deadlock (Kilitlenme): Tespit, Teşhis, Önleme
  14. Dağıtık Transaction’lar: 2PC ve Saga Pattern
  15. Retry, Idempotency ve “Tam Olarak Bir Kez” Yanılsaması
  16. Transactional Kodun Test Edilmesi
  17. Migration ve Şema Evrimi (Alembic)
  18. Performans Değerlendirmeleri
  19. Sık Yapılan Hatalar Kontrol Listesi

1. Temeller: DB-API 2.0 (PEP 249)

Python’daki her ilişkisel veritabanı sürücüsü — sqlite3, psycopg2, psycopg3, mysqlclient, pyodbc, cx_Oracle — PEP 249 tarafından tanımlanan aynı düşük seviyeli sözleşmeyi (contract) uygular. Bu sözleşmeyi anlamak zorunludur, çünkü her ORM (SQLAlchemy, Django, Peewee) sonuçta bunun üzerine inşa edilmiş bir kod üretecidir.

Temel nesneler

import sqlite3

conn = sqlite3.connect("app.db")   # Connection: transaction durumunu tutar
cur = conn.cursor()                # Cursor: neredeyse durumsuz, SQL çalıştırır

cur.execute("SELECT id, name FROM users WHERE active = ?", (True,))
rows = cur.fetchall()

cur.close()
conn.close()

Paramstyle önemlidir

Farklı sürücüler farklı yer tutucu (placeholder) stilleri kullanır:

SürücüparamstyleÖrnek
sqlite3qmarkWHERE id = ?
psycopg2pyformatWHERE id = %(id)s veya %s
pyodbcqmarkWHERE id = ?

Değerleri SQL’e eklemek için asla f-string veya %-formatlama kullanmayın — bu, SQL injection zafiyetlerinin #1 sebebidir. Değerleri her zaman sürücünün yer tutucu mekanizması üzerinden geçirin; böylece kaçış (escaping) doğru yapılır ve sorgu planları (query plan) önbelleğe alınabilir.

# YANLIŞ — SQL injection riski
cur.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# DOĞRU — parametreli
cur.execute("SELECT * FROM users WHERE name = ?", (user_input,))

Örtülü (implicit) transaction başlangıcı

Kritik ve sıkça yanlış anlaşılan bir detay: DB-API 2.0 bağlantıları, veriyi değiştiren ilk ifadede transaction’ı örtülü olarak başlatır (spesifikasyona göre, sürücüye bağlı olarak bazen SELECT için bile). API yüzeyinde açık bir “BEGIN” gerekmez — ama kesinlikle açık bir transaction vardır. commit() veya rollback() çağrısını mutlaka siz yapmalısınız; DB-API, sürücü aksini belirtmedikçe bağlantıların varsayılan olarak autocommit = kapalı olmasını zorunlu kılar (not: modern Python’da sqlite3, isolation_level ayarına göre farklı davranabilir; psycopg2 varsayılan olarak autocommit’i kapalı tutar).

2. Bağlantılar, Cursor’lar ve Round-Trip Maliyeti

Bağlantılar pahalı, cursor’lar ucuzdur

TCP bağlantısı kurmak, TLS anlaşması yapmak ve veritabanı sunucusuna karşı kimlik doğrulamak onlarca milisaniyeye mal olabilir. Bu yüzden:

Round-trip gecikmesi baskındır

Python veritabanı kodundaki en yaygın performans hatası N+1 sorgu problemidir: satırlar üzerinde döngü kurup her satır için yeni bir sorgu göndermek.

# KÖTÜ: N+1 sorgu — kullanıcı başına 1 round trip
for user_id in user_ids:
    cur.execute("SELECT * FROM orders WHERE user_id = ?", (user_id,))
    orders = cur.fetchall()

# İYİ: 1 round trip
cur.execute(
    "SELECT * FROM orders WHERE user_id IN ({})".format(
        ",".join("?" * len(user_ids))
    ),
    user_ids,
)
orders = cur.fetchall()

ORM bağlamında bu, select_related / joinedload kullanmayı unutup her ilişkili nesne için tembel (lazy) bir sorgu tetiklemek şeklinde ortaya çıkar.

Sunucu tarafı (server-side) vs istemci tarafı (client-side) cursor

Bazı sürücüler (özellikle psycopg2) sonuç kümesinin tamamını istemci tarafında belleğe almak yerine sunucudan akış (stream) halinde getiren isimli (server-side) cursor’ları destekler. Milyonlarca satır üzerinde iterasyon yaparken istemci bellek patlamasını önlemek için bunları kullanın:

with conn.cursor(name="server_side_cursor") as cur:
    cur.itersize = 2000  # toplu getirme boyutu
    cur.execute("SELECT * FROM huge_table")
    for row in cur:  # akış halinde, hepsi bir kerede yüklenmez
        process(row)

3. ACID ve Byte Seviyesinde Gerçek Anlamı

Bunun Python kodunda önemi

Python kodunuz ACID’i uygulamaz — bunu veritabanı motoru yapar. Uygulama geliştiricisi olarak sizin işiniz transaction sınırlarını (boundary) doğru çizmektir: çok dar çizerseniz kısmi/tutarsız yazmalar oluşur; çok geniş çizerseniz kilitleri gereğinden uzun tutar, çekişme (contention) yaratır ve eşzamanlılığı düşürürsünüz.

# Transaction sınırı çok geniş: yavaş I/O sırasında kilitler tutuluyor
with conn:
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = ?", (from_id,))
    send_email_confirmation(from_id)  # YAVAŞ, alakasız I/O — kilitler tüm süre boyunca tutulur!
    cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = ?", (to_id,))

Çözüm: transaction’ları kısa tutun, transaction dışı yan etkileri (e-posta, HTTP çağrıları, loglama) transaction’ın dışına taşıyın; güvenilirlik için ideal olarak bir outbox pattern kullanın.

4. Transaction İzolasyon Seviyeleri — Derinlemesine

SQL, dört standart izolasyon seviyesi tanımlar. Python sürücüleri bunları bağlantı öznitelikleri veya SET TRANSACTION ISOLATION LEVEL ifadeleri üzerinden sunar.

SeviyeDirty ReadNon-repeatable ReadPhantom ReadNotlar
READ UNCOMMITTEDMümkünMümkünMümkünNadiren kullanılır; PostgreSQL bunu READ COMMITTED gibi ele alır
READ COMMITTEDHayırMümkünMümkünPostgreSQL, Oracle, SQL Server’da varsayılan
REPEATABLE READHayırHayırMümkün*MySQL/InnoDB’de varsayılan; PostgreSQL’in uygulaması aslında snapshot isolation’dır ve pratikte phantom’ları engeller
SERIALIZABLEHayırHayırHayırPostgreSQL’de Serializable Snapshot Isolation (SSI) ile uygulanır — uygulamanın yeniden denemesi (retry) gereken serialization hatalarıyla transaction’ları iptal edebilir

Python’da izolasyon seviyesi ayarlama

psycopg2:

import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_SERIALIZABLE

conn = psycopg2.connect(dsn)
conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)

psycopg3:

import psycopg

conn = psycopg.connect(dsn)
conn.isolation_level = psycopg.IsolationLevel.SERIALIZABLE

sqlite3 (SQLite, motor seviyesinde gerçekten sadece SERIALIZABLE’ı destekler, ama kendi kilitleme modlarına sahiptir: DEFERRED, IMMEDIATE, EXCLUSIVE):

conn = sqlite3.connect("app.db", isolation_level=None)  # autocommit modu
conn.execute("BEGIN IMMEDIATE")   # yazma kilidini hemen al, deadlock'lardan kaçın
...
conn.commit()

Serialization hataları yeniden denenmelidir

SERIALIZABLE altında veritabanı, transaction’ınızı could not serialize access due to concurrent update gibi bir hatayla iptal edebilir. Bu bir hata (bug) değildir — izolasyon mekanizmasının doğru çalıştığının göstergesidir. Kodunuz bunu yakalayıp yeniden denemelidir:

import time
import psycopg2
from psycopg2 import errors

def run_with_serializable_retry(fn, conn, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            with conn:
                return fn(conn)
        except errors.SerializationFailure:
            conn.rollback()
            if attempt == max_attempts - 1:
                raise
            time.sleep(0.05 * (2 ** attempt))  # üstel geri çekilme (exponential backoff)

5. Context Manager Olarak Transaction’lar

sqlite3 ve psycopg2: connection’ı context manager olarak kullanmak ≠ bağlantıyı kapatmak

Birçok geliştiriciyi yanıltan bir incelik: hem sqlite3 hem psycopg2‘de connection nesnesini context manager olarak kullanmak transaction’ı commit veya rollback eder — bağlantıyı kapatmaz.

conn = sqlite3.connect("app.db")

with conn:  # başarıda commit, hata durumunda rollback — bağlantı AÇIK kalır
    conn.execute("INSERT INTO logs(msg) VALUES (?)", ("started",))

# conn burada hâlâ açık!
conn.close()  # kendiniz kapatmalısınız

Hem transaction’ı yönetmek hem bağlantıyı kapatmak için context manager’ları iç içe kullanın veya contextlib.closing kullanın:

from contextlib import closing

with closing(sqlite3.connect("app.db")) as conn:
    with conn:
        conn.execute("INSERT INTO logs(msg) VALUES (?)", ("done",))

Kendi transaction context manager’ınızı yazmak

Yerleşik transaction context manager’ı olmayan sürücüler/frameworkler için küçük, yeniden kullanılabilir bir tane yazın:

from contextlib import contextmanager

@contextmanager
def transaction(conn):
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise

with transaction(conn) as tx:
    cur = tx.cursor()
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = %s", (1,))
    cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = %s", (2,))

6. Savepoint’ler ve İç İçe Transaction’lar

Çoğu SQL motorunda gerçek anlamda iç içe (nested) transaction yoktur — ama savepoint’ler, tek bir üst düzey transaction içinde kısmi rollback imkânı sağlar. Bu, uygulama kodunda “iç içe” transaction semantiğini uygulamak için (örn. SQLAlchemy’nin begin_nested()‘ı) gereklidir.

conn = psycopg2.connect(dsn)
cur = conn.cursor()

cur.execute("BEGIN")
cur.execute("INSERT INTO orders (id, status) VALUES (1, 'pending')")

cur.execute("SAVEPOINT sp1")
try:
    cur.execute("INSERT INTO order_items (order_id, sku) VALUES (1, 'BAD-SKU')")
except psycopg2.IntegrityError:
    cur.execute("ROLLBACK TO SAVEPOINT sp1")  # sadece bu kısmı geri al
    cur.execute("RELEASE SAVEPOINT sp1")

conn.commit()  # order satırı hayatta kalır; hatalı item ekleme geri alınmıştır

SQLAlchemy iç içe transaction’ları (savepoint’ler)

from sqlalchemy.orm import Session

with Session(engine) as session:
    with session.begin():
        session.add(Order(id=1, status="pending"))

        try:
            with session.begin_nested():  # SAVEPOINT
                session.add(OrderItem(order_id=1, sku="BAD-SKU"))
                raise ValueError("doğrulama başarısız")
        except ValueError:
            pass  # iç içe rollback sadece OrderItem eklemesini geri alır

        # Order(id=1) burada hâlâ commit için sıraya alınmış durumdadır

Önemli: her SAVEPOINT‘in gerçek bir maliyeti vardır — ekstra round trip ve log kaydı. Bunları düzgün doğrulamanın yerine kullanmayın; gerçekten opsiyonel/kurtarılabilir alt işlemler için kullanın.

7. Connection Pooling (Bağlantı Havuzu)

Neden havuzlama?

Her istek için TCP el sıkışması + kimlik doğrulama + (varsa) TLS anlaşması, saniyede yüzlerce istek sunan bir web uygulamasında kabul edilemez. Bir havuz, alınmaya ve geri verilmeye hazır bir grup canlı bağlantıyı tutar.

psycopg2’nin yerleşik havuzları

from psycopg2 import pool

connection_pool = pool.ThreadedConnectionPool(
    minconn=2, maxconn=20, dsn="postgresql://user:pass@host/db"
)

conn = connection_pool.getconn()
try:
    with conn:
        conn.cursor().execute("SELECT 1")
finally:
    connection_pool.putconn(conn)  # HER ZAMAN geri verin, hata olsa bile

SQLAlchemy’nin havuzlaması (ORM dışında, Core üzerinden bile kullanılır)

SQLAlchemy’nin Engine‘i varsayılan olarak bir connection pool’u (QueuePool) sarmalar. Önemli ayarlar:

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg2://user:pass@host/db",
    pool_size=10,          # kararlı durum havuz boyutu
    max_overflow=5,        # yoğun yük altında izin verilen ekstra bağlantılar
    pool_timeout=30,       # hata vermeden önce bağlantı için bekleme süresi (saniye)
    pool_recycle=1800,     # bundan eski bağlantıları geri dönüştür (bayat TCP'yi önler)
    pool_pre_ping=True,    # bir bağlantıyı vermeden önce hafif bir SELECT 1 çalıştırır
)

pool_pre_ping=True, yük dengeleyicilerin veya güvenlik duvarlarının boşta duran TCP bağlantılarını sessizce kapattığı bulut ortamlarında kritiktir — bu olmadan gizemli OperationalError: server closed the connection unexpectedly hataları alırsınız.

Harici havuzlayıcılar: PgBouncer

Özellikle PostgreSQL için, uygulama düzeyi havuzlama (SQLAlchemy’nin havuzu) ile PgBouncer gibi ağ düzeyi bir havuzlayıcı farklı problemleri çözer. PgBouncer, uygulamanız (potansiyel olarak birçok süreç, örn. Gunicorn worker’ları) ile Postgres arasında durur ve binlerce istemci bağlantısını az sayıda gerçek sunucu bağlantısına çoğullar (multiplex). Dikkat: PgBouncer’ın transaction pooling modu, dikkatlice yapılandırılmadıkça oturum durumuna (session state) dayanan özellikleri (SET, prepared statement’lar, LISTEN/NOTIFY, advisory lock’lar) bozar.

Asenkron havuzlar

import asyncpg

pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)

async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = $1", 1)
        await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = $1", 2)

8. SQLAlchemy Core: Açık (Explicit) Transaction Yönetimi

SQLAlchemy Core, tam ORM nesne eşlemesi olmadan SQL seviyesinde kontrol sağlar — toplu işlemler, raporlama, veya ORM’in identity map ek yükünün değmediği durumlar için iyidir.

from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg2://user:pass@host/db")

# engine.connect() otomatik commit YAPMAZ; transaction'ı siz kontrol edersiniz
with engine.connect() as conn:
    with conn.begin():  # açık (explicit) transaction
        conn.execute(
            text("UPDATE accounts SET balance = balance - :amt WHERE id = :id"),
            {"amt": 100, "id": 1},
        )
        conn.execute(
            text("UPDATE accounts SET balance = balance + :amt WHERE id = :id"),
            {"amt": 100, "id": 2},
        )
    # hata yoksa otomatik commit, varsa rollback edilir

engine.begin() kısayolu

with engine.begin() as conn:  # bağlan + transaction başlat, tek çağrıda
    conn.execute(text("INSERT INTO audit_log(msg) VALUES (:m)"), {"m": "transfer executed"})

Autocommit benzeri davranış (2.0 stili) vs eski (legacy) autocommit

SQLAlchemy 1.x, açık transaction dışında DML için örtülü bir “autocommit” moduna sahipti — bu, sık sık kafa karışıklığına yol açtı ve SQLAlchemy 2.0’da kaldırıldı. 2.0’da, her unit of work açık bir begin()/commit veya Session kullanımı gerektirir. Bu, geliştiricileri transaction sınırları konusunda açık olmaya zorlamak için bilinçli bir tasarım kararıdır.

9. SQLAlchemy ORM: Unit of Work ve Session Semantiği

Session bir bağlantı değildir — bir unit of work’tür

Session, nesne durumunu (new, dirty, deleted) bellekte izler ve değişiklikleri yalnızca gerektiğinde (etkilenecek bir sorgudan önce veya commit() çağrısında) SQL olarak veritabanına iletir (flush). Bu, Unit of Work desenidir.

from sqlalchemy.orm import Session

with Session(engine) as session:
    user = session.get(User, 1)
    user.balance -= 100          # "dirty" olarak izlenir, henüz SQL yok

    other = session.get(User, 2) # session.get() gerekirse önce flush tetikleyebilir
    other.balance += 100

    session.commit()             # flush + COMMIT burada gerçekleşir

Dış transaction sınırı olarak session.begin()

with Session(engine) as session:
    with session.begin():
        session.add(Order(...))
        session.add(OrderItem(...))
    # başarıda otomatik commit, hatada otomatik rollback

Autoflush ve autocommit tuzakları

Commit sonrası expiration (geçersiz kılma)

Varsayılan olarak expire_on_commit=True — commit() sonrası tüm ORM nesneleri “expired” (süresi dolmuş) olarak işaretlenir ve herhangi bir özniteliğe erişmek yeni bir SELECT tetikler. Bu güvenlidir ama bir döngü içinde commit() yapıp sonra öznitelik okuyorsanız incelikli N+1 sorunlarına yol açabilir.

with Session(engine) as session:
    for order in session.query(Order).all():
        order.status = "shipped"
        session.commit()  # YÜKLENMİŞ TÜM nesneleri expire eder
        print(order.id)   # sadece `id`'yi yeniden almak için yeni bir SELECT tetikler!

Çözüm: commit’leri döngü dışında toplu yapın, veya bayatlık (staleness) ödünleşimini anladığınızda expire_on_commit=False kullanın.

İlişki yükleme stratejileri transaction şeklini etkiler

from sqlalchemy.orm import joinedload, selectinload

# N+1 riski: erişim anında tembel çalıştırılan, kullanıcı başına bir sorgu
users = session.query(User).all()
for u in users:
    print(u.orders)  # kullanıcı başına ayrı bir SELECT!

# Düzeltilmiş: tek bir JOIN sorgusu
users = session.query(User).options(joinedload(User.orders)).all()

# Farklı şekilde düzeltilmiş: toplam 2 sorgu (1 kullanıcılar için, 1 tüm siparişler için IN sorgusu)
users = session.query(User).options(selectinload(User.orders)).all()

10. Django ORM Transaction Yönetimi

Django, DB-API bağlantılarını kendi transaction yönetim katmanıyla sarmalar.

ATOMIC_REQUESTS ayarı

DATABASES = {
    "default": {
        ...,
        "ATOMIC_REQUESTS": True,  # her view'i bir transaction'a sarar — ölçekte dikkatli kullanın
    }
}

Bu kullanışlıdır ama her HTTP isteğini bir transaction’a sarar; istek boyunca (yavaş şablon render’ı veya harici API çağrıları dâhil) bir bağlantıyı (ve muhtemelen kilitleri) tutar. Çoğu büyük Django kod tabanı bunu global olarak kapatır ve bunun yerine açık atomic() bloklarını kullanır.

Açık atomic() blokları

from django.db import transaction

@transaction.atomic
def transfer_funds(from_id, to_id, amount):
    from_account = Account.objects.select_for_update().get(id=from_id)
    to_account = Account.objects.select_for_update().get(id=to_id)

    if from_account.balance < amount:
        raise InsufficientFunds()

    from_account.balance -= amount
    to_account.balance += amount
    from_account.save()
    to_account.save()

select_for_update(), SELECT ... FOR UPDATE çalıştırarak satır seviyesinde bir kilit alır — eşzamanlı erişim altında oku-değiştir-yaz (read-modify-write) desenlerinde kayıp güncellemeleri (lost update) önlemek için gereklidir (bkz. Bölüm 12).

İç içe atomic() üzerinden savepoint’ler

with transaction.atomic():
    order = Order.objects.create(status="pending")
    try:
        with transaction.atomic():  # SAVEPOINT
            OrderItem.objects.create(order=order, sku="BAD-SKU")
            validate_sku_or_raise("BAD-SKU")
    except ValidationError:
        pass  # sadece iç savepoint geri alınır

on_commit hook’ları — hafif outbox deseni

Çok yaygın bir Python/Django hatası: bir e-posta göndermek veya bir Celery görevi kuyruğa almak, sonradan rollback olan bir transaction’ın içinde yapılırsa, aslında hiç kalıcı hale gelmemiş veri için yan etki tetiklenir (ya da daha kötüsü: görev, transaction commit olmadan önce çalışır ve görünürlük kurallarından ötürü bayat/eksik veri okur).

from django.db import transaction

def create_order(data):
    with transaction.atomic():
        order = Order.objects.create(**data)
        transaction.on_commit(lambda: send_confirmation_email.delay(order.id))
        # e-posta görevi sadece transaction gerçekten commit olursa/olduğunda kuyruğa alınır

Transaction izolasyon seviyesi yapılandırması

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "OPTIONS": {
            "isolation_level": "serializable",  # arka planda psycopg2 sabitleri üzerinden
        },
    }
}

11. Asenkron Veritabanı Erişimi

DB I/O için async neden önemlidir

Asenkron veritabanı sürücüleri, ağ I/O sırasında event loop’u bloklamaktan kaçınır ve tek bir sürecin binlerce eşzamanlı bağlantıyı yönetmesine izin verir — yüksek verimli asenkron web framework’leri (FastAPI, Starlette, aiohttp) için kritiktir.

asyncpg (PostgreSQL, çok hızlı, binary protokol)

import asyncpg

async def transfer(pool, from_id, to_id, amount):
    async with pool.acquire() as conn:
        async with conn.transaction():
            await conn.execute(
                "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
                amount, from_id,
            )
            await conn.execute(
                "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
                amount, to_id,
            )

asyncpg‘nin transaction() context manager’ı izolasyon seviyesini ve salt-okunur bayraklarını doğrudan destekler:

async with conn.transaction(isolation="serializable", readonly=False):
    ...

aiosqlite (SQLite, asenkron sarmalayıcı — altyapıda hâlâ tek yazarlı)

import aiosqlite

async def log_event(db_path, msg):
    async with aiosqlite.connect(db_path) as conn:
        async with conn.execute("INSERT INTO logs(msg) VALUES (?)", (msg,)):
            pass
        await conn.commit()

Önemli bir incelik: SQLite’ın kendisi, asenkron sarmalamadan bağımsız olarak aynı anda sadece bir yazara izin verir — aiosqlite, SQLite’ın temel tek-yazarlı eşzamanlılık modelini değiştirmez, sadece (senkron, thread’li) SQLite sürücüsünü beklerken Python event loop’unu bloklamaktan kaçınır.

SQLAlchemy 2.0 asenkron ORM

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")

async def transfer(from_id, to_id, amount):
    async with AsyncSession(engine) as session:
        async with session.begin():
            from_acc = await session.get(Account, from_id)
            to_acc = await session.get(Account, to_id)
            from_acc.balance -= amount
            to_acc.balance += amount
        # başarılı çıkışta otomatik commit

Senkron ve asenkronu karıştırmak: #1 async DB hatası

Senkron bir sürücüyü (örn. psycopg2, düz sqlite3) doğrudan bir async def içinde çağırmak tüm event loop’u bloklar, diğer tüm coroutine’leri durdurur. Asenkron bir uygulamada senkron bir sürücü kullanmak zorundaysanız, bunu başka bir thread’e taşıyın:

import asyncio

async def run_sync_query(conn, query, params):
    return await asyncio.to_thread(conn.execute, query, params)

Daha iyisi: senkron bir sürücüyü sarmalamak yerine native asenkron sürücü (asyncpg, MySQL için asyncmy) kullanın.

12. Kilitleme Stratejileri: Optimistic vs Pessimistic

Pessimistic (kötümser) kilitleme: SELECT ... FOR UPDATE

Çakışmaların olası olduğunu varsayar; baştan bir satır kilidi alır ve commit/rollback’e kadar diğer transaction’ların aynı satırları değiştirmesini (hatta FOR UPDATE ile bazen okumasını bile) engeller.

# psycopg2 / ham SQL
cur.execute("BEGIN")
cur.execute("SELECT balance FROM accounts WHERE id = %s FOR UPDATE", (account_id,))
balance = cur.fetchone()[0]
cur.execute("UPDATE accounts SET balance = %s WHERE id = %s", (balance - 100, account_id))
cur.execute("COMMIT")

# SQLAlchemy ORM
account = session.query(Account).filter_by(id=account_id).with_for_update().one()
account.balance -= 100
session.commit()

FOR UPDATE SKIP LOCKED, doğrudan SQL içinde iş kuyruğu (job queue) oluşturmak için güçlü bir varyanttır — worker’lar, diğer worker’ların zaten tuttuğu satırlarda bloklanmadan bir sonraki uygun satırı alır:

cur.execute("""
    SELECT id FROM job_queue
    WHERE status = 'pending'
    ORDER BY created_at
    FOR UPDATE SKIP LOCKED
    LIMIT 1
""")

Optimistic (iyimser) kilitleme: versiyon sütunları

Çakışmaların nadir olduğunu varsayar; kilitlemek yerine, yazma anında bir versiyon numarasını (veya zaman damgasını) kontrol eder ve okuma sonrası değiştiyse yazmayı başarısız kılar.

# şema: accounts(id, balance, version)

cur.execute("SELECT balance, version FROM accounts WHERE id = %s", (account_id,))
balance, version = cur.fetchone()

new_balance = balance - 100
cur.execute(
    "UPDATE accounts SET balance = %s, version = version + 1 "
    "WHERE id = %s AND version = %s",
    (new_balance, account_id, version),
)

if cur.rowcount == 0:
    raise ConcurrentModificationError("satır başka bir transaction tarafından değiştirildi")

SQLAlchemy, version_id_col üzerinden yerleşik destek sunar:

class Account(Base):
    __tablename__ = "accounts"
    id = Column(Integer, primary_key=True)
    balance = Column(Numeric)
    version_id = Column(Integer, nullable=False)

    __mapper_args__ = {"version_id_col": version_id}

try:
    session.commit()
except StaleDataError:
    session.rollback()
    # yeniden dene: yeniden yükle ve değişikliği tekrar uygula

Hangi durumda hangisi kullanılmalı

SenaryoTercih
Yüksek çekişme (aynı satırlarda birçok yazar)Pessimistic
Düşük çekişme, yüksek okuma/yazma oranıOptimistic
Okuma ile yazma arasında uzun kullanıcı “düşünme süresi” (örn. bir formu düzenlerken)Optimistic (kullanıcı düşünme süresi boyunca asla bir DB kilidi tutmayın!)
İş kuyrukları / görev dağıtımıSKIP LOCKED ile Pessimistic

13. Deadlock (Kilitlenme): Tespit, Teşhis, Önleme

Bir deadlock, Transaction A’nın Transaction B’nin ihtiyaç duyduğu bir kilidi tutması ve tam tersinin de doğru olması durumunda oluşur. Veritabanları bu döngüyü tespit eder ve bir hata ile bir transaction’ı iptal eder (kurban - “victim”) — bu bir donma değildir, ele almanız gereken bir istisnadır.

Klasik deadlock’u yeniden üretmek

# Transaction 1                      # Transaction 2
UPDATE accounts SET ... WHERE id=1;  UPDATE accounts SET ... WHERE id=2;
UPDATE accounts SET ... WHERE id=2;  UPDATE accounts SET ... WHERE id=1;
# T1, T2'nin id=2 üzerindeki kilidini bekler   # T2, T1'in id=1 üzerindeki kilidini bekler → DEADLOCK

Python’da deadlock’ları yakalamak

import psycopg2
from psycopg2 import errors
import time

def transfer_with_deadlock_retry(conn, from_id, to_id, amount, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            with conn:
                cur = conn.cursor()
                cur.execute(
                    "UPDATE accounts SET balance = balance - %s WHERE id = %s",
                    (amount, from_id),
                )
                cur.execute(
                    "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                    (amount, to_id),
                )
            return
        except errors.DeadlockDetected:
            if attempt == max_attempts - 1:
                raise
            time.sleep(0.01 * (attempt + 1))

Önleme: tutarlı kilit sıralaması

En etkili tek deadlock önleme tekniği: tüm kod yolları boyunca kilitleri her zaman aynı sırada almak, örn. her zaman artan primary key sırasına göre.

def transfer(conn, id_a, id_b, amount):
    first, second = sorted([id_a, id_b])  # kanonik sıralama
    ...
    # her zaman `first`'ü `second`'dan önce kilitle/güncelle

14. Dağıtık Transaction’lar: 2PC ve Saga Pattern

İki Aşamalı Commit (Two-Phase Commit, 2PC)

DB-API 2.0, XA tarzı dağıtık transaction’ları destekleyen sürücüler için (örn. PostgreSQL’in PREPARE TRANSACTION‘ına karşı psycopg2) tpc_begin(), tpc_prepare(), tpc_commit(), tpc_rollback() sunar.

conn = psycopg2.connect(dsn)
xid = conn.xid(42, "transfer", "branch-1")

conn.tpc_begin(xid)
cur = conn.cursor()
cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")

conn.tpc_prepare()   # aşama 1: tüm katılımcılar hazırlanır
# --- koordinatör TÜM katılımcıların başarıyla hazırlandığını onaylar ---
conn.tpc_commit()    # aşama 2: commit
# veya herhangi bir katılımcı hazırlanamazsa conn.tpc_rollback()

Pratikte, 2PC modern Python mikroservis mimarilerinde nadiren kullanılır, çünkü:

Saga pattern: pragmatik alternatif

Servisler arasında atomiklik yerine, Saga; her biri, sonraki bir adım başarısız olursa geri almak için karşılık gelen bir telafi edici (compensating) transaction‘a sahip bir dizi yerel transaction’dır.

class TransferSaga:
    def execute(self, from_account, to_account, amount):
        steps_completed = []
        try:
            self.debit(from_account, amount)
            steps_completed.append(("credit", from_account, amount))

            self.credit(to_account, amount)
            steps_completed.append(("debit", to_account, amount))

            self.notify(from_account, to_account, amount)
        except Exception:
            self._compensate(steps_completed)
            raise

    def _compensate(self, steps_completed):
        for action, account, amount in reversed(steps_completed):
            getattr(self, action)(account, amount)  # örn. tekrar kredi/borç

Servisler arası olayların 2PC olmadan en az bir kez (at-least-once) teslimini garanti etmek için sagaları bir outbox tablosu ile birleştirin (“yayınlanacak olay"ı, iş değişikliğiyle aynı yerel transaction içinde yazın, sonra ayrı bir poller bunu yayınlar).

def create_order(session, order_data):
    with session.begin():
        order = Order(**order_data)
        session.add(order)
        session.add(OutboxEvent(
            event_type="OrderCreated",
            payload=json.dumps(order_data),
        ))
    # ayrı bir arka plan worker'ı OutboxEvent'i poll eder ve bir mesaj broker'ına yayınlar

15. Retry, Idempotency ve “Tam Olarak Bir Kez” Yanılsaması

Ağ üzerinden tam olarak bir kez (exactly-once) teslimat diye bir şey yoktur

İstemciniz bir yazma gönderir ve yanıt gelmeden önce bağlantı koparsa, yazmanın başarılı olup olmadığını bilemezsiniz. Tek sağlam mühendislik çözümü idempotency‘dir: işlemleri, yeniden denemenin güvenli olacağı şekilde tasarlayın.

Idempotency anahtarları

def process_payment(session, idempotency_key, amount, account_id):
    existing = session.query(PaymentAttempt).filter_by(
        idempotency_key=idempotency_key
    ).one_or_none()

    if existing is not None:
        return existing.result  # zaten işlenmiş — aynı sonucu döndür

    with session.begin():
        result = charge_account(account_id, amount)
        session.add(PaymentAttempt(
            idempotency_key=idempotency_key,
            result=result,
        ))
    return result

idempotency_key üzerindeki benzersizlik kısıtlaması veritabanı seviyesinde (UNIQUE index) uygulanmalıdır, sadece uygulama kodunda kontrol-edip-ekleme (kendisi bir yarış durumuna sahiptir) şeklinde değil — INSERT ... ON CONFLICT DO NOTHING kullanın veya IntegrityError‘ı yakalayın.

try:
    session.add(PaymentAttempt(idempotency_key=key, ...))
    session.commit()
except IntegrityError:
    session.rollback()
    # başka bir eşzamanlı istek bu anahtarı zaten eklemiş — sonucunu getirip döndürün

Backoff ile retry dekoratörleri (tenacity kullanarak)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from psycopg2 import errors

@retry(
    retry=retry_if_exception_type((errors.SerializationFailure, errors.DeadlockDetected)),
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=0.05, max=2),
)
def run_transaction(fn, conn):
    with conn:
        return fn(conn)

İdempotent olmayan işlemleri (örn. “otomatik oluşturulan bir ID ile yeni bir satır ekle”) bir idempotency anahtarı olmadan asla körü körüne yeniden denemeyin — mükerrer kayıtlar oluşturursunuz.

16. Transactional Kodun Test Edilmesi

“Her testi geri alınan (rolled-back) bir transaction’a sarma” deseni

Hızlı, izole veritabanı testleri için standart yaklaşım: her testten önce bir transaction (ve SQLAlchemy’de iç içe bir savepoint) başlatın, sonrasında geri alın — hiçbir veri gerçekten kalıcı olmaz ve testler birbirine karışmaz.

import pytest
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    session.begin_nested()  # SAVEPOINT, test kodu içindeki session.commit()'in dışarı taşmaması için

    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(sess, trans):
        if trans.nested and not trans._parent.nested:
            sess.begin_nested()

    yield session

    session.close()
    transaction.rollback()  # HER ŞEYİ geri al, "commit edilmiş" veri dâhil
    connection.close()

Django’nun TestCase‘i bunu otomatik yapar

from django.test import TestCase

class TransferTests(TestCase):
    # Django her test metodunu bir transaction'a sarar ve otomatik olarak geri alır
    def test_transfer_moves_funds(self):
        transfer_funds(self.acc1.id, self.acc2.id, 100)
        self.acc1.refresh_from_db()
        self.assertEqual(self.acc1.balance, 900)

Not: TestCase, transaction.atomic() çağıran ve gerçek commit/rollback sınırlarının önemli olmasını bekleyen kodu (örn. on_commit hook’larını test etmeyi) test edemez — bunun için, çok daha yavaş testler (gerçek commit’ler + testler arası tablo kırpma) pahasına TransactionTestCase kullanın.

from django.test import TransactionTestCase

class OnCommitHookTests(TransactionTestCase):
    def test_email_sent_only_after_commit(self):
        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            create_order(data)
        self.assertEqual(len(callbacks), 1)

Deadlock ve yarış durumlarını test etmek

Eşzamanlılık hataları, ortaya çıkmak için gerçek eşzamanlılık gerektirir. Gerçek bir (genellikle testcontainers ile konteynerize edilmiş) veritabanına karşı thread’ler veya multiprocessing kullanın:

import threading

def test_concurrent_updates_dont_lose_writes(pg_container):
    errors = []
    def worker():
        try:
            transfer_funds(acc1, acc2, 10)
        except Exception as e:
            errors.append(e)

    threads = [threading.Thread(target=worker) for _ in range(20)]
    for t in threads: t.start()
    for t in threads: t.join()

    # sıralamadan bağımsız olarak nihai bakiyenin doğru olduğunu doğrulayın

17. Migration ve Şema Evrimi (Alembic)

Autogenerate vs elle yazılan migration’lar

# alembic revision --autogenerate -m "add version column to accounts"

Autogenerate, ORM modellerinizi canlı şema ile karşılaştırır ve bir migration taslağı hazırlar — oluşturulan betiği her zaman gözden geçirin; autogenerate sıklıkla veri migration’larını, index yeniden adlandırmalarını ve check-constraint değişikliklerini kaçırır.

def upgrade():
    op.add_column("accounts", sa.Column("version_id", sa.Integer(), nullable=True))
    op.execute("UPDATE accounts SET version_id = 0")   # mevcut satırları geriye doldur
    op.alter_column("accounts", "version_id", nullable=False)

def downgrade():
    op.drop_column("accounts", "version_id")

Sıfır kesinti (zero-downtime) migration’lar çok adımlı deploy gerektirir

Canlı bir sistemde sıfır kesinti ile güvenli bir şekilde bir NOT NULL sütun eklemek genellikle üç ayrı deploy gerektirir:

  1. Sütunu bir varsayılan değerle nullable olarak ekleyin, yazan kodu deploy edin.
  2. Mevcut satırları (uzun kilitlerden kaçınmak için partiler hâlinde) geriye doldurun (backfill), okuyan kodu deploy edin.
  3. Backfill’in tamamlandığı doğrulandıktan sonra sütunu NOT NULL yapın.
# Adım 1
op.add_column("accounts", sa.Column("version_id", sa.Integer(), server_default="0"))

# Adım 2 (ayrı bir migration, tüm tabloyu kilitlememek için partiler hâlinde çalıştırılır)
op.execute("""
    UPDATE accounts SET version_id = 0
    WHERE version_id IS NULL
    AND id IN (SELECT id FROM accounts WHERE version_id IS NULL LIMIT 10000)
""")  # 0 satır etkilenene kadar tekrarlayın

# Adım 3 (ayrı bir migration, backfill'in tamamlandığı doğrulandıktan sonra)
op.alter_column("accounts", "version_id", nullable=False)

Büyük bir PostgreSQL tablosuna index eklemek, yazmaları kilitlememek için CREATE INDEX CONCURRENTLY kullanmalıdır — ama bu bir transaction içinde çalışamaz, bu yüzden bunu yapan Alembic migration’ları o revizyon için transactional DDL’i kapatmalıdır:

def upgrade():
    with op.get_context().autocommit_block():
        op.create_index(
            "ix_accounts_email", "accounts", ["email"],
            postgresql_concurrently=True,
        )

18. Performans Değerlendirmeleri

19. Sık Yapılan Hatalar Kontrol Listesi