SQLAlchemy 2.0: Core, ORM, ilişkiler ve transaction

Neden önemli?

sqlite3 ile Python dersinde veritabanına doğrudan SQL metni gönderdin. Küçük betiklerde bu yeterlidir. Ama gerçek bir uygulamada onlarca tablo, yüzlerce sorgu, farklı veritabanları (geliştirmede SQLite, üretimde PostgreSQL) ve iş nesneleri (müşteri, sipariş, hesap) vardır. Her sorgunun sonucunu elle sözlüğe çevirmek, her tabloda aynı INSERT kalıbını tekrar yazmak, bağlantıları ve transaction'ları doğru yönetmek zorlaşır.

SQLAlchemy, Python dünyasının en yaygın veritabanı araç takımıdır. İki katmanı vardır: Core (SQL'i Python ifadeleriyle kuran, veritabanından bağımsız katman) ve ORM (Object Relational Mapper: tablo satırlarını Python nesnelerine eşleyen katman). Flask, FastAPI, Airflow ve pandas (read_sql) gibi araçların hepsi SQLAlchemy ile konuşur. Bu derste 2.0 sürümünün modern API'sini kullanacağız; eski session.query(...) tarzını da yerinde göreceksin.

Engine, bağlantı ve text()

Her şey bir Engine ile başlar. Bağlantı adresi (URL) lehçe+sürücü://kullanıcı:şifre@sunucu/veritabanı biçimindedir. Bellek içi SQLite için sqlite+pysqlite:///:memory: yazarsın; PostgreSQL'e geçmek çoğu zaman yalnızca bu adresi değiştirmektir.

from sqlalchemy import create_engine, text

engine = create_engine("sqlite+pysqlite:///:memory:")

# engine.begin(): blok başarıyla biterse COMMIT, hata olursa ROLLBACK
with engine.begin() as conn:
    conn.execute(text("CREATE TABLE siparis (id INTEGER PRIMARY KEY, sehir TEXT, tutar REAL)"))
    conn.execute(
        text("INSERT INTO siparis (sehir, tutar) VALUES (:sehir, :tutar)"),
        [{"sehir": "İzmir", "tutar": 120.0}, {"sehir": "Ankara", "tutar": 300.0},
         {"sehir": "İzmir", "tutar": 80.0}],
    )

# engine.connect(): yalnız okuma; yazarsan conn.commit() gerekir
with engine.connect() as conn:
    sonuc = conn.execute(text("SELECT sehir, SUM(tutar) AS ciro FROM siparis "
                              "WHERE tutar >= :alt GROUP BY sehir ORDER BY ciro DESC"),
                         {"alt": 50})
    for satir in sonuc:
        print(satir.sehir, satir.ciro)       # satıra adla erişim
    print(conn.execute(text("SELECT COUNT(*) FROM siparis")).scalar())

:sehir gibi adlandırılmış yer tutucular bağlı parametrelerdir (bound parameters). Değerler SQL metnine yapıştırılmaz, sürücüye ayrıca gönderilir; bu yüzden SQL enjeksiyonu (SQL injection) mümkün değildir. Bir liste sözlük verdiğinde SQLAlchemy işlemi executemany olarak toplu çalıştırır.

Core: tabloyu Python nesnesiyle tanımlamak

Core'da tablo bir Table nesnesidir. Sorgular select, insert, update, delete fonksiyonlarıyla kurulur ve SQLAlchemy bunları kullandığın veritabanının lehçesine çevirir. Sorguyu print edersen üretilen SQL'i görürsün.

from sqlalchemy import (create_engine, MetaData, Table, Column, Integer, String,
                        Float, select, insert, func)

engine = create_engine("sqlite+pysqlite:///:memory:")
meta = MetaData()
urun = Table("urun", meta,
             Column("id", Integer, primary_key=True),
             Column("ad", String(80), nullable=False),
             Column("kategori", String(40)),
             Column("fiyat", Float))
meta.create_all(engine)

with engine.begin() as conn:
    conn.execute(insert(urun), [
        {"ad": "Çay", "kategori": "İçecek", "fiyat": 85.5},
        {"ad": "Kahve", "kategori": "İçecek", "fiyat": 240.0},
        {"ad": "Bisküvi", "kategori": "Atıştırmalık", "fiyat": 25.0},
    ])

sorgu = (select(urun.c.kategori, func.count().label("adet"), func.avg(urun.c.fiyat).label("ort"))
         .group_by(urun.c.kategori)
         .order_by(urun.c.kategori))
print(sorgu)                                 # üretilen SQL
with engine.connect() as conn:
    for k, adet, ort in conn.execute(sorgu):
        print(k, adet, round(ort, 2))

ORM: modeller, Mapped ve mapped_column

ORM'de her tablo bir Python sınıfıdır. 2.0 stilinde sınıflar DeclarativeBase'ten türeyen bir Base sınıfından türetilir; sütunlar Mapped[tip] tip ipucuyla (type hint) bildirilir. Mapped[str] NOT NULL, Mapped[str | None] boş bırakılabilir sütun demektir. Ek ayarlar (birincil anahtar, varsayılan, benzersizlik) mapped_column ile verilir. Metaprogramlama dersinde kurduğun mini model katmanını hatırla: SQLAlchemy de sınıf gövdesini inceleyip tabloyu böyle çıkarır.

from sqlalchemy import create_engine, select, func, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session


class Base(DeclarativeBase):
    pass


class Musteri(Base):
    __tablename__ = "musteri"
    id: Mapped[int] = mapped_column(primary_key=True)
    ad: Mapped[str]
    sehir: Mapped[str]
    siparisler: Mapped[list["Siparis"]] = relationship(back_populates="musteri")


class Siparis(Base):
    __tablename__ = "siparis"
    id: Mapped[int] = mapped_column(primary_key=True)
    tutar: Mapped[float]
    musteri_id: Mapped[int] = mapped_column(ForeignKey("musteri.id"))
    musteri: Mapped["Musteri"] = relationship(back_populates="siparisler")


engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)            # CREATE TABLE ifadelerini üretir
print(Musteri.__table__.c.keys(), Siparis.__table__.c.keys())
print(repr(Siparis.__table__.c.musteri_id.type), Musteri.__table__.c.ad.nullable)

