Birleştirme

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 wrapper

Bu 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 functools
def log_call(func): decorator tanımla; @functools.wraps(func) ile async def wrapper(*args, **kwargs): gövdesinde önce print("çağrıldı: " + func.__name__) sonra return await func(*args, **kwargs) yapsın; return wrapper
@log_call ile async def fetch_tasks(): işlevini yeniden tanımla (DB sorgusu yapan versiyon)

Son satır zaten print(client.get("/tasks").text) olmalı.

Ö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()

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)

Kendini test et

log_call decorator'ü neden functools.wraps kullanır?
- Async yapmaya zorlamak için
- Özgün işlevin kimliğini korumak için
- Hızlandırmak için
- Yorum eklemek için
Cevabı göster

Özgün işlevin kimliğini korumak için — wraps, __name__ ve __doc__ gibi öznitelikleri kopyalar.

wraps çağrısını tamamla

import functools

def log_call(func):
    @functools.____(func)
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)
    return wrapper

Cevabı göster

import functools

def log_call(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)
    return wrapper

Birleştirme kavramlarını eşleştir

log_call

Çağrı günlüğü yazar
functools.wraps

Kimliği korur
@log_call

Decorator uygulaması
wrapper

Saran işlev

Sıradaki ders: İleri seviye bölüm sınavı

Deneyiminizi Geliştiriyoruz

Çerez Tercihleri

Kesinlikle Gerekli Çerezler

Analitik Çerezler