Asenkron Uç Noktalar ve Veritabanı Entegrasyonu

async def Uç Noktaları

from fastapi import FastAPI
import asyncio
app = FastAPI()

@app.get("/slow")
async def slow():
    await asyncio.sleep(1)   # non-blocking
    return {"done": True}

PostgreSQL için asyncpg

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool("postgresql://user:pass@host/db")
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name FROM users")
    await pool.close()
    return rows

Asenkron SQLAlchemy

# pip install sqlalchemy aiosqlite
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///db.sqlite")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

Veritabanı Bağımlılığı

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

SQLAlchemy ORM Modelleri

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"

    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    email: Mapped[str]

CRUD İşlemleri

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def create_user(db: AsyncSession, name: str, email: str):
    user = User(name=name, email=email)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

Geçişler için Alembic

# pip install alembic
# alembic init alembic
# Edit alembic/env.py to point to your models

# Generate a migration:
# alembic revision --autogenerate -m "add users table"

# Apply:
# alembic upgrade head

Asenkron HTTP Çağrıları için httpx

# pip install httpx
import httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/weather")
async def weather(city: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://wttr.in/{city}?format=j1")
        return r.json()

Arka Plan Görevleri

from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

def send_welcome_email(email: str):
    # blocking I/O — runs in background thread
    smtp_send(email, "Welcome!")

@app.post("/register")
def register(email: str, bg: BackgroundTasks):
    create_user(email)
    bg.add_task(send_welcome_email, email)
    return {"status": "registered"}

Yaşam Döngüsü Olayları

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(DSN)
    yield
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)

FastAPI'yi httpx ile Test Etme

import pytest, httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/ping")
async def ping(): return {"pong": True}

@pytest.mark.asyncio
async def test_ping():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        r = await client.get("/ping")
    assert r.json() == {"pong": True}

Hızlı Kontrol

Özet