relationship sütun değil, Python tarafındaki bağlantıdır: musteri.siparisler bir liste, siparis.musteri tek bir nesne verir. back_populates iki tarafı birbirine bağlar; birine eklediğin diğerinde de görünür. Asıl ilişkiyi veritabanında taşıyan şey ForeignKey("musteri.id") sütunudur.

Session ve CRUD

Session, ORM'in çalışma birimidir (unit of work): nesneleri izler, değişiklikleri toplar ve commit() anında doğru sırada INSERT/UPDATE/DELETE üretir. Aynı oturumda aynı birincil anahtarlı satır her zaman aynı Python nesnesidir (identity map).

from sqlalchemy import create_engine, select, func, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session


class Base(DeclarativeBase):
    pass


class Musteri(Base):
    __tablename__ = "musteri"
    id: Mapped[int] = mapped_column(primary_key=True)
    ad: Mapped[str]
    sehir: Mapped[str]
    siparisler: Mapped[list["Siparis"]] = relationship(back_populates="musteri")


class Siparis(Base):
    __tablename__ = "siparis"
    id: Mapped[int] = mapped_column(primary_key=True)
    tutar: Mapped[float]
    musteri_id: Mapped[int] = mapped_column(ForeignKey("musteri.id"))
    musteri: Mapped["Musteri"] = relationship(back_populates="siparisler")


engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    # CREATE
    ayse = Musteri(ad="Ayşe", sehir="İzmir")
    ayse.siparisler.append(Siparis(tutar=120.0))
    s.add_all([ayse, Musteri(ad="Mehmet", sehir="Ankara")])
    s.commit()
    print("id:", ayse.id)

with Session(engine) as s:
    # READ
    m = s.get(Musteri, 1)                              # birincil anahtarla
    izmirliler = s.scalars(select(Musteri).where(Musteri.sehir == "İzmir")).all()
    print(m.ad, [x.ad for x in izmirliler])
    # UPDATE: nesneyi değiştir, commit et
    m.sehir = "Manisa"
    # DELETE
    s.delete(s.scalars(select(Musteri).where(Musteri.ad == "Mehmet")).one())
    s.commit()
    print(s.scalars(select(Musteri.ad)).all(), m.sehir)
    print(s.scalar(select(func.count()).select_from(Siparis)))

