PostgreSQL ile Python: psycopg, transaction, indeks ve toplu yükleme

Neden PostgreSQL?

Anadolu Market'in ilk sürümü tek bir SQLite dosyasıyla çalıştı. Sonra mobil uygulama, mağaza kasaları ve depo sistemi aynı anda sipariş yazmaya başladı. SQLite tek dosya ve tek yazar (single writer) mantığıyla tasarlanmıştır: aynı anda iki işlem yazmak isterse biri bekler, ağ üzerinden erişim yoktur, kullanıcı ve yetki kavramı yoktur. Bu noktada ekipler çoğunlukla PostgreSQL'e geçer: istemci-sunucu (client-server) mimarisinde çalışan, aynı anda yüzlerce bağlantıyı MVCC (çok sürümlü eşzamanlılık denetimi, multi-version concurrency control) ile yöneten, rolleri, şemaları, zengin türleri (JSONB, dizi, aralık, UUID) ve gelişmiş indeksleri olan açık kaynak ilişkisel veritabanı.

Bu derste PostgreSQL'e Python'dan psycopg 3 ile bağlanmayı, transaction ve kilitleri, ON CONFLICT ile upsert'i, RETURNING'i, COPY ile toplu yüklemeyi, indeks ve EXPLAIN okumayı öğreneceksin. SQL'in temelini sqlite-python dersinde gördün; ORM katmanı sqlalchemy-orm dersinde gelecek. Bu ders ikisinin arasındaki köprüdür: ham sürücü (driver) ile üretim veritabanı.

SQLite ile PostgreSQL arasındaki farklar

Docker'da PostgreSQL başlatmak

