Python
SQLAlchemy katmanı
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.
Örnek çözümü göster
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()))
Kendini test et
-
FastAPI TestClient ile bellek içi SQLite kullanırken ne gerekir?
- Ek ayar gerekmez
- StaticPool ve check_same_thread=False
- Yalnızca StaticPool
- Yalnızca check_same_thread=False
Cevabı göster
StaticPool ve check_same_thread=False — StaticPool bağlantıyı paylaşır; check_same_thread=False iş parçacığına izin verir.
-
Tablo adını tamamla
class Task(Base):
____ = "tasks"
Cevabı göster
class Task(Base):
__tablename__ = "tasks"
-
SQLAlchemy kavramlarını eşleştir
- create_engine
- Motor oluşturur
- declarative_base
- Model temeli
- Column
- Sütun tanımlar
- session.query
- Sorgu yapar
Sıradaki ders: FastAPI katmanı
