Veritabanı ve ORM: SQLAlchemy

Önceki derslerde sqlite3 modülüyle SQL komutları yazdık, tablolar oluşturduk, veri ekledik. İşe yaradı — ama bir sorun var. SQL string'lerini Python kodunun içine gömmek, projen büyüdükçe kabusa dönüşür. Tablodan bir sütun eklesen tüm sorguları güncellemelisin. SQL injection riski her yerde seni bekliyor. Ve en önemlisi: Python'da nesnelerle çalışıyorsun ama veritabanıyla konuşurken aniden string birleştirmeye geçiyorsun.

İşte ORM (Object-Relational Mapping) tam bu sorunu çözer. Python nesneleriyle veritabanı tablolarını birbirine eşler — sen nesne üzerinde çalışırsın, ORM bunu SQL'e çevirir. Python dünyasının en güçlü ORM'si ise SQLAlchemy'dir.

1. Neden ORM Kullanırız?

🔌 Analoji: Evrensel Adaptör

Yurt dışına gittin ve yanında Türkiye'ye uygun bir şarj aleti var. Ama prizler farklı. Bir adaptör alırsın — senin fişin aynı kalır, adaptör onu o ülkenin prizine uygun hale getirir.

ORM de aynen böyle çalışır. Senin "fişin" Python nesneleri, "priz" ise veritabanı tabloları. ORM, aradaki uyumsuzluğu giderin adaptörüdür. Hangi veritabanını kullanırsan kullan (SQLite, PostgreSQL, MySQL), sen hep Python nesneleriyle çalışırsın.

Raw SQL vs ORM

Raw SQL ile kullanıcı ekleme:

import sqlite3

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

# SQL injection riski — dikkat!
name = "Ahmet"
email = "ahmet@test.com"
cursor.execute(
    "INSERT INTO users (name, email) VALUES (?, ?)",
    (name, email)
)
conn.commit()

# Kullanıcı çekme
cursor.execute("SELECT id, name, email FROM users WHERE id = ?", (1,))
row = cursor.fetchone()
# row bir tuple: (1, 'Ahmet', 'ahmet@test.com')
# row[0] nedir? row[1]? Koda bakınca anlamak zor.

ORM ile aynı işlem:

from sqlalchemy.orm import Session

# Yeni kullanıcı — Python nesnesi oluştur
user = User(name="Ahmet", email="ahmet@test.com")
session.add(user)
session.commit()

# Kullanıcı çekme
user = session.get(User, 1)
print(user.name)   # "Ahmet" — anlamlı isimlerle erişim
print(user.email)  # "ahmet@test.com"

Fark açık:

ORM her zaman doğru seçim mi? Hayır. Çok karmaşık raporlama sorguları, bulk işlemler veya performansın kritik olduğu durumlarda raw SQL daha iyi olabilir. Ama günlük CRUD işlemlerinin %90'ında ORM hayat kurtarır.

2. SQLAlchemy Mimarisi: Core vs ORM

SQLAlchemy aslında iki katmandan oluşur:

SQLAlchemy Core

Düşük seviyeli katman. SQL sorgularını Python nesneleriyle oluşturursun ama hâlâ tablolar ve sütunlarla çalışırsın:

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, select

engine = create_engine("sqlite:///app.db")
metadata = MetaData()

# Tablo tanımı — ama class değil
users = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("email", String(100))
)

# Tablo oluştur
metadata.create_all(engine)

# Sorgu — SQL benzeri ama Python
stmt = select(users).where(users.c.name == "Ahmet")
with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(row.name, row.email)

Core, SQL'i Python syntax'ıyla yazmana olanak tanır. Framework yazanlar, karmaşık sorgular gerektirenler veya ORM overhead'ini istemeyenler için uygundur.

SQLAlchemy ORM

Yüksek seviyeli katman. Python class'larını tablolara eşler. Nesnelerle çalışırsın:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100))

Bu derste ağırlıklı olarak ORM katmanını kullanacağız — çünkü günlük geliştirmede en çok buna ihtiyacın olacak.

3. Kurulum ve Temel Yapı

Kurulum

pip install sqlalchemy

PostgreSQL veya MySQL kullanacaksan ek sürücü gerekir:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install pymysql

SQLite için ek kurulum gerekmez — Python'un yerleşik sqlite3 modülünü kullanır.

Engine: Veritabanı Bağlantısı

Engine, veritabanına bağlantı kuran ana nesnedir. Bağlantı URL'si formatı:

dialect+driver://username:password@host:port/database

