PYTHON FASTAPI: ASYNC API GELİŞTİRME

# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import logging
 
logger = logging.getLogger(__name__)
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Uygulama başlatılıyor...")
    yield
    logger.info("Uygulama kapatılıyor...")
 
app = FastAPI(
    title="Blog API",
    description="CodeForge blog platformu API'si",
    version="1.0.0",
    lifespan=lifespan,
)
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
@app.get("/health")
async def saglik_kontrol():
    return {"durum": "sağlıklı", "versiyon": "1.0.0"}

# schemas.py
from pydantic import BaseModel, EmailStr, field_validator, model_validator
from datetime import datetime
from typing import Optional
from uuid import UUID
import re
 
class KullaniciOlustur(BaseModel):
    ad:    str
    email: EmailStr
    sifre: str
 
    @field_validator('ad')
    @classmethod
    def ad_dogrula(cls, v: str) -> str:
        v = v.strip()
        if len(v) < 2:
            raise ValueError('Ad en az 2 karakter olmalıdır.')
        if len(v) > 100:
            raise ValueError('Ad en fazla 100 karakter olabilir.')
        return v
 
    @field_validator('sifre')
    @classmethod
    def sifre_guclu_mu(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError('Şifre en az 8 karakter olmalıdır.')
        if not re.search(r'[A-Z]', v):
            raise ValueError('Şifre en az bir büyük harf içermelidir.')
        if not re.search(r'\d', v):
            raise ValueError('Şifre en az bir rakam içermelidir.')
        return v
 
class KullaniciBilgi(BaseModel):
    id:          UUID
    ad:          str
    email:       EmailStr
    rol:         str
    olusturuldu: datetime
 
    model_config = {"from_attributes": True}  # ORM nesnelerini kabul et
 
class MakaleOlustur(BaseModel):
    baslik:    str
    icerik:    str
    etiketler: list[str] = []
    yayinla:   bool = False
 
    @field_validator('baslik')
    @classmethod
    def baslik_dogrula(cls, v: str) -> str:
        v = v.strip()
        if len(v) < 5 or len(v) > 300:
            raise ValueError('Başlık 5-300 karakter arasında olmalıdır.')
        return v
 
class MakaleOnizleme(BaseModel):
    id:          UUID
    baslik:      str
    slug:        str
    excerpt:     str
    olusturuldu: datetime
    yazar:       KullaniciBilgi
    etiketler:   list[str]
 
    model_config = {"from_attributes": True}
 
class SayfalamaCevap(BaseModel):
    veri:      list
    toplam:    int
    sayfa:     int
    son_sayfa: int

# routers/makaleler.py
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from uuid import UUID
from typing import Annotated
 
from database import get_db
from schemas import MakaleOlustur, MakaleOnizleme, SayfalamaCevap
from services.makale_service import MakaleService
from dependencies import aktif_kullanici_al
 
router = APIRouter(prefix="/makaleler", tags=["Makaleler"])
 
DbBaglanti  = Annotated[AsyncSession, Depends(get_db)]
AktifKullanici = Annotated[dict, Depends(aktif_kullanici_al)]
 
@router.get("/", response_model=SayfalamaCevap)
async def makaleleri_listele(
    db: DbBaglanti,
    sayfa:    int   = Query(default=1, ge=1),
    boyut:    int   = Query(default=20, ge=1, le=100),
    kategori: str | None = Query(default=None),
    arama:    str | None = Query(default=None),
):
    service = MakaleService(db)
    return await service.listele(sayfa=sayfa, boyut=boyut, kategori=kategori, arama=arama)
 
@router.get("/{slug}", response_model=MakaleOnizleme)
async def makale_getir(slug: str, db: DbBaglanti):
    service = MakaleService(db)
    makale = await service.slug_ile_getir(slug)
    if not makale:
        raise HTTPException(status_code=404, detail="Makale bulunamadı.")
    return makale
 
@router.post("/", response_model=MakaleOnizleme, status_code=status.HTTP_201_CREATED)
async def makale_olustur(
    govde:     MakaleOlustur,
    db:        DbBaglanti,
    kullanici: AktifKullanici,
):
    service = MakaleService(db)
    return await service.olustur(govde, yazar_id=kullanici["id"])
 
@router.delete("/{makale_id}", status_code=status.HTTP_204_NO_CONTENT)
async def makale_sil(
    makale_id: UUID,
    db:        DbBaglanti,
    kullanici: AktifKullanici,
):
    service = MakaleService(db)
    await service.sil(makale_id, yazar_id=kullanici["id"])

# dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from datetime import datetime, timedelta
import os
 
guvenlik = HTTPBearer()
 
def token_olustur(kullanici_id: str, rol: str) -> str:
    bitis = datetime.utcnow() + timedelta(minutes=15)
    payload = {
        "sub":     kullanici_id,
        "rol":     rol,
        "exp":     bitis,
        "iss":     "blog-api",
        "aud":     "blog-client",
    }
    return jwt.encode(payload, os.environ["JWT_SECRET"], algorithm="HS256")
 
async def aktif_kullanici_al(
    kimlik: HTTPAuthorizationCredentials = Depends(guvenlik),
) -> dict:
    try:
        payload = jwt.decode(
            kimlik.credentials,
            os.environ["JWT_SECRET"],
            algorithms=["HS256"],
            audience="blog-client",
        )
        return {"id": payload["sub"], "rol": payload["rol"]}
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Geçersiz veya süresi dolmuş token.",
            headers={"WWW-Authenticate": "Bearer"},
        )
 