scalars() her satırın ilk sütununu (burada model nesnesini) verir. Birden çok sütun seçtiğinde (select(Musteri.ad, Musteri.sehir)) s.execute(...) ile satır demetleri alırsın.

Lazy loading ve N+1 problemi

Varsayılan olarak ilişkiler tembel (lazy) yüklenir: musteri.siparisler'e ilk eriştiğinde ayrı bir SELECT çalışır. 100 müşteriyi listeleyip her birinin siparişlerine dokunursan 1 + 100 sorgu atarsın. Buna N+1 problemi denir ve ORM kullanan uygulamalardaki en yaygın performans sorunudur. Çözüm, ilişkiyi istekli (eager) yüklemektir.

from sqlalchemy import create_engine, select, func, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session


class Base(DeclarativeBase):
    pass


class Musteri(Base):
    __tablename__ = "musteri"
    id: Mapped[int] = mapped_column(primary_key=True)
    ad: Mapped[str]
    sehir: Mapped[str]
    siparisler: Mapped[list["Siparis"]] = relationship(back_populates="musteri")


class Siparis(Base):
    __tablename__ = "siparis"
    id: Mapped[int] = mapped_column(primary_key=True)
    tutar: Mapped[float]
    musteri_id: Mapped[int] = mapped_column(ForeignKey("musteri.id"))
    musteri: Mapped["Musteri"] = relationship(back_populates="siparisler")

from sqlalchemy import event
from sqlalchemy.orm import selectinload, joinedload

engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as s:
    s.add_all([
        Musteri(ad="Ayşe", sehir="İzmir", siparisler=[Siparis(tutar=120.0), Siparis(tutar=80.0)]),
        Musteri(ad="Mehmet", sehir="Ankara", siparisler=[Siparis(tutar=300.0)]),
        Musteri(ad="Zeynep", sehir="İzmir", siparisler=[Siparis(tutar=50.0)]),
        Musteri(ad="Can", sehir="Bursa"),
    ])
    s.commit()

sorgular = []
event.listen(engine, "before_cursor_execute",
             lambda conn, cur, sql, params, ctx, many: sorgular.append(sql))

def rapor(secenek=None):
    sorgular.clear()
    with Session(engine) as s:
        q = select(Musteri).order_by(Musteri.id)
        if secenek is not None:
            q = q.options(secenek)
        # joinedload koleksiyonda satırları çoğaltır; unique() tekrarları birleştirir
        satirlar = [(m.ad, len(m.siparisler)) for m in s.scalars(q).unique()]
    return satirlar, len(sorgular)

print("lazy        :", rapor())
print("selectinload:", rapor(selectinload(Musteri.siparisler)))
print("joinedload  :", rapor(joinedload(Musteri.siparisler))[1], "sorgu")

joinedload bire-çok ilişkide JOIN yüzünden her müşteriyi sipariş sayısı kadar satırda döndürür; bu yüzden 2.0'da sonucu .unique() ile tekilleştirmen zorunludur. selectinload'da böyle bir gereklilik yoktur; koleksiyonlar için çoğu zaman onu tercih etmen bundandır.

Transaction: ya hep ya hiç

Session her zaman bir transaction içindedir. commit() kalıcı yapar, rollback() geri alır. with Session(engine) as s, s.begin(): kalıbı blok başarıyla biterse commit, istisna olursa rollback yapar. Ilgaz Bank'ta havale gibi birden çok satırı değiştiren işlemler mutlaka tek transaction'da olmalı.

from sqlalchemy import create_engine, select, CheckConstraint
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Hesap(Base):
    __tablename__ = "hesap"
    no: Mapped[str] = mapped_column(primary_key=True)
    bakiye: Mapped[float] = mapped_column(CheckConstraint("bakiye >= 0"))

engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as s, s.begin():
    s.add_all([Hesap(no="TR-1", bakiye=1000), Hesap(no="TR-2", bakiye=50)])