from sqlalchemy import create_engine

# SQLite — dosya tabanlı
engine = create_engine("sqlite:///app.db", echo=True)

# SQLite — bellekte (test için ideal)
engine = create_engine("sqlite:///:memory:", echo=True)

# PostgreSQL
engine = create_engine("postgresql://user:pass@localhost:5432/mydb")

# MySQL
engine = create_engine("mysql+pymysql://user:pass@localhost:3306/mydb")

echo=True parametresi SQLAlchemy'nin ürettiği SQL sorgularını konsola yazdırır — geliştirme aşamasında çok faydalı.

Session: Veritabanı Oturumu

Session, veritabanıyla konuştuğun penceredir. Tüm CRUD işlemlerini session üzerinden yaparsın:

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

engine = create_engine("sqlite:///app.db")

# Session factory oluştur
SessionLocal = sessionmaker(bind=engine)

# Kullanım 1: Manuel
session = SessionLocal()
try:
    # ... işlemler
    session.commit()
except:
    session.rollback()
    raise
finally:
    session.close()

# Kullanım 2: Context manager (önerilen)
with Session(engine) as session:
    # ... işlemler
    session.commit()
# session otomatik kapatılır

💡 İpucu: with Session(engine) as session kullanmak kaynakların düzgün temizlenmesini garanti eder. Hata olsa bile session kapanır. Her zaman bu yöntemi tercih et.

4. Model Tanımlama

SQLAlchemy 2.0'da modern Python type hint'leri ile model tanımlamak çok temiz:

from datetime import datetime
from typing import Optional
from sqlalchemy import String, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)
    bio: Mapped[Optional[str]] = mapped_column(Text, default=None)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now()
    )
    
    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

Burada neler oluyor:

• __tablename__ → Veritabanındaki tablo adı

• Mapped[int] → Bu sütunun Python tipini belirtir

• mapped_column(primary_key=True) → Primary key

• String(50) → VARCHAR(50) — maksimum 50 karakter

• unique=True → Bu sütundaki değerler tekrarlanamaz

• Optional[str] → Bu sütun NULL olabilir

• server_default=func.now() → Veritabanı seviyesinde varsayılan değer

Tabloları Oluşturma

from sqlalchemy import create_engine

engine = create_engine("sqlite:///app.db", echo=True)

# Tüm modellerin tablolarını oluştur
Base.metadata.create_all(engine)

create_all() modelleri tarar ve eksik tabloları oluşturur. Zaten var olan tabloları değiştirmez — migration için Alembic kullanman gerekir (dersin sonunda göreceğiz).

5. CRUD İşlemleri

Create (Oluşturma)

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Tek kayıt ekleme
    user = User(name="Ahmet", email="ahmet@test.com")
    session.add(user)
    session.commit()
    
    print(f"Yeni kullanıcı ID: {user.id}")  # commit'ten sonra ID atanır
    
    # Toplu ekleme
    users = [
        User(name="Ayşe", email="ayse@test.com"),
        User(name="Mehmet", email="mehmet@test.com"),
        User(name="Fatma", email="fatma@test.com"),
    ]
    session.add_all(users)
    session.commit()

session.add() nesneyi session'a ekler ama veritabanına henüz yazmaz. session.commit() çağrıldığında tüm değişiklikler tek seferde veritabanına yazılır — bu bir transaction.

Read (Okuma)

from sqlalchemy import select

with Session(engine) as session:
    # Primary key ile getir — en hızlı yöntem
    user = session.get(User, 1)
    print(user.name)  # "Ahmet"
    
    # select() ile sorgu oluştur
    stmt = select(User).where(User.name == "Ayşe")
    user = session.execute(stmt).scalar_one_or_none()
    
    # Tüm kullanıcıları getir
    stmt = select(User).order_by(User.name)
    users = session.execute(stmt).scalars().all()
    for u in users:
        print(f"{u.id}: {u.name} ({u.email})")
    
    # İlk eşleşeni getir
    stmt = select(User).where(User.is_active == True)
    first_active = session.execute(stmt).scalars().first()

session.get(User, 1) primary key ile doğrudan erişir — en performanslı yöntemdir. select() ise karmaşık sorgular için kullanılır.

scalar_one_or_none() — Tam bir sonuç döner ya da None. Birden fazla sonuç varsa hata fırlatır. scalars().all() — Tüm sonuçları liste olarak döner. scalars().first() — İlk sonucu döner ya da None.

Update (Güncelleme)