def rol_gerektir(*roller: str):
    def kontrol(kullanici: dict = Depends(aktif_kullanici_al)) -> dict:
        if kullanici["rol"] not in roller:
            raise HTTPException(status_code=403, detail="Yetkiniz yok.")
        return kullanici
    return kontrol

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
import os
 
motor = create_async_engine(
    os.environ["DATABASE_URL"],
    echo=os.environ.get("NODE_ENV") == "development",
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
)
 
SessionLocal = async_sessionmaker(motor, expire_on_commit=False)
 
class Base(DeclarativeBase):
    pass
 
async def get_db():
    async with SessionLocal() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()
 
# models.py
from sqlalchemy import String, Boolean, Integer, ForeignKey, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from datetime import datetime
import uuid
 
class Kullanici(Base):
    __tablename__ = "kullanicilar"
 
    id:          Mapped[uuid.UUID]  = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    ad:          Mapped[str]        = mapped_column(String(100), nullable=False)
    email:       Mapped[str]        = mapped_column(String(255), unique=True, nullable=False)
    sifre_hash:  Mapped[str]        = mapped_column(Text, nullable=False)
    rol:         Mapped[str]        = mapped_column(String(20), nullable=False, default="okuyucu")
    olusturuldu: Mapped[datetime]   = mapped_column(default=datetime.utcnow)
 
    makaleler: Mapped[list["Makale"]] = relationship(back_populates="yazar")

from fastapi import BackgroundTasks
import smtplib
 
def hosgeldin_emaili_gonder(email: str, ad: str):
    # Bu fonksiyon istemci yanıt aldıktan sonra çalışır
    with smtplib.SMTP(os.environ["SMTP_HOST"]) as server:
        server.sendmail(
            "noreply@blog.com",
            email,
            f"Subject: Hoş geldiniz!\n\nMerhaba {ad}, kaydınız tamamlandı!"
        )
 
@router.post("/kayit", response_model=KullaniciBilgi)
async def kayit_ol(
    govde:       KullaniciOlustur,
    db:          DbBaglanti,
    arka_gorev:  BackgroundTasks,
):
    kullanici = await kullanici_service.olustur(db, govde)
    arka_gorev.add_task(hosgeldin_emaili_gonder, kullanici.email, kullanici.ad)
    return kullanici
    # İstemci anında yanıt alır, e-posta arka planda gönderilir