SQLAlchemy katmanı
- Python kursu
- İleri seviye
- Ders 81/83
- 6 dk
SQLAlchemy ile veritabanı işlemleri yazarsın.
SQLAlchemy katmanı
Üçüncü katman, sabit listeyi gerçek bir veritabanı sorgusuyla değiştirir. SQLAlchemy ile bellek içi SQLite kurar, bir Task modeli tanımlar ve session.query(Task).all() ile sorgularız.
from sqlalchemy import create_engine, Column, Integer, String, StaticPool
from sqlalchemy.orm import declarative_base, sessionmaker
engine = create_engine("sqlite:///:memory:", poolclass=StaticPool)Bellek içi SQLite, StaticPool olmadan her bağlantıda yeni bir veritabanı oluşturur. FastAPI TestClient ayrı bir iş parçacığı kullanacağından, check_same_thread=False ve StaticPool bağlantıyı paylaşmak için şarttır.
Task modeli
SQLAlchemy modeli, declarative_base()'ten türetilmiş bir sınıftır. __tablename__ tablo adını, Column ise sütunları tanımlar.
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True)
title = Column(String)Bu sınıf, Python nesneleriyle veritabanı satırları arasında bir köprüdür. session.add(Task(...)) ile yeni satır ekler; session.commit() ile kalıcı yaparsın. Sorgu, sınıf adından tabloya otomatik geçer.
Uygulama Task modeli ve veritabanı sorgusu ekle
Başlangıç kodu, önceki dersin fetch_tasks ve fetch_all çözümlerini içerir. Şunları ekle:
from sqlalchemy import create_engine, Column, Integer, String, StaticPoolfrom sqlalchemy.orm import declarative_base, sessionmakerBase = declarative_base()class Task(Base):—__tablename__ = "tasks",id = Column(Integer, primary_key=True),title = Column(String)- Motoru oluştur:
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine),Session = sessionmaker(bind=engine),session = Session()- İki Task ekle ve commit yap
fetch_tasks'i yeniden tanımla:return [{"id": t.id, "title": t.title} for t in session.query(Task).all()]
Sonra print(asyncio.run(fetch_tasks())) ile sonucu yazdır.
Başlangıç kodu
import asyncio
async def fetch_tasks():
return [{"id": 1, "title": "Görev 1"}, {"id": 2, "title": "Görev 2"}]
async def fetch_all():
return await asyncio.gather(fetch_tasks(), fetch_tasks())
# sqlalchemy'den create_engine, Column, Integer, String, StaticPool'ü içe aktar
# sqlalchemy.orm'dan declarative_base ve sessionmaker'ı içe aktar
# Base'i tanımla, Task modelini yaz (id, title sütunları)
# motoru oluştur (bellek içi sqlite, StaticPool, check_same_thread=False)
# tabloları yarat, oturum aç, iki Task ekle ve commit yap
# fetch_tasks'i veritabanından sorgu yapacak şekilde yeniden tanımla
# asyncio.run ile fetch_tasks'i çalıştır ve sonucu yazdırÖrnek çözüm
import asyncio
async def fetch_tasks():
return [{"id": 1, "title": "Görev 1"}, {"id": 2, "title": "Görev 2"}]
async def fetch_all():
return await asyncio.gather(fetch_tasks(), fetch_tasks())
from sqlalchemy import create_engine, Column, Integer, String, StaticPool
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True)
title = Column(String)
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(Task(id=1, title="Görev 1"))
session.add(Task(id=2, title="Görev 2"))
session.commit()
async def fetch_tasks():
return [{"id": t.id, "title": t.title} for t in session.query(Task).all()]
print(asyncio.run(fetch_tasks()))Beklenen çıktı
[{'id': 1, 'title': 'Görev 1'}, {'id': 2, 'title': 'Görev 2'}]
Soru ve cevaplar
Dersin sorularını cevaplarıyla birlikte inceleyebilirsin. Kendini denemek istersen dersi interaktif olarak çöz.
1 FastAPI TestClient ile bellek içi SQLite kullanırken ne gerekir?
- Ek ayar gerekmez
- StaticPool ve check_same_thread=False Doğru cevap
- Yalnızca StaticPool
- Yalnızca check_same_thread=False
Açıklama: StaticPool bağlantıyı paylaşır; check_same_thread=False iş parçacığına izin verir.
2 Tablo adını tamamla
class Task(Base):
____ = "tasks"class Task(Base):
__tablename__ = "tasks"Açıklama: __tablename__ tablo adını belirler.
3 SQLAlchemy kavramlarını eşleştir
| Kavram | Anlamı |
|---|---|
create_engine |
Motor oluşturur |
declarative_base |
Model temeli |
Column |
Sütun tanımlar |
session.query |
Sorgu yapar |