Python
FastAPI katmanı
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) # 200
Yanı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") ile def read_tasks(): tanımla; gövdesi return asyncio.run(fetch_tasks()) yapsınclient = TestClient(app)print(client.get("/tasks").text) ile yanıt gövdesini 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()]
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)
Kendini test et
-
@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
Cevabı göster
read_tasks'i /tasks yoluyla eşleştirir — @app.get('/tasks') read_tasks'i /tasks yoluyla eşleştirir.
-
TestClient çağrısını tamamla
from fastapi.testclient import TestClient
client = ____(app)
Cevabı göster
from fastapi.testclient import TestClient
client = TestClient(app)
-
FastAPI katmanı kavramlarını eşleştir
- FastAPI()
- Uygulama oluşturur
- @app.get
- Yol tanımlar
- TestClient
- İstemci verir
- .text
- Gövde verir
Sıradaki ders: Birleştirme