def havale(gonderen, alici, tutar):
    with Session(engine) as s, s.begin():          # hata olursa otomatik rollback
        g, a = s.get(Hesap, gonderen), s.get(Hesap, alici)
        a.bakiye += tutar
        g.bakiye -= tutar
        if g.bakiye < 0:
            raise ValueError("Yetersiz bakiye")

havale("TR-1", "TR-2", 300)
try:
    havale("TR-2", "TR-1", 9999)
except ValueError as h:
    print("İptal:", h)
with Session(engine) as s:
    print([(h.no, h.bakiye) for h in s.scalars(select(Hesap).order_by(Hesap.no))])

İkinci havalede önce alıcıya para eklendi, sonra hata fırladı; ama rollback sayesinde hiçbir değişiklik kalıcı olmadı. Ayrıca veritabanındaki CHECK kısıtı ikinci bir savunma hattıdır: Python kodunda hata olsa bile negatif bakiye kaydedilemez.

Alembic ile migration (gösterim)

create_all yalnız eksik tabloları oluşturur; mevcut tabloya sütun eklemez. Üretimde şema değişiklikleri Alembic ile sürümlenir: her değişiklik bir migration dosyasıdır, git'te saklanır, tüm ortamlara aynı sırayla uygulanır.

pip install alembic
alembic init migrations              # migrations/ klasörü ve alembic.ini
# alembic.ini: sqlalchemy.url = postgresql+psycopg://...
# migrations/env.py: target_metadata = Base.metadata
alembic revision --autogenerate -m "musteriye telefon ekle"
alembic upgrade head                 # bekleyen tüm migration'ları uygula
alembic downgrade -1                 # son migration'ı geri al

# migrations/versions/3f2a_musteriye_telefon_ekle.py (otomatik üretilir, gözden geçirilir)
def upgrade():
    op.add_column("musteri", sa.Column("telefon", sa.String(20), nullable=True))

def downgrade():
    op.drop_column("musteri", "telefon")

Ham SQL mi, ORM mi?

"""musteriye telefon ekle"""
from alembic import op
import sqlalchemy as sa

revision = "b7c2e1"
down_revision = "a41f09"      # bir önceki halka; ilk migration'da None

def upgrade() -> None:
    op.add_column("musteri", sa.Column("telefon", sa.String(20), nullable=True))
    op.create_index("ix_musteri_telefon", "musteri", ["telefon"])

def downgrade() -> None:
    op.drop_index("ix_musteri_telefon", table_name="musteri")
    op.drop_column("musteri", "telefon")

import veri
conn = veri.baglan("anadolu_market", tablolar=["customers"])
conn.execute("CREATE TABLE alembic_version (version_num TEXT)")
conn.execute("INSERT INTO alembic_version VALUES ('base')")

KODLAR = {"İstanbul": 34, "Ankara": 6, "İzmir": 35, "Bursa": 16}

def m001_upgrade(c):
    c.execute("ALTER TABLE customers ADD COLUMN plaka INTEGER")    # 1) boş (nullable) sütun

def m002_upgrade(c):                                              # 2) veri migration'ı
    for sehir, kod in KODLAR.items():
        c.execute("UPDATE customers SET plaka = ? WHERE city = ?", (kod, sehir))

zincir = [("m001", m001_upgrade), ("m002", m002_upgrade)]
for rev, fn in zincir:
    with conn:                                   # her halka kendi transaction'ında
        fn(conn)
        conn.execute("UPDATE alembic_version SET version_num = ?", (rev,))

print(conn.execute("SELECT version_num FROM alembic_version").fetchone()[0])
print(conn.execute("SELECT plaka, COUNT(*) FROM customers GROUP BY plaka ORDER BY plaka").fetchall())

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import selectinload

engine = create_async_engine("postgresql+asyncpg://app:gizli@db/market", pool_size=10)
Oturum = async_sessionmaker(engine, expire_on_commit=False)

async def musteri_siparisleri(musteri_id: int):
    async with Oturum() as s:
        sorgu = (select(Musteri)
                 .where(Musteri.id == musteri_id)
                 .options(selectinload(Musteri.siparisler)))   # tembel yükleme yok: önceden yükle
        m = (await s.execute(sorgu)).scalar_one()
        return [x.tutar for x in m.siparisler]

async def tablolari_kur():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)   # senkron API'yi async içinde çalıştır