FastAPI katmanı
- Python kursu
- İleri seviye
- Ders 82/83
- 6 dk
FastAPI ile REST üç noktaları yazarsın.
FastAPI katmanı
Dördüncü katman, FastAPI ile bir HTTP yolu ekler. @app.get("/tasks") dekoratörü, read_tasks işlevini /tasks yoluyla eşleştirir. İşlev, asyncio.run(fetch_tasks()) ile veritabanını sorgular.
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/tasks")
def read_tasks():
return asyncio.run(fetch_tasks())TestClient, uygulamayı bir HTTP sunucusu gibi sorgular; client.get("/tasks").text yanıt gövdesini verir. Yanıt, Python sözlük listesinden otomatik olarak yapısal veri biçimine çevrilir.
TestClient ile sorgu
TestClient(app), FastAPI uygulamasını bellek içi sorgular. client.get("/tasks") bir yanıt nesnesi döndürür; .text gövdeyi, .status_code durum kodunu verir.
client = TestClient(app)
resp = client.get("/tasks")
print(resp.text)
print(resp.status_code) # 200Yanıt gövdesi, Python listesinin yapısal veri biçimine çevrilmiş halidir. Bu sayede sunucu başlatmadan tüm akışı test edebilirsin; gerçek bir HTTP isteğinin davranışını taklit eder.
Uygulama FastAPI yolunu ekle
Başlangıç kodu, önceki dersin tüm çözümünü içerir (async + SQLAlchemy). Şunları ekle:
from fastapi import FastAPIfrom fastapi.testclient import TestClientapp = FastAPI()@app.get("/tasks")iledef read_tasks():tanımla; gövdesireturn asyncio.run(fetch_tasks())yapsınclient = TestClient(app)print(client.get("/tasks").text)ile yanıt gövdesini 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())
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()]
# fastapi'den FastAPI'yi içe aktar
# fastapi.testclient'dan TestClient'ı içe aktar
# FastAPI sınıfını çağırarak uygulamayı tanımla
# app.get dekoratörüyle '/tasks' yoluna read_tasks'i bağla, asyncio.run(fetch_tasks()) döndürsün
# TestClient ile '/tasks' yolunu sorgula ve gövdeyi 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()]
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/tasks")
def read_tasks():
return asyncio.run(fetch_tasks())
client = TestClient(app)
print(client.get("/tasks").text)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 @app.get("/tasks") dekoratörü ne yapar?
- Decorator tanımlar
- Veritabanını oluşturur
- Asyncio döngüsünü başlatır
- read_tasks'i /tasks yoluyla eşleştirir Doğru cevap
Açıklama: @app.get('/tasks') read_tasks'i /tasks yoluyla eşleştirir.
2 TestClient çağrısını tamamla
from fastapi.testclient import TestClient
client = ____(app)from fastapi.testclient import TestClient
client = TestClient(app)Açıklama: TestClient(app) ile test istemcisi oluşur.
3 FastAPI katmanı kavramlarını eşleştir
| Kavram | Anlamı |
|---|---|
FastAPI() |
Uygulama oluşturur |
@app.get |
Yol tanımlar |
TestClient |
İstemci verir |
.text |
Gövde verir |