with Session(engine) as session:
    # Yöntem 1: Nesneyi getir, değiştir, commit'le
    user = session.get(User, 1)
    if user:
        user.name = "Ahmet Yılmaz"
        user.email = "ahmet.yilmaz@test.com"
        session.commit()  # Değişiklikler otomatik algılanır
    
    # Yöntem 2: Toplu güncelleme (bulk update)
    from sqlalchemy import update
    
    stmt = (
        update(User)
        .where(User.is_active == False)
        .values(bio="Pasif hesap")
    )
    session.execute(stmt)
    session.commit()

SQLAlchemy nesne üzerindeki değişiklikleri otomatik takip eder — buna dirty tracking denir. user.name = "yeni isim" yaptığında session bunu algılar ve commit'te uygun SQL'i üretir.

Delete (Silme)

with Session(engine) as session:
    # Tek kayıt silme
    user = session.get(User, 1)
    if user:
        session.delete(user)
        session.commit()
    
    # Toplu silme
    from sqlalchemy import delete
    
    stmt = delete(User).where(User.is_active == False)
    result = session.execute(stmt)
    session.commit()
    print(f"{result.rowcount} kayıt silindi")

⚠️ Dikkat: session.delete() ile silinen nesneyi commit'ten sonra kullanmaya çalışma — DetachedInstanceError alırsın. Silinen nesneye referans tutma.

6. Filtreleme, Sıralama ve Gruplama

SQLAlchemy'nin sorgu API'si çok güçlüdür. SQL'de yapabildiğin neredeyse her şeyi Python'da yapabilirsin.

Filtreleme (WHERE)

from sqlalchemy import select, and_, or_, not_

with Session(engine) as session:
    # Eşitlik
    stmt = select(User).where(User.name == "Ahmet")
    
    # Karşılaştırma
    stmt = select(User).where(User.id > 5)
    stmt = select(User).where(User.id.between(3, 10))
    
    # LIKE — metin arama
    stmt = select(User).where(User.name.like("%met%"))
    stmt = select(User).where(User.name.ilike("%MET%"))  # Case-insensitive
    
    # IN — liste içinde arama
    stmt = select(User).where(User.id.in_([1, 3, 5, 7]))
    
    # NULL kontrolü
    stmt = select(User).where(User.bio.is_(None))
    stmt = select(User).where(User.bio.is_not(None))
    
    # AND — birden fazla koşul
    stmt = select(User).where(
        and_(
            User.is_active == True,
            User.name.like("A%")
        )
    )
    # Veya zincirleme where (otomatik AND)
    stmt = select(User).where(User.is_active == True).where(User.name.like("A%"))
    
    # OR
    stmt = select(User).where(
        or_(
            User.name == "Ahmet",
            User.name == "Ayşe"
        )
    )
    
    # NOT
    stmt = select(User).where(not_(User.is_active))
    
    results = session.execute(stmt).scalars().all()

Sıralama (ORDER BY)

from sqlalchemy import select, desc

with Session(engine) as session:
    # Artan sıra (varsayılan)
    stmt = select(User).order_by(User.name)
    
    # Azalan sıra
    stmt = select(User).order_by(desc(User.created_at))
    
    # Birden fazla sıralama kriteri
    stmt = select(User).order_by(User.is_active.desc(), User.name.asc())
    
    users = session.execute(stmt).scalars().all()

Gruplama ve Aggregate (GROUP BY)

from sqlalchemy import select, func

with Session(engine) as session:
    # Toplam kullanıcı sayısı
    stmt = select(func.count(User.id))
    count = session.execute(stmt).scalar()
    print(f"Toplam: {count} kullanıcı")
    
    # Aktif/pasif kullanıcı sayısı (GROUP BY)
    stmt = (
        select(User.is_active, func.count(User.id))
        .group_by(User.is_active)
    )
    results = session.execute(stmt).all()
    for is_active, count in results:
        status = "Aktif" if is_active else "Pasif"
        print(f"{status}: {count}")
    
    # HAVING — grup filtresi
    # "2'den fazla kullanıcı olan domain'ler"
    domain = func.substr(User.email, func.instr(User.email, "@") + 1)
    stmt = (
        select(domain.label("domain"), func.count(User.id).label("total"))
        .group_by(domain)
        .having(func.count(User.id) > 2)
    )
    results = session.execute(stmt).all()

Sayfalama (Pagination)

with Session(engine) as session:
    page = 2
    per_page = 10
    
    stmt = (
        select(User)
        .order_by(User.id)
        .offset((page - 1) * per_page)
        .limit(per_page)
    )
    users = session.execute(stmt).scalars().all()

