Birleştirme
- Python kursu
- İleri seviye
- Ders 83/83
- 6 dk
Tüm katmanları birleştirerek çalışan bir mini framework yazarsın.
Birleştirme: decorator
Son katman, bir decorator ile çağrı günlüğü ekler. log_call, async işlevleri saran bir decorator'dur; functools.wraps ile özgün işlevin kimliğini korur.
import functools
def log_call(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
print("çağrıldı: " + func.__name__)
return await func(*args, **kwargs)
return wrapperBu decorator'ü fetch_tasks'e uygularsın: @log_call. Her çağrıda önce çağrıldı: fetch_tasks yazılır, sonra işlevin sonucu döner. Böylece API çağrıları izlenebilir.
Tüm katmanlar birleşti
Artık tüm katmanlar yerli yerinde:
- fetch_tasks: async iskelet + decorator.
- fetch_all: asyncio.gather.
- SQLAlchemy: Task modeli ve sorgu.
- FastAPI: /tasks yolu ve TestClient.
@log_call
async def fetch_tasks():
return [{...} for t in session.query(Task).all()]Bu desen, gerçek dünyadaki API'lerin omurgasıdır: async G/Ç, kalıcı veri, HTTP yolu ve gözlemlenebilirlik. Her katman ayrı ayrı test edilebilir; decorator eklendiğinde mevcut davranış bozulmaz.
Uygulama log_call decorator'ünü ekle
Başlangıç kodu, önceki dersin tüm çözümünü içerir (async + SQLAlchemy + FastAPI). Şunları ekle:
import functoolsdef log_call(func):decorator tanımla;@functools.wraps(func)ileasync def wrapper(*args, **kwargs):gövdesinde önceprint("çağrıldı: " + func.__name__)sonrareturn await func(*args, **kwargs)yapsın;return wrapper@log_callileasync def fetch_tasks():işlevini yeniden tanımla (DB sorgusu yapan versiyon)
Son satır zaten print(client.get("/tasks").text) olmalı.
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()
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
# functools'ü içe aktar
# log_call adlı bir decorator tanımla (functools.wraps ile saran async wrapper)
# wrapper önce işlev adını 'çağrıldı:' önekiyle yazdırsın, sonra func'ı await'le
# @log_call ile fetch_tasks'i DB sorgusu yapacak şekilde yeniden 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()
import functools
def log_call(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
print("çağrıldı: " + func.__name__)
return await func(*args, **kwargs)
return wrapper
@log_call
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ı
çağrıldı: fetch_tasks
[{"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 log_call decorator'ü neden functools.wraps kullanır?
- Async yapmaya zorlamak için
- Özgün işlevin kimliğini korumak için Doğru cevap
- Hızlandırmak için
- Yorum eklemek için
Açıklama: wraps, __name__ ve __doc__ gibi öznitelikleri kopyalar.
2 wraps çağrısını tamamla
Soru
import functools
def log_call(func):
@functools.____(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapper
Cevap
import functools
def log_call(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapperAçıklama: functools.wraps özgün işlevin kimliğini kopyalar.
3 Birleştirme kavramlarını eşleştir
| Kavram | Anlamı |
|---|---|
log_call |
Çağrı günlüğü yazar |
functools.wraps |
Kimliği korur |
@log_call |
Decorator uygulaması |
wrapper |
Saran işlev |