Bilgisayarına PostgreSQL kurmak yerine resmi Docker imajını çalıştırmak en temiz yoldur (Docker'ı docker-deploy dersinde göreceksin):

docker run --name pg-anadolu -e POSTGRES_PASSWORD=gizli -e POSTGRES_DB=market \
  -p 5432:5432 -d postgres:16
docker exec -it pg-anadolu psql -U postgres -d market   # SQL kabuğu

Bağlantı bilgisi tek bir metinle (DSN ya da bağlantı URL'si) verilir: postgresql://postgres:gizli@localhost:5432/market. Parolayı koda yazma; ortam değişkeninden oku (yapilandirma dersi).

psycopg 3: bağlantı, imleç ve parametreler

import os
import psycopg
from psycopg.rows import dict_row

DSN = os.environ["DATABASE_URL"]

with psycopg.connect(DSN, row_factory=dict_row) as con:      # çıkışta commit (hata varsa rollback) + kapanış
    with con.cursor() as cur:
        cur.execute(
            "SELECT customer_id, city FROM customers WHERE city = %s AND segment = %s",
            ("İzmir", "Premium"),
        )
        for satir in cur.fetchall():
            print(satir["customer_id"], satir["city"])      # dict_row: sözlük gibi erişim

Üç kural: (1) değerleri her zaman parametre olarak ver, asla f-string ile SQL'e gömme (SQL enjeksiyonu); (2) %s Python'un yüzde biçimlendirmesi değildir, psycopg'nin yer tutucusudur ve tırnaksız yazılır; (3) bağlantıyı with ile aç ki hata olsa da kapansın. Aynı fikir sqlite3'te ? ve row_factory ile çalışır:

import sqlite3
import veri

con = veri.baglan("anadolu_market", tablolar=["customers"])
con.row_factory = sqlite3.Row                  # psycopg'deki dict_row'un karşılığı
sql = "SELECT customer_id, city, segment FROM customers WHERE city = ? AND segment = ? ORDER BY customer_id LIMIT 3"
for satir in con.execute(sql, ("İzmir", "Premium")):
    print(dict(satir))

Bağlantı havuzu

Her HTTP isteğinde yeni bağlantı açmak pahalıdır (TCP, kimlik doğrulama, sunucuda yeni süreç). Web uygulamaları bir bağlantı havuzu (connection pool) kullanır: birkaç bağlantı açık tutulur, istek gelince ödünç verilir, iş bitince geri alınır.

from psycopg_pool import ConnectionPool

havuz = ConnectionPool(DSN, min_size=2, max_size=10, timeout=5)   # 5 sn içinde boş bağlantı yoksa PoolTimeout

def musteri_getir(musteri_id):
    with havuz.connection() as con:             # ödünç al; blok bitince commit + havuza iade
        return con.execute("SELECT * FROM customers WHERE customer_id = %s", (musteri_id,)).fetchone()

Transaction, izolasyon ve kilitler

Transaction (işlem bütünlüğü), birden çok SQL komutunun "ya hep ya hiç" çalışmasıdır. psycopg'de bağlantı açıldığında ilk komutla transaction kendiliğinden başlar; with psycopg.connect(...) bloğu başarıyla biterse commit, hata çıkarsa rollback yapılır. Bir bağlantı içinde ayrı bir alt blok için with con.transaction(): kullanılır.

Klasik yarış durumu (race condition): iki müşteri son kalan ürünü aynı anda alıyor. İkisi de "stok 1" okur, ikisi de 0'a düşürür, depo eksiye girer. PostgreSQL'de çözüm satırı okurken kilitlemektir: SELECT ... FOR UPDATE. Aynı satırı kilitlemek isteyen ikinci işlem birincinin commit'ini bekler. Havayolu koltuk seçimi gibi kuyruk işlerinde FOR UPDATE SKIP LOCKED kilitli satırları atlayıp boş olana geçer.

with psycopg.connect(DSN) as con:
    with con.transaction():
        stok = con.execute(
            "SELECT stock FROM products WHERE product_id = %s FOR UPDATE", (urun_id,)
        ).fetchone()[0]
        if stok < adet:
            raise ValueError("Yetersiz stok")
        con.execute("UPDATE products SET stock = stock - %s WHERE product_id = %s", (adet, urun_id))

Çoğu zaman daha basit ve aynı derecede güvenli bir yol vardır: koşullu güncelleme. Kontrolü ve değişikliği tek bir atomik UPDATE içine koyar, kaç satırın etkilendiğine bakarsın. Bu kalıp SQLite'ta da, PostgreSQL'de de çalışır:

import veri

con = veri.baglan("anadolu_market", tablolar=["products"])

def stok_dus(con, urun_id, adet):
    with con:                                   # commit ya da hata olursa rollback
        cur = con.execute(
            "UPDATE products SET stock = stock - ? WHERE product_id = ? AND stock >= ?",
            (adet, urun_id, adet),
        )
    return cur.rowcount == 1                    # 0 ise stok yetmedi (ya da ürün yok)

print("ilk stok:", con.execute("SELECT stock FROM products WHERE product_id = 1").fetchone()[0])
print(stok_dus(con, 1, 70), stok_dus(con, 1, 10))
print("son stok:", con.execute("SELECT stock FROM products WHERE product_id = 1").fetchone()[0])

Upsert ve RETURNING

Tedarikçiden fiyat listesi geliyor: bazı ürünler yeni, bazıları mevcut. "Varsa güncelle, yoksa ekle" işlemine upsert denir. PostgreSQL ve SQLite aynı sözdizimini kullanır: INSERT ... ON CONFLICT (anahtar) DO UPDATE SET .... excluded eklenmek istenen satırdır. RETURNING ise eklenen ya da güncellenen satırı ayrı bir SELECT'e gerek kalmadan geri döndürür; sunucunun ürettiği id'yi almanın standart yoludur.

import veri

con = veri.baglan("anadolu_market", tablolar=["price_updates"])
con.execute("CREATE TABLE guncel_fiyat (product_id INTEGER PRIMARY KEY, fiyat REAL, guncellendi TEXT)")

guncellemeler = con.execute(
    "SELECT product_id, new_price, updated_at FROM price_updates ORDER BY updated_at"
).fetchall()
with con:
    for urun, fiyat, zaman in guncellemeler:
        con.execute("""
            INSERT INTO guncel_fiyat VALUES (?, ?, ?)
            ON CONFLICT (product_id) DO UPDATE
               SET fiyat = excluded.fiyat, guncellendi = excluded.guncellendi
             WHERE excluded.guncellendi > guncel_fiyat.guncellendi
        """, (urun, fiyat, zaman))

print(len(guncellemeler), "güncelleme ->", con.execute("SELECT COUNT(*) FROM guncel_fiyat").fetchone()[0], "ürün")
print(con.execute("SELECT * FROM guncel_fiyat WHERE product_id IN (30, 31) ORDER BY product_id").fetchall())

cur = con.execute("INSERT INTO guncel_fiyat VALUES (99, 120.0, '2025-01-01') RETURNING product_id, fiyat")
print("RETURNING:", cur.fetchall())

Upsert'teki WHERE koşulu çok önemlidir: dosyalar geç ya da sırasız gelirse eski fiyatın yenisinin üstüne yazılmasını engeller. Böylece aynı dosyayı iki kez yüklemek de sonucu değiştirmez (idempotent yükleme, veri-boru-hatti).

Toplu yükleme: COPY ve executemany

Satır satır INSERT her satır için sunucuya bir gidiş-dönüş demektir. 100 bin satırda bu dakikalar sürer. Hız sıralaması kabaca şöyledir: döngüde tek tek execute < executemany < çok satırlı VALUES < COPY. psycopg 3'te COPY Python'dan akış olarak beslenebilir:

with psycopg.connect(DSN) as con, con.cursor() as cur:
    with cur.copy("COPY order_items (item_id, order_id, product_id, quantity, unit_price, discount) FROM STDIN") as cp:
        for satir in satirlar:            # demetler
            cp.write_row(satir)

DuckDB'de de aynı COPY ... FROM komutu vardır ve CSV dosyasını tek komutla tabloya yükler. Gerçek veriyle dene:

import duckdb
import veri

con = duckdb.connect()
con.execute("""CREATE TABLE siparis (order_id INTEGER PRIMARY KEY, customer_id INTEGER,
               order_date DATE, status TEXT, channel TEXT)""")
con.execute(f"COPY siparis FROM '{veri.yol('anadolu_market', 'orders')}' (HEADER)")
print(con.execute("SELECT status, COUNT(*) FROM siparis GROUP BY status ORDER BY status").fetchall())

İndeks ve EXPLAIN

İndeks, kitabın sonundaki dizin gibidir: tüm sayfaları taramak yerine doğrudan ilgili yere gider. PostgreSQL'de en sık iki tür:

• B-tree (varsayılan): eşitlik, aralık (<, BETWEEN), sıralama. customer_id, order_date gibi sütunlar.
• GIN: bir değerin içindeki parçaları indeksler; JSONB anahtarları, diziler ve tam metin arama için.

EXPLAIN ANALYZE sorguyu gerçekten çalıştırır ve planı süreleriyle gösterir. Seq Scan tüm tabloyu taramak, Index Scan indeksle gitmektir. Tahmini satır (rows=) ile gerçek satır arasında büyük fark varsa istatistikler eskidir (ANALYZE tablo).

CREATE INDEX ix_orders_customer ON orders (customer_id);
EXPLAIN ANALYZE SELECT * FROM orders WHERE customer_id = 42;
--  Index Scan using ix_orders_customer on orders  (cost=0.28..8.40 rows=6 width=40)
--                                      (actual time=0.015..0.019 rows=6 loops=1)
--  Planning Time: 0.08 ms   Execution Time: 0.04 ms

SQLite'ta karşılığı EXPLAIN QUERY PLAN'dır: SCAN tam tarama, SEARCH ... USING INDEX indeks kullanımıdır.

import veri

con = veri.baglan("anadolu_market", tablolar=["orders"])
sql = "SELECT * FROM orders WHERE customer_id = ?"
plan = lambda: con.execute("EXPLAIN QUERY PLAN " + sql, (42,)).fetchall()[0][-1]
print("önce :", plan())
con.execute("CREATE INDEX ix_orders_customer ON orders (customer_id)")
print("sonra:", plan())
print(len(con.execute(sql, (42,)).fetchall()), "sipariş")

JSONB ve pgvector'a kısa bakış

PostgreSQL yarı yapılandırılmış veriyi jsonb sütununda ikili biçimde saklar ve GIN ile indeksler. Örneğin ürün özellikleri: WHERE ozellik @> '{"renk": "siyah"}'. pgvector eklentisi ise embedding vektörlerini saklayıp en yakın komşu araması yapar; RAG uygulamalarında ayrı bir vektör veritabanı yerine sık seçilir. SQLite'ta da JSON fonksiyonları vardır:

import json
import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE urun (id INTEGER PRIMARY KEY, ozellik TEXT)")
con.executemany("INSERT INTO urun VALUES (?, ?)", [
    (1, json.dumps({"renk": "siyah", "garanti": 2})),
    (2, json.dumps({"renk": "beyaz", "garanti": 1})),
])
print(con.execute("SELECT id, json_extract(ozellik, '$.garanti') FROM urun WHERE json_extract(ozellik, '$.renk') = ?", ("siyah",)).fetchall())

Migration disiplini, sızıntı ve zaman aşımı

Üretim veritabanında tabloyu elle ALTER etmezsin. Her şema değişikliği sürüm numaralı bir migration dosyasıdır ve kodla birlikte gözden geçirilir; Python dünyasında araç Alembic'tir (sqlalchemy-orm dersinde). İki işletim hatası da sık görülür:

• Bağlantı sızıntısı: with kullanmadan açılan ve hata yolunda kapanmayan bağlantılar havuzu tüketir; uygulama "PoolTimeout" ile kilitlenir.
• Açık kalan transaction: idle in transaction durumundaki bağlantı kilitleri tutar, diğer işlemleri bekletir.

psycopg.connect(DSN, connect_timeout=5, options="-c statement_timeout=3000")   # 3 sn'den uzun sorguyu iptal et
-- sunucuda: ALTER ROLE uygulama SET idle_in_transaction_session_timeout = '30s';