offset() ve limit() ile klasik sayfalama yapabilirsin. Büyük veri setlerinde offset yerine cursor-based pagination (son ID'ye göre filtreleme) daha performanslıdır.

7. İlişkiler (Relationships)

Gerçek uygulamalarda tablolar birbirleriyle ilişkilidir. Bir kullanıcının birden fazla yazısı olabilir. Bir yazının birden fazla etiketi olabilir. SQLAlchemy bu ilişkileri Python nesneleri arasında da kurar.

One-to-Many (Bire-Çok)

Bir kullanıcının birçok yazısı var. Her yazı bir kullanıcıya ait.

from sqlalchemy import String, Text, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    
    # İlişki: Kullanıcının yazıları
    posts: Mapped[list["Post"]] = relationship(back_populates="author")
    
    def __repr__(self):
        return f"<User(name='{self.name}')>"

class Post(Base):
    __tablename__ = "posts"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)
    
    # Foreign key — veritabanı seviyesinde bağlantı
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    
    # İlişki: Yazının yazarı
    author: Mapped["User"] = relationship(back_populates="posts")
    
    def __repr__(self):
        return f"<Post(title='{self.title}')>"

Kullanımı:

with Session(engine) as session:
    # Kullanıcı ve yazılarını birlikte oluştur
    user = User(name="Ahmet")
    user.posts = [
        Post(title="Python Öğreniyorum", content="Harika bir dil..."),
        Post(title="SQLAlchemy Rehberi", content="ORM çok kolaylaştırıyor...")
    ]
    session.add(user)
    session.commit()
    
    # Kullanıcının yazılarına eriş
    ahmet = session.execute(
        select(User).where(User.name == "Ahmet")
    ).scalar_one()
    
    for post in ahmet.posts:
        print(f"  - {post.title}")
    
    # Yazıdan yazara eriş
    post = session.get(Post, 1)
    print(f"Yazar: {post.author.name}")

relationship() Python tarafındaki bağlantıyı, ForeignKey ise veritabanı tarafındaki bağlantıyı kurar. back_populates iki yönlü ilişkiyi sağlar — user.posts listesi ve post.author referansı birbirini bilir.

Many-to-Many (Çoka-Çok)

Bir yazının birden fazla etiketi, bir etiketin birden fazla yazısı olabilir. Bunun için ara tablo (association table) kullanılır.

from sqlalchemy import Table, Column, Integer, ForeignKey, String

# Ara tablo — model değil, sadece tablo
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30), unique=True)
    
    # İlişki
    posts: Mapped[list["Post"]] = relationship(
        secondary=post_tags, back_populates="tags"
    )

# Post modeline ekle:
class Post(Base):
    __tablename__ = "posts"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    
    author: Mapped["User"] = relationship(back_populates="posts")
    tags: Mapped[list["Tag"]] = relationship(
        secondary=post_tags, back_populates="posts"
    )

Kullanımı:

with Session(engine) as session:
    # Etiketler oluştur
    python_tag = Tag(name="python")
    web_tag = Tag(name="web")
    db_tag = Tag(name="database")
    session.add_all([python_tag, web_tag, db_tag])
    session.commit()
    
    # Yazıya etiket ekle
    post = session.get(Post, 1)
    post.tags.append(python_tag)
    post.tags.append(db_tag)
    session.commit()
    
    # Yazının etiketleri
    for tag in post.tags:
        print(f"  #{tag.name}")
    
    # Etikete ait yazılar
    for p in python_tag.posts:
        print(f"  - {p.title}")

secondary=post_tags parametresi SQLAlchemy'ye ara tabloyu kullanmasını söyler. Ara tabloyu elle yönetmene gerek yok — etiket ekleme/çıkarma Python listeleri gibi çalışır.

8. Lazy Loading vs Eager Loading

İlişkili verilere erişirken SQLAlchemy varsayılan olarak lazy loading kullanır — yani ilişkili verileri sorgulanan an değil, erişildiğinde yükler.

N+1 Sorgu Problemi

# ❌ N+1 problem — 1 ana sorgu + N ilişki sorgusu
with Session(engine) as session:
    users = session.execute(select(User)).scalars().all()  # 1 sorgu
    for user in users:
        print(f"{user.name}: {len(user.posts)} yazı")  # Her biri 1 sorgu!
    # 100 kullanıcı = 101 SQL sorgusu!

Eager Loading ile Çözüm

from sqlalchemy.orm import joinedload, selectinload

# ✅ joinedload — LEFT JOIN ile tek sorguda
with Session(engine) as session:
    stmt = select(User).options(joinedload(User.posts))
    users = session.execute(stmt).unique().scalars().all()
    for user in users:
        print(f"{user.name}: {len(user.posts)} yazı")
    # Tek SQL sorgusu!

# ✅ selectinload — ikinci bir SELECT ile (büyük veri setlerinde daha iyi)
with Session(engine) as session:
    stmt = select(User).options(selectinload(User.posts))
    users = session.execute(stmt).scalars().all()

joinedload → Tek sorgu, JOIN ile. Az ilişkili veri varsa ideal. selectinload → İki sorgu: önce ana tablo, sonra ilişkililer. Çok ilişkili veri varsa daha iyi.

⚠️ Dikkat: joinedload kullanırken .unique() çağırmayı unutma — JOIN sonucu aynı ana kayıt tekrarlanabilir ve SQLAlchemy bunu tekilleştirmesi gerekir.

9. Cascade İşlemleri

Bir kullanıcı silindiğinde yazıları ne olacak? Cascade kuralları bunu belirler:

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan"  # Kullanıcı silinince yazıları da silinir
    )

Cascade seçenekleri:

"all, delete-orphan" en yaygın kullanılan kombinasyondur — ana kayıtla birlikte yaşayan bağımlı kayıtlar için idealdir.

10. Bütünleşik Örnek: Blog Sorguları

Önceki bölümlerdeki Author, Post, Tag modellerini kullanarak gerçek dünya sorgularını görelim:

# Modellerin tanımlı ve tabloların oluşturulmuş olduğunu varsayıyoruz
with Session(engine) as session:
    # 1. Yayınlanmış yazılar — eager loading ile
    stmt = (
        select(Post)
        .where(Post.published == True)
        .order_by(desc(Post.created_at))
        .options(selectinload(Post.author), selectinload(Post.tags))
    )
    for post in session.execute(stmt).scalars().all():
        tags = ", ".join(f"#{t.name}" for t in post.tags)
        print(f"  [{post.author.name}] {post.title} ({tags})")
    
    # 2. Yazarların yazı sayıları — GROUP BY + JOIN
    stmt = (
        select(Author.name, func.count(Post.id).label("post_count"))
        .join(Post, isouter=True)
        .group_by(Author.id)
        .order_by(desc("post_count"))
    )
    for name, count in session.execute(stmt).all():
        print(f"  {name}: {count} yazı")
    
    # 3. Belirli etiketli yazılar — Many-to-Many JOIN
    stmt = select(Post).join(Post.tags).where(Tag.name == "python")
    for post in session.execute(stmt).scalars().all():
        print(f"  - {post.title}")

Bu örnekte One-to-Many, Many-to-Many, eager loading, filtreleme, gruplama ve JOIN'lerin hepsini bir arada görüyorsun.

11. Migration: Alembic

Uygulaman geliştikçe modellerde değişiklik yaparsın — yeni sütun, yeni tablo, tip değişikliği. create_all() mevcut tabloları değiştirmez. İşte Alembic veritabanı şemasını versiyonlayarak güvenli değişiklik yapmanı sağlar.

Kurulum ve Başlangıç

pip install alembic
alembic init alembic

Bu komut bir alembic/ klasörü ve alembic.ini dosyası oluşturur.

Yapılandırma

alembic.ini dosyasında veritabanı URL'sini ayarla:

sqlalchemy.url = sqlite:///app.db

alembic/env.py dosyasında modellerini import et:

# alembic/env.py içinde
from models import Base  # Senin Base sınıfın
target_metadata = Base.metadata

Migration Oluşturma ve Uygulama

# Otomatik migration oluştur (model değişikliklerini algılar)
alembic revision --autogenerate -m "add bio column to users"

# Migration'ı uygula
alembic upgrade head

# Bir adım geri al
alembic downgrade -1

# Mevcut durumu gör
alembic current

# Migration geçmişini gör
alembic history

Oluşan migration dosyası şöyle görünür:

"""add bio column to users"""

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column("users", sa.Column("bio", sa.Text(), nullable=True))

def downgrade():
    op.drop_column("users", "bio")

upgrade() ileri taşır, downgrade() geri alır. Her migration bir versiyon numarasına sahiptir ve sırayla uygulanır — tıpkı Git commit'leri gibi.

💡 İpucu: --autogenerate her şeyi algılayamaz. Sütun yeniden adlandırma, veri taşıma gibi işlemleri elle yazman gerekebilir. Her zaman oluşan migration dosyasını kontrol et.

12. Connection Pooling

Her veritabanı bağlantısı açmak maliyetlidir (TCP handshake, authentication). Connection pooling bağlantıları yeniden kullanarak performansı artırır.

SQLAlchemy'de engine otomatik olarak bir connection pool yönetir:

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# Varsayılan: QueuePool (pool_size=5, max_overflow=10)
engine = create_engine(
    "postgresql://user:pass@localhost/mydb",
    pool_size=10,         # Havuzda tutulacak bağlantı sayısı
    max_overflow=20,      # Havuz dolunca ekstra açılabilecek bağlantı
    pool_timeout=30,      # Bağlantı beklerken timeout (saniye)
    pool_recycle=3600,    # Bağlantıyı yenileme süresi (saniye)
    pool_pre_ping=True,   # Bağlantıyı kullanmadan önce test et
)

pool_pre_ping=True çok önemlidir — özellikle veritabanı bağlantısı kopabilecek (timeout, restart) durumlarda. Kopmuş bir bağlantıyı algılayıp otomatik olarak yenisini açar.

SQLite dosya tabanlı olduğu için farklıdır — StaticPool veya NullPool tercih edilir.

13. Yaygın Hatalar

1. Session'ı Commit'lemeden Kapatmak

# ❌ Değişiklikler kaybolur
with Session(engine) as session:
    user = User(name="Test")
    session.add(user)
    # commit yok — session kapanınca rollback olur!

# ✅ Commit'i unutma
with Session(engine) as session:
    user = User(name="Test")
    session.add(user)
    session.commit()

2. Session Dışında Lazy Loading

# ❌ DetachedInstanceError
with Session(engine) as session:
    user = session.get(User, 1)

# Session kapandı!
print(user.posts)  # HATA! Lazy loading session gerektirir

# ✅ Session içinde erişim veya eager loading kullan
with Session(engine) as session:
    stmt = select(User).options(selectinload(User.posts)).where(User.id == 1)
    user = session.execute(stmt).scalar_one()
    posts = user.posts  # Session içinde — OK!

3. Engine'i Her İstekte Yeniden Oluşturmak

# ❌ Her fonksiyonda engine oluşturma — pool'u yok eder
def get_users():
    engine = create_engine("sqlite:///app.db")  # YANLIŞ!
    with Session(engine) as session:
        return session.execute(select(User)).scalars().all()

# ✅ Engine modül seviyesinde tek sefer oluştur
engine = create_engine("sqlite:///app.db")

def get_users():
    with Session(engine) as session:
        return session.execute(select(User)).scalars().all()

14. SQLAlchemy + FastAPI Entegrasyonu

SQLAlchemy'yi bir web framework ile nasıl kullanacağını görelim:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker

# Engine ve SessionLocal
engine = create_engine("sqlite:///app.db")
SessionLocal = sessionmaker(bind=engine)

app = FastAPI()

# Dependency: Her istekte yeni session, istek bitince kapat
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users")
def list_users(db: Session = Depends(get_db)):
    users = db.execute(select(User)).scalars().all()
    return [{"id": u.id, "name": u.name} for u in users]

@app.post("/users", status_code=201)
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    user = User(name=name, email=email)
    db.add(user)
    db.commit()
    db.refresh(user)
    return {"id": user.id, "name": user.name, "email": user.email}

Depends(get_db) her endpoint çağrısında yeni bir session oluşturur ve istek tamamlanınca otomatik kapatır. Bu pattern tüm FastAPI + SQLAlchemy projelerinde standarttır.

Özet

• ORM, Python nesneleriyle veritabanı tablolarını eşler — raw SQL yazmadan CRUD yapmanı sağlar.

• SQLAlchemy iki katmandan oluşur: Core (düşük seviye SQL builder) ve ORM (yüksek seviye nesne eşleme).

• Engine veritabanı bağlantısını, Session iş birimini (unit of work) yönetir. Engine'i bir kez oluştur, session'ı her işlem için aç-kapat.

• İlişkiler relationship() + ForeignKey ile tanımlanır. One-to-Many ve Many-to-Many en yaygın kalıplardır.

• Eager loading (joinedload, selectinload) N+1 sorgu problemini çözer — ilişkili verilere performanslı erişim sağlar.

• Alembic ile veritabanı şemasını versiyonla — create_all() production'da yetmez, migration kullan.