Görseldeki ders müfredatını ve FastAPI eğitim içeriğini adım adım analiz ederek, sizden gelen tüm kurallar doğrultusunda (süreleri, "Önizleme" ve "ders" satırlarını, anlamsız ifadeleri temizleyerek) detaylı bir konu anlatım rehberi ve uygulama kılavuzu haline getirdim.
Yarıda kalmamak ve her konuyu derinlemesine, bol kod örneğiyle açıklayabilmek adına bu uzun müfredatı parçalara ayırarak ilerliyoruz. İlk olarak "Öğrenecekleriniz" bölümünün kapsamlı analizi ve ardından ilk ana konuların detaylı anlatımıyla başlıyoruz.
Bu eğitim içeriği, modern web dünyasının en popüler Python framework'lerinden biri olan FastAPI ile sıfırdan ileri seviyeye güvenli bir REST API geliştirmeyi hedefler. Bu bölümde vaat edilen yetkinliklerin teknik karşılıkları şunlardır:
FastAPI ile Hızlı API Geliştirme: FastAPI, Python'ın tip ipuçlarını (Type Hints) kullanarak çok az kodla asenkron, yüksek performanslı ve standartlara uygun (OpenAPI, JSON Schema) API'lar üretmenizi sağlar.
Veri Modelleme ve API Yanıtları: Pydantic kütüphanesi kullanılarak API'ye gelen ve API'den giden verilerin doğrulanması (validation) ve şemalandırılması öğrenilir.
Veritabanı Bağlantısı ve CRUD İşlemleri: İlişkisel bir veritabanına bağlanarak SQLAlchemy ORM (Object-Relational Mapping) aracılığıyla ham SQL yazmadan Veri Ekleme (Create), Okuma (Read), Güncelleme (Update) ve Silme (Delete) süreçleri yönetilir.
HTTP Durum Kodları ve Hata Yönetimi: İstemciye (frontend/mobil) sürecin başarılı olup olmadığını doğru HTTP kodlarıyla (Örn: 201 Created, 404 Not Found) bildirme ve hatalı isteklerde programın çökmesini engelleyerek anlamlı hata mesajları (Exception Handling) dönme pratikleri kazanılır.
İlişkisel
Veri Modelleri ve APIRouter: Gerçek hayattaki gibi (Örn: Bir
satıcının birden fazla ürünü olması) tablolar arası
ilişkiler kurulur. Proje büyüdükçe kodun kirlenmesini önlemek
için yollar APIRouter
ile modüllere ayrılır.
OAuth2 ve JWT ile Güvenlik: API rotalarını yetkisiz erişime kapatmak için şifrelerin hash'lenmesi, kullanıcı giriş süreçleri ve her istekte kullanıcının kimliğini doğrulamak için JWT (JSON Web Token) üretilip çözülmesi süreçleri uçtan uca uygulanır.
API (Application Programming Interface), iki yazılımın birbiriyle konuşmasını sağlayan bir köprüdür. REST API ise web üzerinde HTTP protokolünün metotlarını (GET, POST, PUT, DELETE) kullanarak veri alışverişi yapan standart bir mimaridir.
FastAPI; Python 3.8+ ile standart Python tip ipuçlarına dayanan, modern ve çok yüksek performanslı bir web framework'üdür. En büyük avantajları hızı (NodeJS ve Go ile yarışır seviyede), otomatik dökümantasyon üretmesi ve hataları otomatik yakalamasıdır.
İlk olarak sisteminizde FastAPI ve sunucu olarak çalışabilmesi için Uvicorn'un kurulu olması gerekir.
pip install fastapi uvicorn
Şimdi en basit haliyle bir API
oluşturalım. Bir main.py
dosyası oluşturun:
fromfastapiimportFastAPI# FastAPI nesnesini başlatıyoruzapp= FastAPI(title="Benim İlk Başarılı API'm",description="FastAPI müfredat eğitimi kapsamında oluşturulmuş ilk projedir.",version="1.0.0")# Kök (Root) dizine bir GET isteği tanımlıyoruz@app.get("/")defana_sayfa():return{"mesaj":"FastAPI dünyasına hoş geldiniz!"}
Terminalden şu komutla API'nizi ayağa kaldırın:
uvicorn main:app --reload
--reload
parametresi, kodda bir değişiklik yaptığınızda sunucunun
otomatik olarak yeniden başlamasını sağlar.
API'lara dışarıdan veri aktarmanın en temel iki yolu Path (Yol) ve Query (Sorgu) parametreleridir.
URL yapısının ayrılmaz bir parçası olan dinamik değişkenlerdir. Genellikle doğrudan belirli bir kaynağı (ID ile) işaret etmek için kullanılır.
URL'in sonuna ?
işaretinden sonra eklenen anahtar-değer çiftleridir. Genellikle
filtreleme, sıralama veya sayfalama (pagination) işlemlerinde
tercih edilir.
Aşağıdaki örnekte hem veri tipini zorunlu kılmayı, hem otomatik dökümantasyonu incelemeyi, hem de isteğe bağlı (optional) ve zorunlu (required) parametre mantığını göreceğiz.
fromfastapiimportFastAPIfromtypingimportOptionalapp= FastAPI()# Sahte bir ürün veritabanı simüle edelimURUNLER= {1: {"isim":"Kablosuz Kulaklık","fiyat":1200,"kategori":"Elektronik"},2: {"isim":"Akıllı Saat","fiyat":3500,"kategori":"Elektronik"},3: {"isim":"Koşu Ayakkabısı","fiyat":1800,"kategori":"Spor"}}# 1. Path Parameter ve Tip Kontrolü# '{urun_id}' bir yol parametresidir ve ': int' diyerek tam sayı olması zorunlu tutulmuştur.@app.get("/urunler/{urun_id}")defurun_detay_getir(urun_id: int):ifurun_idinURUNLER:returnURUNLER[urun_id]return{"hata":"Ürün bulunamadı"}# 2. Path & Query Parametrelerinin Birlikte Kullanımı# 'kategori' sorgu parametresi opsiyoneldir, 'limit' ise varsayılan değer olarak 10 almıştır.@app.get("/urun-ara/")defurun_filtrele(kategori: Optional[str] =None,limit: int =10):sonuclar= list(URUNLER.values())ifkategori:sonuclar= [uforuinsonuclarifu["kategori"].lower() == kategori.lower()]returnsonuclar[:limit]
FastAPI'nin en büyüleyici
özelliklerinden biri olan otomatik dökümantasyonu görmek için
tarayıcınızdan şu adrese gidin:
👉 http://127.0.0
Bu arayüz sayesinde yazdığınız tüm fonksiyonlar, beklenen parametre tipleri interaktif bir dökümantasyona dönüşür. Ekstra hiçbir şey yazmadan API'nizi buradan "Try it out" butonuna basarak test edebilirsiniz.
Bir API'ye yeni bir veri kaydetmek istediğimizde (Örn: Yeni kullanıcı kaydı, yeni ürün ekleme), bu veriyi URL üzerinden göndermek güvenli ve pratik değildir. Veriyi HTTP Request Body (İstek Gövdesi) içinde JSON formatında göndeririz.
FastAPI, bu gelen karmaşık JSON verilerini doğrulamak ve Python nesnelerine dönüştürmek için Pydantic kütüphanesini kullanır.
Bu uygulamada iç içe geçmiş (nested) veri tiplerini, metadata tanımlamayı ve örnek veri (example data) sağlamayı göreceğiz.
fromfastapiimportFastAPIfrompydanticimportBaseModel,FieldfromtypingimportList,Optionalfromdatetimeimportdatetimeapp= FastAPI()# 1. Alt Model Tanımlama (Nesting Pydantic Models)classOzellik(BaseModel):renk: strboyut: str# 2. Ana Model Tanımlama (Request Body için)classUrunEklemeModeli(BaseModel):# Field kullanarak kısıtlamalar ve açıklamalar (metadata) ekliyoruzisim: str = Field(..., title="Ürün Adı", min_length=2, max_length=50)fiyat: float = Field(..., gt=0, description="Ürün fiyatı sıfırdan büyük olmalıdır.")etiketler: List[str] = []# Python veri tipi iç içe kullanımıdetaylar: Optional[Ozellik] =None# Başka bir Pydantic modelini iç içe eklemeeklenme_tarihi: datetime = Field(default_factory=datetime.now)# Zaman veri tipi kullanımı# Swagger dökümantasyonunda gözükecek örnek şablon verisiclassConfig:json_schema_extra= {"example": {"isim":"Oyuncu Faresi","fiyat":750.50,"etiketler": ["elektronik","gaming","rgb"],"detaylar": {"renk":"Siyah","boyut":"Standart"}}}# POST metodu ile yeni veri kabul eden API rotası@app.post("/urunler/", status_code=201)defyeni_urun_ekle(urun: UrunEklemeModeli):# Gelen veri Pydantic sayesinde otomatik olarak doğrulanmıştır.# urun.isim veya urun.fiyat şeklinde doğrudan erişebiliriz.urun_sozluk= urun.model_dump()# Gerçek senaryoda burası veritabanına kaydedilir.return{"mesaj":"Ürün başarıyla doğrulandı ve (sanal olarak) eklendi.","eklenen_urun": urun_sozluk}
Eğer bu endpoint'e fiyat:
-5 veya
isim:
"A"
(minimum karakter sınırının altında) şeklinde hatalı bir veri
göndermeye çalışırsanız, FastAPI sizin yerinize isteği
reddeder ve istemciye hatanın tam olarak nerede olduğunu söyleyen
temiz bir 422 Unprocessable Entity
yanıtı döner.
Harika! Tam hızla, müfredatın en kritik ve omurgasını oluşturan veritabanı, CRUD işlemleri ve ilişkisel modelleme bölümlerine geçiyoruz. Temizlik kurallarımızı (süreler, önizleme ve ders ibarelerini kaldırma) koruyarak tamamen üretime yönelik, çalışan kodlar üzerinden ilerleyelim.
Gerçek dünya uygulamalarında veriler bellekte (RAM) tutulmaz; kalıcı olmaları için veritabanına yazılır. Bu bölümde Python dünyasının en güçlü ORM (Object-Relational Mapping) kütüphanelerinden biri olan SQLAlchemy kullanarak veritabanı mimarisini kuracağız. ORM sayesinde SQL kodu yazmadan, Python sınıfları ile tablolar oluşturup veri yöneteceğiz.
Projemizde dosya tabanlı, kurulum gerektirmeyen hafif bir veritabanı olan SQLite kullanacağız.
Büyük projeleri modüler hale getirmek eğitimin ana amaçlarından biridir. Bu yüzden kodumuzu parçalara ayırarak tasarlıyoruz.
database.py
- Bağlantı Ayarları
Bu dosyada veritabanı motorunu
(engine)
oluşturacak ve her istekte açılıp kapanacak veritabanı
oturumlarını (SessionLocal)
tanımlayacağız.
fromsqlalchemyimportcreate_enginefromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmaker# SQLite veritabanı dosyasının yoluSQLALCHEMY_DATABASE_URL="sqlite:///./fastapi_kursu.db"# Engine (Motor) oluşturuluyor.# connect_args={"check_same_thread": False} sadece SQLite için çoklu thread güvenliği sağlamak adına gereklidir.engine= create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread":False})# Veritabanı ile konuşacak oturum fabrikasıSessionLocal= sessionmaker(autocommit=False, autoflush=False, bind=engine)# Modellerimizin miras alacağı ana sınıfBase= declarative_base()# Bağımlılık Enjeksiyonu (Dependency Injection) için fonksiyon# Bu fonksiyon her API isteğinde bir veritabanı oturumu açar ve iş bitince kapatır.defget_db():db= SessionLocal()try:yielddbfinally:db.close()
models.py
- Veritabanı Tablosunun TanımlanmasıSQLAlchemy kullanarak veritabanındaki tablolarımızın şemasını Python sınıfı olarak tasarlıyoruz.
fromsqlalchemyimportColumn,Integer,String,Float,BooleanfromdatabaseimportBaseclassProductModel(Base):__tablename__="products"# Veritabanındaki tablo adıid= Column(Integer, primary_key=True, index=True)name= Column(String, nullable=False)price= Column(Float, nullable=False)is_active= Column(Boolean, default=True)
CRUD (Create, Read, Update,
Delete), veritabanı işlemlerinin temelidir. Bu bölümde hem
veritabanı işlemlerini yapacak endpoint'leri yazacağız hem de
Response Model (Yanıt Modeli)
ile dışarıya döneceğimiz veriyi filtreleyeceğiz. Ayrıca hata
yönetimini (HTTPException)
entegre edeceğiz.
Şimdi schemas.py
dosyasında Pydantic modellerimizi (istek ve yanıt şemaları)
tanımlayalım, ardından main.py
içinde tüm operasyonları birleştirelim.
schemas.py
- Pydantic İstek ve Yanıt ŞemalarıfrompydanticimportBaseModel# Kullanıcıdan ürün eklerken/güncellerken beklediğimiz şemaclassProductBase(BaseModel):name: strprice: floatis_active: bool =TrueclassProductCreate(ProductBase):pass# Ekstra alan gerekirse buraya eklenir# API'den istemciye döneceğimiz şema (Response Model)# Veritabanındaki 'id' alanını da içermelidir.classProductResponse(ProductBase):id: intclassConfig:from_attributes=True# SQLAlchemy nesnesini Pydantic şemasına dönüştürmeyi sağlar (Eski adıyla orm_mode)
main.py
- CRUD Endpoint'lerifromfastapiimportFastAPI,Depends,HTTPException,statusfromsqlalchemy.ormimportSessionimportmodelsimportschemasfromdatabaseimportengine,get_db# Veritabanı tablolarını otomatik oluşturuyoruz (Uygulama başlarken)models.Base.metadata.create_all(bind=engine)app= FastAPI(title="Ürün Yönetim API'si")# 1. CREATE (Veri Ekleme)# response_model kullanarak başarılı yanıtta sadece 'ProductResponse' alanlarının döneceğini garanti ediyoruz.@app.post("/products/", response_model=schemas.ProductResponse, status_code=status.HTTP_201_CREATED)defcreate_product(product: schemas.ProductCreate,db: Session = Depends(get_db)):new_product= models.ProductModel(name=product.name,price=product.price,is_active=product.is_active)db.add(new_product)db.commit()db.refresh(new_product)# Oluşan ID değerini almak için yeniliyoruzreturnnew_product# 2. READ ALL (Tüm Ürünleri Listeleme)@app.get("/products/", response_model=list[schemas.ProductResponse])defget_products(db: Session = Depends(get_db)):products= db.query(models.ProductModel).all()returnproducts# 3. READ ONE (Tek Ürün Getirme & İstisna Yönetme)@app.get("/products/{product_id}", response_model=schemas.ProductResponse)defget_product(product_id: int,db: Session = Depends(get_db)):product= db.query(models.ProductModel).filter(models.ProductModel.id == product_id).first()ifnotproduct:# Eğer ürün bulunamazsa HTTP 404 hatası fırlatıyoruzraiseHTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Aradığınız ürün bulunamadı!")returnproduct# 4. UPDATE (Ürün Güncelleme)@app.put("/products/{product_id}", response_model=schemas.ProductResponse)defupdate_product(product_id: int,updated_data: schemas.ProductCreate,db: Session = Depends(get_db)):product_query= db.query(models.ProductModel).filter(models.ProductModel.id == product_id)product= product_query.first()ifnotproduct:raiseHTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Güncellenmek istenen ürün bulunamadı!")product_query.update(updated_data.model_dump(), synchronize_session=False)db.commit()returnproduct_query.first()# 5. DELETE (Ürün Silme)@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT)defdelete_product(product_id: int,db: Session = Depends(get_db)):product_query= db.query(models.ProductModel).filter(models.ProductModel.id == product_id)product= product_query.first()ifnotproduct:raiseHTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Silinmek istenen ürün bulunamadı!")product_query.delete(synchronize_session=False)db.commit()# 204 No Content dönüldüğü için gövdede bir veri döndürmüyoruz.returnNone
Gerçek senaryolarda tablolar birbirine bağlıdır. Örneğin, projemize bir de Seller (Satıcı) modeli ekleyelim. Bir satıcının birden fazla ürünü olabilir (Bire-Çok İlişki / One-to-Many).
Ayrıca projedeki kod karmaşasını
çözmek için rotalarımızı modüllere ayırarak APIRouter
mimarisini devreye alacağız.
models.py
Güncellemesi (İlişki Kurma)fromsqlalchemyimportColumn,Integer,String,Float,Boolean,ForeignKeyfromsqlalchemy.ormimportrelationshipfromdatabaseimportBaseclassSellerModel(Base):__tablename__="sellers"id= Column(Integer, primary_key=True, index=True)username= Column(String, unique=True, index=True, nullable=False)hashed_password= Column(String, nullable=False)# Şifreleri hash'lenmiş tutacağız# İlişki tanımı: Bir satıcının birden fazla ürünü olabilir (back_populates)products= relationship("ProductModel", back_populates="owner")classProductModel(Base):__tablename__="products"id= Column(Integer, primary_key=True, index=True)name= Column(String, nullable=False)price= Column(Float, nullable=False)is_active= Column(Boolean, default=True)# Yabancı Anahtar (Foreign Key) ile satıcıya bağlanmaseller_id= Column(Integer, ForeignKey("sellers.id"))# İlişki tanımıowner= relationship("SellerModel", back_populates="products")
routers/seller.py
Proje kökünde routers
adında bir klasör açıp içine seller.py
yazarak satıcı işlemlerini APIRouter
ile ana dosyadan yalıtıyoruz. Rotalarımıza prefix (önek) ve tag
(etiket) ekliyoruz.
fromfastapiimportAPIRouter,Depends,HTTPException,statusfromsqlalchemy.ormimportSessionfromdatabaseimportget_dbimportmodels,schemas# APIRouter nesnesi tanımlanıyor. Tags ve Prefix dökümantasyonu düzenler.router= APIRouter(prefix="/sellers",tags=["Satıcı İşlemleri"])# Şifre hash'leme simülasyonu (Gerçek hash'leme Authentication bölümünde yapılacak)deffake_hash_password(password: str):returnpassword +"notsecurehash"@router.post("/", response_model=schemas.SellerResponse, status_code=201)defcreate_seller(seller: schemas.SellerCreate,db: Session = Depends(get_db)):# Kullanıcı adı kontrolüdb_seller= db.query(models.SellerModel).filter(models.SellerModel.username == seller.username).first()ifdb_seller:raiseHTTPException(status_code=400, detail="Bu kullanıcı adı zaten alınmış!")hashed_pwd= fake_hash_password(seller.password)new_seller= models.SellerModel(username=seller.username, hashed_password=hashed_pwd)db.add(new_seller)db.commit()db.refresh(new_seller)returnnew_seller
(Not: schemas.SellerResponse
ve schemas.SellerCreate
şemalarının schemas.py
içinde tanımlandığı varsayılmıştır).
main.py)
Ana main.py
dosyamızı tertemiz hale getirip yazdığımız router'ı içeri
dahil ediyoruz:
fromfastapiimportFastAPIfromdatabaseimportengineimportmodelsfromroutersimportseller# Yazdığımız router modülümodels.Base.metadata.create_all(bind=engine)app= FastAPI(title="Gelişmiş İlişkisel API")# Router'ı uygulamaya dahil ediyoruz (Include Router)app.include_router(seller.router)@app.get("/")defroot():return{"mesaj":"Modüler API Sistemine Hoş Geldiniz"}
Harika! Müfredatın en kritik zirve noktası olan Authentication (Kimlik Doğrulama) sistemini tasarlıyoruz. Bu bölümde şifreleri güvenli bir şekilde hash'lemeyi, kullanıcı giriş yaptığında JWT (JSON Web Token) üretmeyi, bu token'ı doğrulamayı ve belirli API rotalarını sadece yetkili kullanıcıların erişimine açmayı (Rotaları Korumayı) uçtan uca uygulayacağız.
Süreleri, dökümantasyon veya gereksiz ibareleri temizleyerek tamamen üretime hazır, modern ve güvenli bir mimari üzerinden ilerleyelim.
JWT, taraflar arasında güvenli bir şekilde JSON nesnesi olarak veri taşınmasını sağlayan açık bir standarttır (RFC 7519).
Kullanıcı kullanıcı adı ve şifresiyle giriş isteği (POST) yapar.
Sunucu şifreyi
doğrularsa, kullanıcının kimlik bilgisini (örn: kullanıcı
adı, id) içeren, sunucudaki gizli bir anahtar (SECRET_KEY)
ile imzalanmış şifreli bir Token (Token metni) üretir ve
istemciye döner.
İstemci
(Frontend/Mobil), sonraki tüm isteklerinde bu token'ı HTTP
başlığına (Header) Authorization:
Bearer <TOKEN> şeklinde ekler.
Sunucu her istekte veritabanına gitmek yerine sadece bu imzanın geçerli olup olmadığını kontrol eder. Geçerliyse isteğe izin verir.
Öncelikle şifre hash'leme ve JWT operasyonları için gerekli kütüphaneleri terminalden kurmamız gerekir:
pip install "passlib[bcrypt]" PyJWT python-multipart
passlib[bcrypt]:
Şifreleri güvenli bir şekilde hash'lemek (geri döndürülemez
şekilde şifrelemek) için kullanılır.
PyJWT:
JWT üretmek ve gelen JWT'leri çözmek (decode) için kullanılır.
python-multipart:
Giriş formlarından (OAuth2 standart formu) veri okuyabilmek için
FastAPI tarafından zorunlu tutulur.
security.py
- Şifreleme ve Token Yardımcı FonksiyonlarıŞifre yönetimi ve JWT üretim kodlarını tek bir güvenli modülde topluyoruz.
fromdatetimeimportdatetime,timedelta,timezoneimportjwtfrompasslib.contextimportCryptContextfromtypingimportUnion# Şifreleme algoritması ayarları (BCRYPT)pwd_context= CryptContext(schemes=["bcrypt"], deprecated="auto")# JWT Güvenlik Ayarları (Gerçek projelerde bunları .env dosyasından okumalısınız)SECRET_KEY="SUPER_GIZLI_VE_GUVENLI_ANAHTAR_KODUNUZ_BURAYA"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=30# Üretilen token'ın geçerlilik süresi (30 Dakika)# 1. Şifreyi Hash'leme Fonksiyonudefget_password_hash(password: str) -> str:returnpwd_context.hash(password)# 2. Girilen Şifre ile Hash'li Şifreyi Karşılaştırma Fonksiyonudefverify_password(plain_password: str,hashed_password: str) -> bool:returnpwd_context.verify(plain_password, hashed_password)# 3. JWT Token Üretme Fonksiyonu (Utility Function To Create JWT Token)defcreate_access_token(data: dict,expires_delta: Union[timedelta,None] =None) -> str:to_encode= data.copy()# Süre kısıtlaması eklemeifexpires_delta:expire= datetime.now(timezone.utc) + expires_deltaelse:expire= datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp": expire})# Token içine son kullanma tarihini (expiration) ekliyoruz# Token'ı imzalayıp şifreliyoruzencoded_jwt= jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)returnencoded_jwt
Şimdi kullanıcıların kullanıcı
adı ve şifreleriyle giriş yapabileceği, FastAPI'nin yerleşik
OAuth2PasswordRequestForm
standardına uygun giriş rotasını yazalım.
routers/auth.py
- Giriş ve Yetkilendirme RotalarıfromfastapiimportAPIRouter,Depends,HTTPException,statusfromfastapi.securityimportOAuth2PasswordRequestFormfromsqlalchemy.ormimportSessionfromdatetimeimporttimedeltaimportmodels,securityfromdatabaseimportget_dbrouter= APIRouter(tags=["Kimlik Doğrulama"])@router.post("/token")deflogin_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(),db: Session = Depends(get_db)):# 1. Veritabanında kullanıcıyı arauser= db.query(models.SellerModel).filter(models.SellerModel.username == form_data.username).first()# 2. Kullanıcı yoksa veya şifre yanlışsa HTTP 401 Unauthorized hatası dönifnotuserornotsecurity.verify_password(form_data.password, user.hashed_password):raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Kullanıcı adı veya şifre hatalı!",headers={"WWW-Authenticate":"Bearer"},)# 3. Giriş başarılıysa JWT Token üretaccess_token_expires= timedelta(minutes=security.ACCESS_TOKEN_EXPIRE_MINUTES)access_token= security.create_access_token(data={"sub": user.username}, expires_delta=access_token_expires)# OAuth2 standardına uygun yanıt dönüyoruzreturn{"access_token": access_token,"token_type":"bearer"}
Bir rotayı korumak, o rotaya istek
atan kullanıcının HTTP Header'ında geçerli bir JWT gönderip
göndermediğini kontrol etmek demektir. FastAPI'nin Depends
mekanizması ile bunu bir bağımlılık haline getireceğiz.
dependencies.py
- Mevcut Kullanıcıyı Doğrulama (Get Current User)fromfastapiimportDepends,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearerimportjwtfromsqlalchemy.ormimportSessionimportsecurity,modelsfromdatabaseimportget_db# Token'ın hangi URL'den alınacağını FastAPI'ye bildiriyoruz (Swagger UI entegrasyonu için)oauth2_scheme= OAuth2PasswordBearer(tokenUrl="token")defget_current_user(token: str = Depends(oauth2_scheme),db: Session = Depends(get_db)):credentials_exception= HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Geçersiz veya süresi dolmuş kimlik doğrulaması!",headers={"WWW-Authenticate":"Bearer"},)try:# Gelen token'ı gizli anahtarımızla çözüyoruz (Decode)payload= jwt.decode(token, security.SECRET_KEY, algorithms=[security.ALGORITHM])username: str = payload.get("sub")ifusernameisNone:raisecredentials_exceptionexceptjwt.PyJWTError:raisecredentials_exception# Veritabanından kullanıcıyı kontrol etuser= db.query(models.SellerModel).filter(models.SellerModel.username == username).first()ifuserisNone:raisecredentials_exceptionreturnuser# İstek yapan aktif satıcı nesnesini dönüyoruz
routers/product.py
Aşağıdaki rotada, bir ürün
eklenirken current_user:
models.SellerModel = Depends(get_current_user)
parametresi eklenmiştir. Bu sayede, geçerli bir JWT taşımayan hiç
kimse bu rotaya erişemez ve ürün ekleyemez.
fromfastapiimportAPIRouter,Depends,statusfromsqlalchemy.ormimportSessionimportmodels,schemasfromdatabaseimportget_dbfromdependenciesimportget_current_user# Korumalı mekanizmamızrouter= APIRouter(prefix="/products", tags=["Ürün İşlemleri"])# KORUMALI ROTA (Protecting Routes)@router.post("/", response_model=schemas.ProductResponse, status_code=status.HTTP_201_CREATED)defcreate_product_for_seller(product: schemas.ProductCreate,db: Session = Depends(get_db),current_user: models.SellerModel = Depends(get_current_user)# Giriş zorunlu!):# Ürün eklenirken otomatik olarak giriş yapan satıcının id'si (seller_id) atanıyornew_product= models.ProductModel(name=product.name,price=product.price,is_active=product.is_active,seller_id=current_user.id# Token'dan gelen güvenli kullanıcı ID'si)db.add(new_product)db.commit()db.refresh(new_product)returnnew_product
main.py)Yazdığımız tüm bağımsız ve güvenli modülleri ana dosyamızda tek çatı altında topluyoruz:
fromfastapiimportFastAPIfromdatabaseimportengineimportmodelsfromroutersimportauth,seller,product# Tüm modüller# Veritabanı tablolarını oluşturmodels.Base.metadata.create_all(bind=engine)app= FastAPI(title="Tam Teşekküllü FastAPI Kurs Projesi",description="OAuth2, JWT Kimlik Doğrulama, İlişkisel SQLAlchemy ORM ve APIRouter Modüler Mimari İçerir.",version="2.0.0")# Router'ları Sırayla Dahil Ediyoruzapp.include_router(auth.router)# /token rotası içinapp.include_router(seller.router)# /sellers rotaları içinapp.include_router(product.router)# /products rotaları için@app.get("/")defread_root():return{"durum":"Sistem Aktif","mesaj":"Güvenli API dökümantasyonu için lütfen /docs adresini ziyaret edin."}
Uygulamayı uvicorn
main:app --reload
ile başlattıktan sonra http://127.0.0
adresine gittiğinizde:
Sağ üst köşede bir "Authorize" kilidi göreceksiniz.
Önce /sellers
rotasından yeni bir satıcı hesabı oluşturun. (Bu aşamada
şifreniz arka planda bcrypt
ile kırılması imkansız hale getirilerek hash'lenir).
Ardından "Authorize" butonuna basarak oluşturduğunuz kullanıcı adı ve şifre ile giriş yapın.
Arka planda
/token
rotası çalışır, şifrenizi kontrol eder ve size 30 dakika
geçerli bir JWT Token verir. Swagger bu token'ı tarayıcı
hafızasına alır.
Artık korumalı olan POST
/products
rotasına gidip istek attığınızda, sistem kimliğinizi otomatik
tanır ve ürünü doğrudan sizin hesabınıza bağlayarak
kaydeder.
FastAPI - The Complete Course (Beginner + Advanced) eğitim müfredatını, istediğiniz kurallar doğrultusunda (süreleri, ders sayılarını, önizleme yazılarını ve anlamsız satırları temizleyerek) tamamen yapılandırdım. İçeriği özetlemeden, her bir bölümün mantığını, ne işe yaradığını ve nasıl uygulanacağını detaylı konu açıklamaları, kod örnekleri ve pratik uygulamalarla ele aldım.
Eğitim içeriği oldukça kapsamlı ve uzun olduğu için, yönergeniz doğrultusunda ilk büyük parçayı (Giriş, Temel Python, FastAPI Kurulumu ve Proje 1) eksiksiz, derinlemesine ve uygulamalı olarak aşağıda hazırladım.
Bu kurs, modern backend (arka plan) geliştirme dünyasında en çok tercih edilen Python framework'lerinden biri olan FastAPI'yi sıfırdan uzmanlık seviyesine kadar öğretmeyi amaçlar. "Öğrenecekleriniz" başlığı altındaki maddelerin ifade ettiği yetkinlikler şunlardır:
Gerçek Zamanlı RESTful API'ler ve Full Stack Uygulamalar Geliştirmek: Sadece veri alışverişi yapan köprüler (API) kurmayı değil, aynı zamanda bu verileri bir arayüzle birleştirerek eksiksiz çalışan dinamik web uygulamaları üretmeyi öğrenirsiniz.
Canlıya Alma (Deployment): Kendi bilgisayarınızda yazdığınız ve sadece sizin erişebildiğiniz uygulamaları, bulut sunucularına taşıyarak tüm dünyanın erişebileceği aktif web sitelerine dönüştürme yeteneği kazanırsınız.
Modern Kimlik Doğrulama ve Yetkilendirme (Authentication & Authorization): Bir web uygulamasının en kritik noktası güvenliktir. Kullanıcıların şifrelerini veri tabanına açık metin olarak kaydetmek yerine BCrypt algoritmasıyla geri döndürülemez şekilde şifrelemeyi (hashing) ve kullanıcı oturumlarını güvenli şekilde yönetmek için JWT (JSON Web Token) mekanizmasını entegre etmeyi öğrenirsiniz. Bu sayede "Giriş Yapmış Kullanıcı" ile "Yönetici (Admin)" yetkilerini birbirinden ayırabilirsiniz.
Satır Satır Kodlama ve Mantık Anlaşılması: Kurs boyunca hazır kod kopyalamak yerine, her bir kod satırı mantığı açıklanarak yazılır. Bu, ezberlemek yerine backend mimarisinin çalışma prensibini kavramanızı sağlar.
Veri Tabanı İlişkileri Oluşturma: Gerçek hayattaki veriler birbiriyle ilişkilidir (Örn: Bir kullanıcının birden fazla görevi/todo'su olması). SQLAlchemy ORM aracılığıyla veri tabanında "Bire-Çok" (One-to-Many) gibi tablo ilişkilerini profesyonel düzeyde kurmayı öğrenirsiniz.
FastAPI tamamen Python tabanlı ve modern Python özelliklerini (tip belirteçleri, asenkron programlama vb.) yoğun şekilde kullanan bir framework'tür. Bu yüzden güçlü bir API geliştirmeden önce Python'ın temellerine ve Nesne Yönelimli Programlama (OOP) mantığına tam hakim olmak gerekir.
Python'da verileri hafızada tutmak için değişkenleri kullanırız. FastAPI dünyasında, dışarıdan gelen HTTP isteklerindeki verileri işlerken bu değişken tipleri hayati önem taşır.
Uygulama Örneği: Kullanıcıdan alınan verileri formatlayarak ekrana yazdırma ve tip dönüşümü.
# Kullanıcıdan string (metin) türünde veri almakullanici_adi= input("Lütfen kullanıcı adınızı giriniz: ")# Kullanıcıdan alınan her girdi varsayılan olarak string'dir.# Matematiksel işlem yapabilmek için int() veya float() ile dönüştürmeliyiz.kullanici_yasi= int(input("Lütfen yaşınızı giriniz: "))# String Formatting (f-string) kullanımıgosterim_metni=f"Sisteme Hoş Geldiniz,{kullanici_adi}! 10 yıl sonra{kullanici_yasi +10}yaşında olacaksınız."print(gosterim_metni)
API'ler genellikle JSON formatında veri üretir ve tüketir. Python'daki Sözlük (Dictionary) yapısı JSON'ın doğrudan karşılığıdır. Listeler ise veri tabanından çekilen çoklu kayıtları temsil eder.
Uygulama Örneği: Bir API'den dönebilecek örnek bir "Kitap" veri yapısının simülasyonu.
# Sözlük (Dictionary) yapısı: Key-Value (Anahtar-Değer) ilişkisikitap_detay= {"id":1,"baslik":"FastAPI Kılavuzu","yazar":"Tarık","sayfa_sayisi":350,"kategoriler": ["Yazılım","Eğitim"]# Liste (List) yapısı}# Demet (Tuple): Değiştirilemez veri listesi (Örn: Sabit durum kodları)durum_kodlari= ("AKTIF","PASIF","ASKIDA")# Set (Küme): Benzersiz elemanlar tutarbenzersiz_etiketler= {"python","api","python"}# Çıktıda sadece bir kez 'python' yer alır.print(f"Kitap Adı:{kitap_detay['baslik']}")
Gelen isteklerin geçerliliğini kontrol etmek (Örn: Şifre doğru mu? Yaş sınırı geçildi mi?) ve listeler üzerinde dönerek veri ayıklamak için kullanılır.
# Bir todo listesindeki tamamlanmamış görevleri filtreleme uygulamasıgörevler= [{"baslik":"Kod yaz","tamamlandi":True},{"baslik":"API Test et","tamamlandi":False},{"baslik":"Dokümantasyon oku","tamamlandi":False}]print("Tamamlanması Gereken Görevler:")forgorevingörevler:ifnotgorev["tamamlandi"]:print(f"-{gorev['baslik']}")
Kodun tekrar kullanılabilir olmasını sağlar. FastAPI'de yazdığımız her bir endpoint (uç nokta) aslında özelleştirilmiş birer Python fonksiyonudur.
# İki sayıyı toplayan ve tip belirteci (Type Hinting) kullanan fonksiyon# Modern Python ve FastAPI bu tip belirteçlerini veri doğrulaması için zorunlu kılar.defvergi_hesapla(fiyat: float,vergi_orani: float =0.20) -> float:returnfiyat + (fiyat * vergi_orani)toplam_tutar= vergi_hesapla(100.0)print(f"Toplam Tutar (Vergili):{toplam_tutar}")
Büyük projelerde veri tabanı modellerini ve iş mantıklarını sınıflar (Class) halinde tasarlamak gerekir.
Kapsülleme (Encapsulation): Veriyi dış müdahalelerden koruma (Gizli değişkenler).
Kalıtım (Inheritance): Bir sınıfın özelliklerini başka bir sınıfa aktarma (Örn: Kullanıcı sınıfından Yönetici sınıfı türetme).
Çok Biçimlilik (Polymorphism): Aynı isimdeki fonksiyonun farklı sınıflarda farklı çalışması.
Kompozisyon (Composition): Bir sınıfın içinde başka bir sınıfın nesnesini barındırma.
Uygulama Örneği: Kahraman Savaş Simülasyonu (Müfredattaki Hero Battle mantığı).
classSilah:def__init__(self,isim: str,hasar: int):self.isim = isimself.hasar = hasarclassKahraman:def__init__(self,isim: str,saglik: int,silah: Silah):self.isim = isimself.saglik = saglikself.silah = silah# Composition: Kahraman sınıfı içinde Silah nesnesi barındırılıyor.defsaldir(self,dusman:'Kahraman'):dusman.saglik -= self.silah.hasarprint(f"{self.isim},{self.silah.isim}ile saldırdı!{dusman.isim}{self.silah.hasar}hasar aldı. Kalan Sağlık:{dusman.saglik}")# Uygulama çalıştırmakilic= Silah("Efsanevi Kılıç",25)buyu_asasi= Silah("Ateş Asası",35)savasci= Kahraman("Aragon",100, kilic)buyucu= Kahraman("Gandalf",80, buyu_asasi)savasci.saldir(buyucu)
FastAPI, Python'ın async/await
yapısını destekleyen, yüksek performanslı ve otomatik olarak
Swagger API dokümantasyonu üreten modern bir web framework'üdür.
Projelerimizin bağımlılıklarının bilgisayarımızın işletim sistemindeki küresel Python paketleriyle çakışmaması için her projeye özel izole bir alan (Virtual Environment) oluştururuz.
Windows için Kurulum Komutları:
python -m venv venv.\venv\Scripts\activate
Mac / Linux için Kurulum Komutları:
python3 -m venv venvsource venv/bin/activate
FastAPI projesini çalıştırmak için asenkron bir sunucuya (Uvicorn) ihtiyacımız vardır.
pip install fastapi uvicorn
Bu projede veri tabanı kullanmadan, bellekte (hafızada) tutulan bir kitap listesi üzerinden temel HTTP İstek Metotlarını (GET, POST, PUT, DELETE) ve parametre yönetimini öğreneceğiz.
GET: Sunucudan veri çekmek için kullanılır (Kitapları listeleme).
POST: Sunucuya yeni veri göndermek için kullanılır (Yeni kitap ekleme).
PUT: Sunucudaki mevcut bir veriyi güncellemek için kullanılır (Kitap bilgilerini değiştirme).
DELETE: Sunucudan veri silmek için kullanılır (Kitap silme).
Yol
Parametresi (Path Parameter): URL içinde dinamik olarak gönderilen
değişmeyen parça (Örn: /kitaplar/3
-> id'si 3 olan kitap).
Sorgu Parametresi (Query
Parameter): URL sonuna ?
ile eklenen filtreleme amaçlı veriler (Örn:
/kitaplar?yazar=Tarık).
main.py)
Aşağıdaki kodu bilgisayarınızda
bir dosyaya kaydedip uvicorn
main:app --reload
komutuyla çalıştırdığınızda, tarayıcınızdan http://127.0.0
adresine giderek otomatik hazırlanan arayüzden test edebilirsiniz.
fromfastapiimportFastAPI,Bodyapp= FastAPI()# Bellek üzerinde geçici veri tabanı simülasyonuKITAPLAR= [{"id":1,"baslik":"Python Öğreniyorum","yazar":"Ahmet","kategori":"Yazılım"},{"id":2,"baslik":"FastAPI ile Backend","yazar":"Mehmet","kategori":"Yazılım"},{"id":3,"baslik":"Suç ve Ceza","yazar":"Dostoyevski","kategori":"Edebiyat"},]# 1. TÜM KİTAPLARI GETİRME (GET)@app.get("/kitaplar")asyncdeftum_kitaplari_getir():returnKITAPLAR# 2. YOL PARAMETRESİ İLE TEK BİR KİTAP GETİRME (Path Parameter - GET)@app.get("/kitaplar/{kitap_id}")asyncdefkitap_getir(kitap_id: int):forkitapinKITAPLAR:ifkitap["id"] == kitap_id:returnkitapreturn{"hata":"Aradığınız kitap bulunamadı."}# 3. SORGU PARAMETRESİ İLE FİLTRELEME (Query Parameter - GET)# URL Örneği: /kitaplar/filtrele/?kategori=Yazılım@app.get("/kitaplar/filtrele/")asyncdefkategoriye_gore_kitaplar(kategori: str):eslesen_kitaplar= []forkitapinKITAPLAR:ifkitap["kategori"].lower() == kategori.lower():eslesen_kitaplar.append(kitap)returneslesen_kitaplar# 4. YENİ KİTAP EKLEME (POST)@app.post("/kitaplar/olustur")asyncdefyeni_kitap_ekle(yeni_kitap=Body()):KITAPLAR.append(yeni_kitap)return{"mesaj":"Kitap başarıyla eklendi.","eklenen_kitap": yeni_kitap}# 5. KİTAP GÜNCELLEME (PUT)@app.put("/kitaplar/guncelle")asyncdefkitap_guncelle(guncellenecek_kitap=Body()):foriinrange(len(KITAPLAR)):ifKITAPLAR[i]["id"] == guncellenecek_kitap.get("id"):KITAPLAR[i] = guncellenecek_kitapreturn{"mesaj":"Kitap başarıyla güncellendi.","guncel_hali": guncellenecek_kitap}return{"hata":"Güncellenmek istenen kitap bulunamadı."}# 6. KİTAP SİLME (DELETE)@app.delete("/kitaplar/sil/{kitap_id}")asyncdefkitap_sil(kitap_id: int):foriinrange(len(KITAPLAR)):ifKITAPLAR[i]["id"] == kitap_id:silinen= KITAPLAR.pop(i)return{"mesaj":"Kitap başarıyla silindi.","silinen_kitap": silinen}return{"hata":"Silinmek istenen kitap bulunamadı."}
İlk projemizde verileri
doğrulamadan doğrudan Body()
sözlüğü olarak kabul etmiştik. Ancak gerçek üretim
ortamlarında (production) istemciden gelen verilerin veri tiplerini
doğrulamak, eksik verileri yakalamak ve istemciye anlamlı hata
mesajları dönmek zorunludur. Bu bölümde Pydantic v2
kütüphanesini kullanarak veri modellemeyi ve HTTP durum kodlarını
(Status Codes) yönetmeyi öğreneceğiz.
Pydantic, Python'ın tip
belirteçlerini (Type Hinting) kullanarak gelen verileri otomatik
olarak doğrular ve dönüştürür. Field
yapısı sayesinde alanların alabileceği minimum/maksimum
değerleri, karakter sınırlarını ve varsayılan şablonları
belirleyebiliriz.
Bir API sadece veri dönmez; işlemin sonucunu standart HTTP durum kodlarıyla bildirir.
200 OK: İşlem başarılı.
201 Created: Yeni bir kaynak başarıyla oluşturuldu.
404 Not Found: Aranan kaynak sunucuda bulunamadı.
400 Bad Request: Geçersiz istek veya kural ihlali.
books2.py)fromtypingimportOptionalfromfastapiimportFastAPI,Path,Query,HTTPException,statusfrompydanticimportBaseModel,Fieldapp= FastAPI()# Pydantic Şeması (Veri Doğrulama Modeli)classKitapIstek(BaseModel):id: Optional[int] = Field(description="İsteğe bağlı ID alanı", default=None)baslik: str = Field(min_length=3, max_length=100, description="Kitap başlığı en az 3 karakter olmalıdır.")yazar: str = Field(min_length=2, description="Yazar adı zorunludur.")sayfa_sayisi: int = Field(gt=0, description="Sayfa sayısı 0'dan büyük olmalıdır.")puan: int = Field(gt=0, le=5, description="Puan 1 ile 5 arasında olmalıdır.")# Swagger arayüzünde varsayılan bir şablon göstermek için yapılandırmamodel_config= {"json_schema_extra": {"example": {"baslik":"Sefiller","yazar":"Victor Hugo","sayfa_sayisi":1200,"puan":5}}}# Mock Veri TabanıKITAPLAR= [{"id":1,"baslik":"Simyacı","yazar":"Paulo Coelho","sayfa_sayisi":180,"puan":5},{"id":2,"baslik":"Dune","yazar":"Frank Herbert","sayfa_sayisi":700,"puan":4}]# 1. BELİRLİ BİR KİTABI ID İLE GETİRME VE 404 KONTROLÜ# Path(gt=0) ile URL'den gelen id parametresinin 0'dan büyük olması zorunlu tutulur.@app.get("/kitap/{kitap_id}", status_code=status.HTTP_200_OK)asyncdefkitap_oku(kitap_id: int = Path(gt=0)):forkitapinKITAPLAR:ifkitap["id"] == kitap_id:returnkitapraiseHTTPException(status_code=404, detail="Aradığınız kitap bulunamadı.")# 2. DOĞRULANMIŞ VERİ İLE KİTAP EKLEME (POST - 201 Created)@app.post("/kitap-ekle", status_code=status.HTTP_201_CREATED)asyncdefkitap_olustur(yeni_kitap: KitapIstek):# Gelen pydantic modelini sözlüğe dönüştürüyoruzkitap_sozluk= yeni_kitap.model_dump()# ID otomatik atama simülasyonukitap_sozluk["id"] = len(KITAPLAR) +1ifKITAPLARelse1KITAPLAR.append(kitap_sozluk)return{"mesaj":"Kitap başarıyla oluşturuldu.","veri": kitap_sozluk}# 3. KİTAP GÜNCELLEME VE HATA YÖNETİMİ@app.put("/kitap-guncelle/{kitap_id}", status_code=status.HTTP_200_OK)asyncdefkitap_guncelle(kitap_id: int,guncel_veri: KitapIstek):fori,kitapinenumerate(KITAPLAR):ifkitap["id"] == kitap_id:guncel_sozluk= guncel_veri.model_dump()guncel_sozluk["id"] = kitap_id# ID'nin değişmesini engelliyoruzKITAPLAR[i] = guncel_sozlukreturn{"mesaj":"Güncelleme başarılı."}raiseHTTPException(status_code=404, detail="Güncellenmek istenen kitap sistemde yok.")
Verileri bellekte tutmak kalıcı bir çözüm değildir. Uygulama her yeniden başladığında veriler silinir. Bu aşamada, Python nesnelerini SQL sorgularına dönüştüren SQLAlchemy ORM (Object-Relational Mapping) yapısını ve hafif bir ilişkisel veri tabanı olan SQLite3 entegrasyonunu tasarlayacağız.
Projemizi ölçeklenebilir kılmak için dosyaları görevlerine göre ayırmalıyız:
database.py:
Veri tabanı bağlantısını ve oturum (Session) yönetimini
ayarlar.
models.py:
Veri tabanındaki tabloların Python sınıfları olarak karşılığını
tanımlar.
database.py)fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker,declarative_base# SQLite veri tabanı dosyasının konumuSQLALCHEMY_DATABASE_URL="sqlite:///./todosapp.db"# SQLite özelinde 'check_same_thread' ayarını False yapıyoruz çünkü FastAPI asenkron çalışırengine= create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread":False})# Veri tabanı işlemlerini yürütecek oturum fabrikasıSessionLocal= sessionmaker(autocommit=False, autoflush=False, bind=engine)# Modellerimizin türeyeceği ana sınıfBase= declarative_base()# FastAPI bağımlılık enjeksiyonu (Dependency Injection) için oturum fonksiyonu# İşlem bitince veri tabanı bağlantısını otomatik kapatır.defget_db():db= SessionLocal()try:yielddbfinally:db.close()
models.py)fromsqlalchemyimportColumn,Integer,String,Boolean,ForeignKeyfromdatabaseimportBaseclassKullanicilar(Base):__tablename__="users"id= Column(Integer, primary_key=True, index=True)= Column(String, unique=True, index=True)kullanici_adi= Column(String, unique=True, index=True)hashed_sifre= Column(String)is_active= Column(Boolean, default=True)classGörevler(Base):__tablename__="todos"id= Column(Integer, primary_key=True, index=True)baslik= Column(String)aciklama= Column(String)tamamlandi= Column(Boolean, default=False)# İlişki kurulumu: Her görevin bir sahibi (kullanıcı id'si) olmalıdır.sahip_id= Column(Integer, ForeignKey("users.id"))
Uygulamamızdaki görevlerin güvenliğini sağlamak için kullanıcıların kayıt olabilmesi, giriş yapabilmesi ve aldıkları JWT (JSON Web Token) ile güvenli isteklere imza atabilmesi gerekir.
Kayıt Olma: Kullanıcı şifresi ham metin (12345) olarak kaydedilemez. Passlib ve BCrypt kullanarak şifreyi geri döndürülemez bir koda (hash) dönüştürürüz.
Giriş Yapma ve Token Üretimi: Kullanıcı adı ve şifre doğruysa, sunucu kullanıcıya kriptografik olarak imzalanmış kısa süreli bir JWT (erişim anahtarı) verir.
Token Doğrulama: Kullanıcı
veri tabanına bir görev eklemek istediğinde bu anahtarı isteğin
başlığında (Header) sunucuya gönderir. Sunucu anahtarı
doğrular ve sahip_id
bilgisini token içerisinden çözer.
auth.py)Gerekli paketlerin kurulumu:
pip install passlib[bcrypt] pyjwt python-multipartfromdatetimeimportdatetime,timedelta,timezonefromtypingimportAnnotatedfromfastapiimportAPIRouter,Depends,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormfromjoseimportjwt,JWTErrorfrompasslib.contextimportCryptContextfrompydanticimportBaseModelfromsqlalchemy.ormimportSessionfromdatabaseimportget_dbimportmodelsrouter= APIRouter(prefix="/auth", tags=["Kimlik Doğrulama"])# Güvenlik ve Şifreleme AyarlarıSECRET_KEY="CokGizliKriptografikAnahtarUretimOrtamindaDegistirilmeli"ALGORITHM="HS256"bcrypt_context= CryptContext(schemes=["bcrypt"], deprecated="auto")# FastAPI'nin token toplama mekanizması (Swagger arayüzünde kilit ikonu açar)oauth2_bearer= OAuth2PasswordBearer(tokenUrl="auth/token")classKullaniciOlusturIstek(BaseModel):kullanici_adi: str: strsifre: str# Şifre şifreleme ve doğrulama yardımcı fonksiyonlarıdefsifre_hashle(sifre: str):returnbcrypt_context.hash(sifre)defsifre_verifiye_et(duz_sifre,hashlenmis_sifre):returnbcrypt_context.verify(duz_sifre, hashlenmis_sifre)# Token üretme fonksiyonudeftoken_olustur(kullanici_adi: str,kullanici_id: int,sure: timedelta):payload= {"sub": kullanici_adi,"id": kullanici_id,"exp": datetime.now(timezone.utc) + sure}returnjwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)# 1. KULLANICI KAYIT ENDPOINT'I@app.post("/kayit", status_code=status.HTTP_201_CREATED)asyncdefkullanici_kaydet(db: Annotated[Session, Depends(get_db)],istek: KullaniciOlusturIstek):yeni_kullanici= models.Kullanicilar(email=istek.email,kullanici_adi=istek.kullanici_adi,hashed_sifre=sifre_hashle(istek.sifre),is_active=True)db.add(yeni_kullanici)db.commit()return{"mesaj":"Kullanıcı başarıyla oluşturuldu."}# 2. GİRİŞ YAPMA VE TOKEN ALMA ENDPOINT'I@app.post("/token")asyncdeftoken_icin_giris_yap(form_verisi: Annotated[OAuth2PasswordRequestForm, Depends()],db: Annotated[Session, Depends(get_db)]):kullanici= db.query(models.Kullanicilar).filter(models.Kullanicilar.kullanici_adi == form_verisi.username).first()ifnotkullaniciornotsifre_verifiye_et(form_verisi.password, kullanici.hashed_sifre):raiseHTTPException(status_code=401, detail="Kullanıcı adı veya şifre hatalı.")token= token_olustur(kullanici.kullanici_adi, kullanici.id, timedelta(minutes=30))return{"access_token": token,"token_type":"bearer"}# 3. KORUMALI ENDPOINT'LER İÇİN MEVCUT KULLANICIYI DOĞRULAYAN BAĞIMLILIK (Dependency)asyncdefmevcut_kullaniciyi_getir(token: Annotated[str, Depends(oauth2_bearer)]):try:payload= jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])kullanici_adi: str = payload.get("sub")kullanici_id: int = payload.get("id")ifkullanici_adiisNoneorkullanici_idisNone:raiseHTTPException(status_code=401, detail="Geçersiz token bileşenleri.")return{"kullanici_adi": kullanici_adi,"id": kullanici_id}exceptJWTError:raiseHTTPException(status_code=401, detail="Token doğrulanamadı.")
Artık veri tabanındaki görev ekleme endpoints'lerimizi bu kimlik doğrulamaya bağlayabiliriz. Böylece hiç kimse başkasının Todo görevini göremez veya değiştiremez.
@app.post("/todo/ekle", status_code=status.HTTP_201_CREATED)asyncdeftodo_olustur(guncel_kullanici: Annotated[dict, Depends(mevcut_kullaniciyi_getir)],db: Annotated[Session, Depends(get_db)],baslik: str,aciklama: str):# guncel_kullanici dict'i doğrudan JWT'den çözülen güvenli veridir.yeni_govev= models.Görevler(baslik=baslik,aciklama=aciklama,tamamlandi=False,sahip_id=guncel_kullanici.get("id")# Görev doğrudan giriş yapan kullanıcıya bağlanır.)db.add(yeni_govev)db.commit()return{"durum":"Başarılı","görev_id": yeni_govev.id}
Canlı ortamdaki (production) bir uygulamada, veri tabanı modellerinde (tablolarda) değişiklik yaptığınızda (örneğin yeni bir sütun ekleme veya veri tipini değiştirme), veri tabanını sıfırlayamazsınız. İçerideki mevcut verileri kaybetmeden veri tabanı şemasını güvenli bir şekilde güncellemek için veri göçü (migration) araçları kullanılır. Python ekosisteminde SQLAlchemy ile en entegre çalışan araç Alembic'tir.
Alembic Init: Projede Alembic yapılandırma dosyalarını oluşturur.
Revision
(Oluşturma): Modellerdeki değişiklikleri algılayarak
otomatik (--autogenerate)
bir göç senaryosu (Python dosyası) hazırlar.
Upgrade (Yükseltme): Yazılan göç senaryosunu veri tabanına uygulayarak şemayı günceller.
Downgrade (Geri Alma): Eğer yapılan güncelleme bir hataya sebep olduysa, veri tabanını bir önceki kararlı şema sürümüne geri döndürür.
Öncelikle gerekli kütüphaneyi terminal üzerinden kuruyoruz:
pip install alembicArdından projenin ana dizininde Alembic mimarisini başlatıyoruz:
alembic init alembic
Bu komut projenizde alembic.ini
dosyası ve alembic/
klasörü oluşturur.
alembic/env.py
Dosyasının Yapılandırılması
Alembic'in modellerimizi otomatik
algılayabilmesi için env.py
dosyası içerisindeki target_metadata
alanına kendi modellerimizin metadata bilgisini bağlamamız
gerekir.
# alembic/env.py dosyasının içi (İlgili satırlar güncellenir)fromdatabaseimportBase# Projemizdeki Base sınıfı içe aktarılırimportmodels# Modellerimizin yüklenmesi sağlanır# Alembic'in tabloları karşılaştırabilmesi için hedef metadata atanırtarget_metadata= Base.metadata
Modellerimizde (models.py)
değişiklik yaptığımızı varsayalım:
# models.py içindeki Kullanicilar sınıfına yeni sütun eklenmesiclassKullanicilar(Base):__tablename__="users"id= Column(Integer, primary_key=True, index=True)= Column(String, unique=True, index=True)kullanici_adi= Column(String, unique=True, index=True)hashed_sifre= Column(String)is_active= Column(Boolean, default=True)# Yeni eklenen sütun:telefon_numarasi= Column(String, nullable=True)
Terminalden otomatik göç dosyasını oluşturma komutu verilir:
alembic revision --autogenerate -m "kullanicilar tablosuna telefon numarasi eklendi"
Bu komut alembic/versions/
klasörü altında benzersiz bir ID ile yeni bir Python dosyası
üretir. İçeriği otomatik olarak şu şekilde tasarlanır:
# alembic/versions/xxxx_kullanicilar_tablosuna_telefon_numarasi_eklendi.pyfromalembicimportopimportsqlalchemyassadefupgrade():# Veri tabanına yeni sütunu ekleyen fonksiyonop.add_column('users', sa.Column('telefon_numarasi', sa.String(), nullable=True))defdowngrade():# Değişikliği geri almak istiyorsak sütunu silen fonksiyonop.drop_column('users','telefon_numarasi')
Değişikliği veri tabanına yansıtmak (Yükseltmek) için:
alembic upgrade headEğer bir sorun çıkarsa ve bu değişikliği geri almak (Düşürmek) istersek:
alembic downgrade -1
Geliştirilen API endpoints'lerinin
canlı ortama çıkmadan önce hatasız çalıştığından emin
olmak için test otomasyonu şarttır. Pytest,
Python dünyasındaki en popüler test çatısıdır. FastAPI, test
işlemleri için Starlette
tabanlı bir TestClient
sunar.
Canlı veri tabanını (üretim
verilerini) testler sırasında kirletmemek veya bozmamak adına
testler için ayrı bir geçici veri tabanı (Test DB)
kurgulanmalıdır. Pytest'in fixture
mekanizmasını kullanarak her test fonksiyonundan önce temiz bir
test veri tabanı oluşturulmasını ve test bitince silinmesini
sağlayacağız.
Gerekli paketlerin kurulumu:
pip install pytest httpxconftest.py)
Pytest, conftest.py
isimli dosyayı otomatik olarak tanır. Bu dosyada test süresince
kullanılacak ortak bağımlılıklar (fixtures) tanımlanır.
# conftest.pyimportpytestfromfastapi.testclientimportTestClientfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmakerfromdatabaseimportBase,get_dbfrommainimportapp# FastAPI uygulama nesneniz# Testler için izole, geçici bellek içi (in-memory) SQLite veri tabanıTEST_DATABASE_URL="sqlite:///./test_todos.db"engine= create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread":False})TestingSessionLocal= sessionmaker(autocommit=False, autoflush=False, bind=engine)@pytest.fixture(scope="session", autouse=True)defveri_tabanini_kur hazirla():# Test oturumu başladığında tabloları sıfırdan oluştururBase.metadata.create_all(bind=engine)yield# Bütün testler bittiğinde test veri tabanı tablolarını temizlerBase.metadata.drop_all(bind=engine)@pytest.fixturedefdb_oturumu():# Her test fonksiyonu için izole bir veri tabanı oturumu sağlarconnection= engine.connect()transaction= connection.begin()session= TestingSessionLocal(bind=connection)yieldsessionsession.close()transaction.rollback()connection.close()@pytest.fixturedeftest_istemcisi(db_oturumu):# FastAPI'nin gerçek get_db bağımlılığını, test veri tabanı oturumu ile değiştiriyoruz (Override)defoverride_get_db():try:yielddb_oturumufinally:passapp.dependency_overrides[get_db] = override_get_dbyieldTestClient(app)# Test bitince bağımlılık enjeksiyonunu eski haline getirirapp.dependency_overrides.clear()
test_api.py)Aşağıda hem kimlik doğrulama gerektirmeyen (herkese açık) hem de mock (taklit) verilerle korumalı endpoints'lerin nasıl test edileceğine dair işlevsel senaryolar yer almaktadır.
# test_api.pyimportpytestfromfastapiimportstatus# Senaryo 1: Kitap listeleme endpoint'inin testi (GET)deftest_tum_kitaplari_listele(test_istemcisi):# conftest.py dosyasındaki test_istemcisi fixture'ı otomatik enjekte ediliryanit= test_istemcisi.get("/kitaplar")assertyanit.status_code == status.HTTP_200_OKassertisinstance(yanit.json(), list)# Dönen verinin bir liste olduğunu doğrula# Senaryo 2: Geçersiz ID ile kitap çağrıldığında 404 hatasının doğrulanmasıdeftest_gecersiz_id_ile_kitap_bulunamadi_hatasi(test_istemcisi):yanit= test_istemcisi.get("/kitap/9999")assertyanit.status_code == status.HTTP_404_NOT_FOUNDassertyanit.json()["detail"] =="Aradığınız kitap bulunamadı."# Senaryo 3: Pydantic şemasına uymayan eksik veri gönderildiğinde 422 hatasının doğrulanması (POST)deftest_eksik_veri_ile_kitap_olusturma_hatasi(test_istemcisi):# 'yazar' ve 'sayfa_sayisi' alanları eksik bırakıldıhatali_veri= {"baslik":"Eksik Kitap"}yanit= test_istemcisi.post("/kitap-ekle", json=hatali_veri)# 422 Unprocessable Entity: FastAPI/Pydantic'in otomatik fırlattığı veri doğrulama hata koduassertyanit.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY# Senaryo 4: Token simülasyonu ile korumalı Todo ekleme rotasının entegrasyon testideftest_korumali_rotaya_todo_ekleme(test_istemcisi):# Korumalı endpoints'leri test ederken 'Authentication & Authorization' katmanını geçmek için# test_istemcisi başlığına geçerli şablonda bir Bearer token ekleriz (Mock kullanıcı doğrulaması için)headers= {"Authorization":"Bearer sahte_veya_test_jwt_token_icerigi"}# Not: Eğer JWT doğrulaması katıysa, conftest içinde JWT kütüphanesiyle# test süresince geçerli gerçek bir test token'ı üretilip buraya basılabilir.todo_verisi= {"baslik":"Test Görevi","aciklama":"Pytest ile sistem entegrasyonu test ediliyor"}# Parametreleri Query veya Body olarak gönderen endpoint tasarımı test ediliryanit= test_istemcisi.post("/todo/ekle?baslik=Test%20G%C3%B6revi&aciklama=Pytest%20Entegrasyonu",headers=headers)assertyanit.status_code == status.HTTP_201_CREATEDassertyanit.json()["durum"] =="Başarılı"
Testleri çalıştırmak için terminale sadece şu komutu yazmanız yeterlidir:
pytest -v
-v
(verbose) parametresi, hangi test fonksiyonlarının başarıyla
geçtiğini (PASSED)
veya kaldığını (FAILED)
detaylı bir rapor halinde ekrana basar.
FastAPI sadece ham veri (JSON) dönen bir API sunucusu olarak değil, aynı zamanda dinamik HTML sayfaları üreten bir Full Stack web sunucusu olarak da çalışabilir. Bu bölümde, Python kodlarımızı HTML sayfalarıyla birleştirmek için Jinja2 şablon motorunu (template engine) kullanacağız. Ayrıca stil işlemleri için CSS ve dinamik etkileşimler için JavaScript entegrasyonu sağlayacağız.
Statik dosyaların (CSS/JS) ve HTML şablonlarının düzenli çalışması için proje dizinini şu mimariyle tasarlıyoruz:
├── static/│ ├── css/│ │ └── style.css│ └── js/│ └── main.js├── templates/│ ├── layout.html│ ├── login.html│ ├── register.html│ └── todos.html├── main.py└── ...
main.py)Gerekli kütüphaneyi kuruyoruz:
pip install jinja2FastAPI uygulamasında Jinja2 şablon motorunu ve statik dosyaları tanımlıyoruz:
fromfastapiimportFastAPI,Request,Depends,statusfromfastapi.staticfilesimportStaticFilesfromfastapi.templatingimportJinja2Templatesfromfastapi.responsesimportHTMLResponseapp= FastAPI()# CSS ve JS dosyalarımızın yer aldığı 'static' klasörünü sisteme tanıtıyoruzapp.mount("/static", StaticFiles(directory="static"), name="static")# HTML şablonlarımızın yer aldığı 'templates' klasörünü bağlıyoruztemplates= Jinja2Templates(directory="templates")# Örnek Arayüz Rotası (Giriş Sayfası)@app.get("/login-page", response_class=HTMLResponse)asyncdefgiris_sayfasi(request: Request):# Jinja2 şablon motoruna zorunlu olarak 'request' nesnesini geçmeliyizreturntemplates.TemplateResponse("login.html", {"request": request,"sayfa_basligi":"Giriş Yap"})
templates/layout.html)Tüm sayfalarda ortak olan navigasyon barı, CSS ve JS bağlantılarını tek bir dosyada toplamak için Jinja2'nin kalıtım (inheritance) özelliğini kullanıyoruz.
<!DOCTYPE html><htmllang="tr"><head><metacharset="UTF-8"><metaname="viewport"content="width=device-width, initial-scale=1.0"><title>{{ sayfa_basligi }} - TodoApp</title><!-- Ortak CSS Bağlantısı --><linkrel="stylesheet"href="{{ url_for('static', path='/css/style.css') }}"></head><body><navclass="navbar"><ahref="/todo-page"class="logo">🎯 Görev Yöneticisi</a><divclass="nav-links"><ahref="/login-page">Giriş Yap</a><ahref="/register-page">Kayıt Ol</a></div></nav><mainclass="container"><!-- Dinamik olarak değişecek içerik alanı -->{% block icerik %}{% endblock %}</main><!-- Ortak JS Bağlantısı --><scriptsrc="{{ url_for('static', path='/js/main.js') }}"></script></body></html>
templates/register.html)Ana şablondan miras alarak sadece formu yerleştirdiğimiz sayfa tasarımı:
{% extends"layout.html"%}{% block icerik %}<divclass="form-card"><h2>Yeni Hesap Oluştur</h2><formid="registerForm"><divclass="form-group"><labelfor="username">Kullanıcı Adı</label><inputtype="text"id="username"name="username"required></div><divclass="form-group"><labelfor="email">E-posta</label><inputtype="email"id="email"name="email"required></div><divclass="form-group"><labelfor="password">Şifre</label><inputtype="password"id="password"name="password"required></div><buttontype="submit"class="btn">Kayıt Ol</button></form></div>{% endblock %}
static/css/style.css)Uygulamanın profesyonel ve modern görünmesini sağlayacak CSS kodları:
:root {--primary-color: #4f46e5;--background-color: #f3f4f6;--card-color: #ffffff;--text-color: #1f2937;}body {font-family:'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;background-color: var(--background-color);color: var(--text-color);margin:0;}.navbar {background-color: var(--primary-color);padding:1rem2rem;display: flex;justify-content: space-between;align-items: center;}.navbar .logo, .navbar .nav-links a {color: white;text-decoration: none;font-weight: bold;}.container {max-width:800px;margin:2remauto;padding:01rem;}.form-card {background: var(--card-color);padding:2rem;border-radius:8px;box-shadow:04px6pxrgba(0,0,0,0.1);}.form-group {margin-bottom:1.5rem;}.form-group label {display: block;margin-bottom:0.5rem;}.form-group input {width:100%;padding:0.75rem;border:1pxsolid #ccc;border-radius:4px;box-sizing: border-box;}.btn {background-color: var(--primary-color);color: white;padding:0.75rem1.5rem;border: none;border-radius:4px;cursor: pointer;width:100%;}
static/js/main.js)
Form verilerini FastAPI backend
servislerimize asenkron olarak (fetch
ile) göndermek ve kullanıcıyı yönlendirmek için kullanılan
JavaScript mantığı:
document.addEventListener("DOMContentLoaded", () => {constregisterForm= document.getElementById("registerForm");if(registerForm) {registerForm.addEventListener("submit",async(e) => {e.preventDefault();// Sayfanın yenilenmesini engelleconstformData=newFormData(registerForm);constdata= Object.fromEntries(formData.entries());// Backend API'mize JSON isteği atıyoruzconstresponse=awaitfetch("/auth/kayit", {method:"POST",headers: {"Content-Type":"application/json"},body: JSON.stringify({kullanici_adi: data.username,: data.email,sifre: data.password})});if(response.ok) {alert("Kayıt başarılı! Giriş sayfasına yönlendiriliyorsunuz.");window.location.href ="/login-page";}else{consterrorData=awaitresponse.json();alert("Hata: "+ (errorData.detail ||"Kayıt başarısız."));}});}});
Geliştirdiğimiz Full Stack uygulamasını yerel bilgisayarımızdan çıkarıp tüm dünyanın erişebileceği canlı bir bulut sunucusuna (Render) taşıyacağız. Bu süreçte yerel SQLite veri tabanımızı, bulut ortamında çalışan ve üretime hazır (production-ready) bir PostgreSQL veri tabanına dönüştüreceğiz.
Canlı sunucunun (Render) projemizi nasıl çalıştıracağını bilmesi ve gerekli Python kütüphanelerini yükleyebilmesi için iki temel dosyaya ihtiyacımız vardır:
requirements.txt
Dosyasının Oluşturulması
Projede kullanılan tüm paketleri
listeler. PostgreSQL entegrasyonu için sürücü paketi olan
psycopg2-binary
kütüphanesini de listeye ekliyoruz.
fastapiuvicornsqlalchemypydanticjinja2passlib[bcrypt]pyjwtpython-multipartpsycopg2-binary
database.py
Güncellemesi)
Yerel geliştirme yaparken SQLite,
canlı sunucuda ise Render'ın bize vereceği PostgreSQL bağlantı
adresini (DATABASE_URL) dinamik olarak seçmek için Python'ın os
modülünü kullanırız.
importosfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker,declarative_base# Çevre değişkenlerinde DATABASE_URL varsa onu kullan (Render ortamı), yoksa yerel SQLite kullanDATABASE_URL= os.getenv("DATABASE_URL","sqlite:///./todosapp.db")# PostgreSQL URL şeması 'postgres://' ile başlıyorsa, SQLAlchemy v2 uyumluluğu için 'postgresql://' ile değiştirilirifDATABASE_URL.startswith("postgres://"):DATABASE_URL= DATABASE_URL.replace("postgres://","postgresql://",1)# Veri tabanı motoru konfigürasyonuif"sqlite"inDATABASE_URL:engine= create_engine(DATABASE_URL, connect_args={"check_same_thread":False})else:engine= create_engine(DATABASE_URL)# PostgreSQL için ek argümana gerek yokturSessionLocal= sessionmaker(autocommit=False, autoflush=False, bind=engine)Base= declarative_base()
Render, kodları doğrudan bir GitHub deposundan (repository) çekerek canlıya alır. Bu yüzden projemizi Git ile sarmalayıp GitHub'a yüklemeliyiz.
Terminalden uygulanacak adımlar:
# Git deposunu başlatgit init# Gereksiz dosyaların (venv, test DB, pycache) sunucuya yüklenmesini engellemek için .gitignore oluşturecho"venv/\n*.db\n__pycache__/"> .gitignore# Dosyaları sahneye ekle ve ilk kaydı oluşturgit add .git commit -m"Canlıya alma hazırlığı tamamlandı"# GitHub üzerinde yeni ve boş bir repository oluşturduktan sonra aşağıdaki komutlarla kodu uzak sunucuya gönderingit branch -M maingit remote add origin https://github.comgit push -u origin main
Render.com adresine giriş yapın ve panelden New > PostgreSQL seçeneğini seçin.
Veri tabanına
bir isim verin (todo-prod-db),
bölgeyi (Region) seçin ve Create Database butonuna
tıklayın.
Veri tabanı oluşturulduğunda panelde yer alan Internal Database URL bilgisini kopyalayın.
Render panelinden New > Web Service seçeneğine tıklayın.
GitHub hesabınızı bağlayarak az önce oluşturduğunuz ve kodlarınızı yüklediğiniz depoyu (Repository) seçin.
Açılan konfigürasyon sayfasındaki ayarları şu şekilde doldurun:
Name:
my-fastapi-todo-app
Runtime:
Python
3
Build
Command: pip
install -r requirements.txt (Bağımlılıkları
yükleme komutu)
Start
Command: uvicorn
main:app --host 0.0.0.0 --port $PORT
(Uygulamayı başlatma komutu)
Aynı sayfanın altındaki Advanced butonuna tıklayarak Environment Variables (Çevre Değişkenleri) alanına gelin.
Add Environment Variable seçeneğine tıklayın.
Key:
DATABASE_URL
Value: (Adım 1'de kopyaladığınız PostgreSQL Internal Database URL bilgisini buraya yapıştırın)
En alttaki Create Web Service butonuna tıklayın.
Render önce projeniz için izole
bir linux konteyneri kuracak, ardından requirements.txt
içindeki paketleri yükleyecek ve uvicorn
sunucusunu ayağa kaldıracaktır. Birkaç dakika içinde size
https://onrender.com
şeklinde dünyanın her yerinden erişilebilir, canlı ve veri
tabanı bağlantısı güvenli bir web adresi teslim edecektir.
Bu bölümde, canlı ortamdaki (production) Full Stack uygulamamızı kurumsal düzeydeki standartlara ulaştırmak için üç kritik mimariyi sisteme entegre edeceğiz:
Güvenlik ve İletişim: E-posta doğrulama ve şifre sıfırlama mekanizmaları.
Yönetim Katmanı: Sadece yönetici (Admin) rolüne sahip kullanıcıların erişebileceği dinamik bir admin paneli arayüzü ve backend mimarisi.
Veri Tabanı Optimizasyonu: Canlıdaki PostgreSQL veri tabanımızın yük altında milisaniyeler içinde yanıt vermesini sağlayacak doğru indeksleme (indexing) stratejileri.
Kullanıcıların sisteme sahte e-postalarla kaydolmasını engellemek ve şifrelerini unuttuklarında güvenli bir şekilde sıfırlayabilmelerini sağlamak için zamana bağlı şifreli token (Timed Token) mimarisi ve arka planda e-posta gönderim işlemlerini üstlenecek FastAPI-Mail entegrasyonu kuracağız.
pip install fastapi-mailmodels.py)
Kullanıcının e-posta doğrulama
durumunu takip etmek için is_verified
alanını ekliyoruz.
# models.py içindeki Kullanicilar sınıfına ekleme yapın:classKullanicilar(Base):__tablename__="users"# ... önceki alanlar ...is_verified= Column(Boolean, default=False)# E-posta doğrulandı mı?
auth_services.py)
FastAPI-Mail yapılandırmasını
kuruyoruz ve token üretim mantığını kurguluyoruz. Güvenlik
amacıyla, şifre sıfırlama ve e-posta doğrulama token'larının
içine kısa bir son kullanma süresi (exp)
ekliyoruz.
importosfromdatetimeimportdatetime,timedelta,timezonefromjoseimportjwt,JWTErrorfromfastapi_mailimportConnectionConfig,FastMail,MessageSchema,MessageTypefromfastapiimportHTTPException# E-Posta Sunucu Ayarları (Çevre değişkenlerinden beslenir)mail_config= ConnectionConfig(MAIL_USERNAME=os.getenv("MAIL_USERNAME","ornek_email@gmail.com"),MAIL_PASSWORD=os.getenv("MAIL_PASSWORD","uygulama_ozel_sifresi_xyz"),MAIL_FROM=os.getenv("MAIL_FROM","ornek_email@gmail.com"),MAIL_PORT=587,MAIL_SERVER="://gmail.com",MAIL_STARTTLS=True,MAIL_SSL_TLS=False,USE_CREDENTIALS=True,VALIDATE_CERTS=True)SECRET_KEY="CokGizliKriptografikAnahtarUretimOrtamindaDegistirilmeli"ALGORITHM="HS256"# E-posta doğrulama veya şifre sıfırlama için 15 dakika geçerli token üretirdefozel_islem_token_uret(: str,islem_tipi: str) -> str:payload= {"sub": email,"type": islem_tipi,# "eposta_dogrulama" veya "sifre_sifirlama""exp": datetime.now(timezone.utc) + timedelta(minutes=15)}returnjwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)# Asenkron olarak kullanıcıya e-posta gönderen fonksiyonasyncdefasenkron_eposta_gonder(: str,subject: str,body_html: str):message= MessageSchema(subject=subject,recipients=[email],body=body_html,subtype=MessageType.html)fm= FastMail(mail_config)awaitfm.send_message(message)
auth.py
Eki)Kullanıcı kayıt olduğunda doğrulama e-postası tetiklenir. Şifre sıfırlama istendiğinde ise benzersiz bir link üretilerek gönderilir.
fromfastapiimportAPIRouter,Depends,status,Requestfromsqlalchemy.ormimportSessionfromdatabaseimportget_dbimportmodelsfromauth_servicesimportozel_islem_token_uret,asenkron_eposta_gonder,SECRET_KEY,ALGORITHMfromjoseimportjwt,JWTErrorfrompasslib.contextimportCryptContextrouter= APIRouter(prefix="/auth", tags=["Gelişmiş Kimlik Yönetimi"])bcrypt_context= CryptContext(schemes=["bcrypt"], deprecated="auto")# 1. KAYIT ESNASINDA E-POSTA TETİKLEME (GÜNCELLEME)# Kullanıcı başarıyla db.add() edildikten hemen sonra çalışır:asyncdefdogrulama_epostasi_gonder(: str,request: Request):token= ozel_islem_token_uret(email,"eposta_dogrulama")# Canlı ortam veya yerel host adresini dinamik yakalarana_url= str(request.base_url)dogrulama_linki=f"{ana_url}auth/dogrula?token={token}"html_icerik=f"""<h3>TodoApp'e Hoş Geldiniz!</h3><p>Hesabınızı aktifleştirmek için lütfen aşağıdaki bağlantıya tıklayın:</p><a href="{dogrulama_linki}" target="_blank">Hesabımı Doğrula</a><p>Bu bağlantı 15 dakika içinde geçerliliğini yitirecektir.</p>"""awaitasenkron_eposta_gonder(email,"TodoApp Hesap Doğrulama", html_icerik)# 2. E-POSTA DOĞRULAMA ENDPOINT'I@router.get("/dogrula")asyncdefeposta_dogrula(token: str,db: Session = Depends(get_db)):try:payload= jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])= payload.get("sub")islem_tipi= payload.get("type")ifislem_tipi !="eposta_dogrulama":raiseHTTPException(status_code=400, detail="Geçersiz işlem tipi.")kullanici= db.query(models.Kullanicilar).filter(models.Kullanicilar.email == email).first()ifnotkullanici:raiseHTTPException(status_code=404, detail="Kullanıcı bulunamadı.")kullanici.is_verified =Truedb.commit()return{"mesaj":"E-posta adresiniz başarıyla doğrulandı! Giriş yapabilirsiniz."}exceptJWTError:raiseHTTPException(status_code=400, detail="Token geçersiz veya süresi dolmuş.")# 3. ŞİFRE SIFIRLAMA TALEBİ OLUŞTURMA@router.post("/sifre-sifirlama-talebi")asyncdefsifre_sifirlama_talebi(: str,request: Request,db: Session = Depends(get_db)):kullanici= db.query(models.Kullanicilar).filter(models.Kullanicilar.email == email).first()ifnotkullanici:# Güvenlik nedeniyle "Kullanıcı yok" demek yerine sahte bir başarılı mesajı dönmek siber saldırıları (user enumeration) engeller.return{"mesaj":"Eğer e-posta sistemde kayıtlıysa, sıfırlama linki gönderilecektir."}token= ozel_islem_token_uret(email,"sifre_sifirlama")ana_url= str(request.base_url)sifrilama_linki=f"{ana_url}auth/sifre-yenile-sayfasi?token={token}"html_icerik=f"""<h3>Şifre Sıfırlama Talebi</h3><p>Yeni şifrenizi belirlemek için lütfen aşağıdaki bağlantıyı kullanın:</p><a href="{sifrilama_linki}">Şifremi Yenile</a>"""awaitasenkron_eposta_gonder(email,"TodoApp Şifre Sıfırlama", html_icerik)return{"mesaj":"Sıfırlama linki e-posta adresinize gönderildi."}
Sistemde sıradan kullanıcılar ile yöneticileri ayırmak için rol tabanlı yetkilendirme (RBAC) kurgulamamız gerekir. Yönetici, tüm kullanıcıların oluşturduğu görevleri (todos) tek bir ekrandan izleyebilmeli ve gerektiğinde müdahale edebilmelidir.
models.py)# models.py içindeki Kullanicilar sınıfına ekleme:classKullanicilar(Base):__tablename__="users"# ...rol= Column(String, default="user")# "user" veya "admin"
admin.py
Backend Mimarisi)FastAPI bağımlılık enjeksiyonunu (Dependency Injection) kullanarak sadece rolü "admin" olanların geçebileceği bir koruma duvarı oluşturuyoruz.
fromfastapiimportAPIRouter,Depends,HTTPException,status,Requestfromfastapi.responsesimportHTMLResponsefromfastapi.templatingimportJinja2Templatesfromsqlalchemy.ormimportSessionfromdatabaseimportget_dbimportmodels# 6. Bölümde yazdığımız mevcut kullanıcıyı doğrulamak için import edinfromauthimportmevcut_kullaniciyi_getirrouter= APIRouter(prefix="/admin", tags=["Yönetim Paneli"])templates= Jinja2Templates(directory="templates")# Admin kontrol fonksiyonuasyncdefadmin_yetki_kontrolu(mevcut_kullanici: dict = Depends(mevcut_kullaniciyi_getir),db: Session = Depends(get_db)):kullanici= db.query(models.Kullanicilar).filter(models.Kullanicilar.id == mevcut_kullanici.get("id")).first()ifnotkullaniciorkullanici.rol !="admin":raiseHTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Bu işlem için yetkiniz bulunmamaktadır.")returnkullanici# TÜM KULLANICILARIN GÖREVLERİNİ LİSTELEYEN ADMIN PANEL SAYFASI@router.get("/dashboard", response_class=HTMLResponse)asyncdefadmin_paneli_goster(request: Request,db: Session = Depends(get_db),admin_kullanici: models.Kullanicilar = Depends(admin_yetki_kontrolu)):# Veri tabanındaki tüm görevleri sahiplik gözetmeksizin çekiyoruztum_gorevler= db.query(models.Görevler).all()# Görevleri ekrana basarken hangi kullanıcının olduğunu göstermek için kullanıcı listesini de çekiyoruztum_kullanicilar= db.query(models.Kullanicilar).all()kullanici_haritasi= {k.id: k.kullanici_adiforkintum_kullanicilar}returntemplates.TemplateResponse("admin_dashboard.html", {"request": request,"gorevler": tum_gorevler,"kullanicilar": kullanici_haritasi,"admin_adi": admin_kullanici.kullanici_adi,"sayfa_basligi":"Sistem Yönetim Paneli"})
templates/admin_dashboard.html)Tüm sistemdeki görevlerin durumunu, kime ait olduğunu gösteren ve yönetimsel aksiyon butonları barındıran şık, tablo tabanlı bir HTML tasarımı.
{% extends"layout.html"%}{% block icerik %}<divclass="admin-panel"><divclass="panel-header"><h2>Yönetici Kontrol Paneli</h2><p>Hoş geldiniz, Sayın <strong>{{ admin_adi }}</strong>. Sistemdeki tüm aktiviteleri buradan takip edebilirsiniz.</p></div><tableclass="admin-table"><thead><tr><th>Görev ID</th><th>Sahibi (Kullanıcı)</th><th>Görev Başlığı</th><th>Açıklama</th><th>Durum</th><th>Yönetimsel İşlem</th></tr></thead><tbody>{% for gorev in gorevler %}<tr><td>#{{ gorev.id }}</td><td><spanclass="user-badge">{{ kullanicilar.get(gorev.sahip_id,'Bilinmeyen Kullanıcı') }}</span></td><td><strong>{{ gorev.baslik }}</strong></td><td>{{ gorev.aciklama }}</td><td>{% if gorev.tamamlandi %}<spanclass="status-badge success">Tamamlandı</span>{% else %}<spanclass="status-badge warning">Bekliyor</span>{% endif %}</td><td><buttonclass="btn-action delete"onclick="sistemdenGorevSil({{ gorev.id }})">Sistemden Sil</button></td></tr>{% else %}<tr><tdcolspan="6"style="text-align: center;">Sistemde henüz kayıtlı görev bulunmamaktadır.</td></tr>{% endfor %}</tbody></table></div><style>.panel-header { background: #1e293b; color: white; padding:1.5rem; border-radius:8px; margin-bottom:2rem; }.admin-table { width:100%; border-collapse: collapse; background: white; border-radius:8px; overflow: hidden; box-shadow:04px6px rgba(0,0,0,0.05); }.admin-table th, .admin-table td { padding:1rem; text-align: left; border-bottom:1px solid #e5e7eb; }.admin-table th { background-color: #4f46e5; color: white; }.user-badge { background: #e0e7ff; color: #4338ca; padding:0.25rem0.5rem; border-radius:4px; font-size:0.85rem; font-weight: bold; }.status-badge { padding:0.25rem0.5rem; border-radius:4px; font-size:0.85rem; font-weight: bold; }.status-badge.success { background: #d1fae5; color: #065f46; }.status-badge.warning { background: #fef3c7; color: #92400e; }.btn-action.delete { background: #ef4444; color: white; border: none; padding:0.5rem1rem; border-radius:4px; cursor: pointer; }.btn-action.delete:hover { background: #dc2626; }</style>{% endblock %}
Uygulamanız büyüdükçe ve kullanıcı sayısı binlere/milyonlara ulaştıkça, veri tabanı tablolarında arama yapmak yavaşlar. PostgreSQL, indeksleme yapılmamış bir tabloda arama yaparken tüm satırları sırayla okur (Sequential Scan). Doğru sütunlara indeks eklemek, sorgu süresini O(N) karmaşıklığından $O(\log N)$ (B-Tree yapısı) seviyesine indirerek performansı yüzlerce kat artırır.
Hangi sütunların indekslenmesi gerektiğini belirlemek için API sorgularımızın davranışlarını inceleriz:
WHERE
kullanıcı_adı = ... veya WHERE
email = ... filtreleri kimlik doğrulamada her
saniye çalışır. Bunlar Unique Index olmalıdır.
WHERE
sahip_id = ... filtresi kullanıcıların
sadece kendi Todo listelerini çekmek için sürekli çalışır. Bu
yabancı anahtara (Foreign Key) mutlaka indeks atılmalıdır.
Görev aramalarında veya
filtrelemelerinde başlık (baslik)
üzerinden arama yapılıyorsa, performans için parçasal veya
metinsel indeks tasarlanmalıdır.
models.py
Güncellemesi)
SQLAlchemy modellerimizde
index=True
tanımını eklediğimizde, ORM tablonun oluşturulma aşamasında
veya Alembic göçünde PostgreSQL mimarisine uygun optimize CREATE
INDEX
komutlarını otomatik olarak arka planda çalıştırır.
fromsqlalchemyimportColumn,Integer,String,Boolean,ForeignKey,IndexfromdatabaseimportBaseclassKullanicilar(Base):__tablename__="users"id= Column(Integer, primary_key=True, index=True)# email ve kullanici_adi alanlarında index=True yapıldığında otomatik B-Tree dizini oluşturulur= Column(String, unique=True, index=True)kullanici_adi= Column(String, unique=True, index=True)hashed_sifre= Column(String)is_active= Column(Boolean, default=True)is_verified= Column(Boolean, default=False)rol= Column(String, default="user")classGörevler(Base):__tablename__="todos"id= Column(Integer, primary_key=True, index=True)baslik= Column(String, index=True)# Başlığa göre aramalarda (Filter) hızı artırıraciklama= Column(String)tamamlandi= Column(Boolean, default=False, index=True)# Sadece tamamlananları veya bekleyenleri listelerken performans sağlarsahip_id= Column(Integer, ForeignKey("users.id"), index=True)# İlişkili tablolarda JOIN ve WHERE performansını uçurur# Çoklu Sütun İndeksi (Composite Index) Tasarımı:# Eğer uygulamanızda sıklıkla "belirli bir kullanıcının sadece tamamlanmış görevlerini getir"# şeklinde ikili filtreleme yapılıyorsa, iki sütunu kapsayan tek bir indeks yazmak en performanslı çözümdür.__table_args__= (Index('idx_sahip_ve_durum','sahip_id','tamamlandi'),)
Eğer canlı sunucudaki (Render PostgreSQL) mevcut tablolara bu indeksleri kod yazmadan doğrudan ham SQL olarak basmak isterseniz, veri tabanı konsolundan (PgAdmin veya Render CLI ekosisteminden) şu komutları koşturabilirsiniz:
-- 1. Kullanıcı İlişki İndeksi (Foreign Key Optimizasyonu)CREATEINDEXIFNOTEXISTSidx_todos_sahip_idONtodos(sahip_id);-- 2. Görev Başlığı İçin Metinsel Arama İndeksiCREATEINDEXIFNOTEXISTSidx_todos_baslikONtodos(baslik);-- 3. Performans Doğrulama (Query Plan Analizi)-- Yazdığınız indeksin PostgreSQL tarafından aktif kullanılıp kullanılmadığını anlamak için sorgunun başına EXPLAIN ANALYZE ekleyin:EXPLAINANALYZESELECT*FROMtodosWHEREsahip_id =5ANDtamamlandi =false;
EXPLAIN
ANALYZE
çıktısında "Index Scan using
idx_sahip_ve_durum ..."
ibaresini görmek, kurguladığınız indeksleme stratejisinin
başarıyla çalıştığını ve disk okuma maliyetlerini sıfıra
yaklaştırdığını kanıtlar.
Bu bölümde, web uygulamamızı siber saldırılara karşı korumalı hale getirmek, dosya yönetimini entegre etmek ve kullanıcı hesaplarını en yüksek güvenlik standardına ulaştırmak için üç ileri seviye backend mekanizması tasarlayacağız:
Medya Yönetimi: Görevlere resim veya doküman eklenebilmesi için dosya yükleme (File Upload) mimarisi.
Hız Sınırlandırma (Rate Limiting): Sunucuyu kaba kuvvet (Brute-Force) ve DDoS türevi bot saldırılarından koruyan mekanizma.
İki Faktörlü Kimlik Doğrulama (2FA): Google Authenticator tabanlı, zamana duyarlı tek kullanımlık şifre (TOTP) doğrulama sistemi.
Kullanıcıların görevlerine
belge veya görsel yükleyebilmesi için FastAPI'nin UploadFile
yapısını kullanacağız. Güvenlik açığı (Remote Code
Execution) oluşmaması adına, gelen dosyaların uzantılarını
kontrol eden bir güvenlik filtresi ekleyecek ve dosyaları sunucu
üzerinde izole bir klasörde depolayacağız.
models.py)# models.py içindeki Görevler sınıfına dosya yolunu tutacak sütunu ekliyoruz:classGörevler(Base):__tablename__="todos"# ... önceki alanlar ...dosya_yolu= Column(String, nullable=True)# Yüklenen dosyanın sunucudaki konumu
main.py)importosimportshutilfromuuidimportuuid4fromfastapiimportAPIRouter,Depends,UploadFile,File,HTTPException,statusfromsqlalchemy.ormimportSessionfromdatabaseimportget_dbimportmodelsfromauthimportmevcut_kullaniciyi_getirrouter= APIRouter(prefix="/todo", tags=["Görev Medya Yönetimi"])# İzin verilen güvenli dosya uzantılarıIZIN_VERILEN_UZANTILAR= {".png",".jpg",".jpeg",".pdf",".docx"}YUKLEME_KLASORU="static/uploads"# Klasör yoksa otomatik oluşturos.makedirs(YUKLEME_KLASORU, exist_ok=True)@router.post("/{todo_id}/dosya-yukle")asyncdefgoreve_dosya_ekle(todo_id: int,file: UploadFile = File(...),db: Session = Depends(get_db),mevcut_kullanici: dict = Depends(mevcut_kullaniciyi_getir)):# 1. Görevin mevcudiyetini ve sahipliğini kontrol etgorev= db.query(models.Görevler).filter(models.Görevler.id == todo_id,models.Görevler.sahip_id == mevcut_kullanici.get("id")).first()ifnotgorev:raiseHTTPException(status_code=404, detail="Görev bulunamadı veya yetkiniz yok.")# 2. Dosya uzantısı güvenlik kontrolüdosya_adi,uzanti= os.path.splitext(file.filename)ifuzanti.lower()notinIZIN_VERILEN_UZANTILAR:raiseHTTPException(status_code=400,detail=f"Geçersiz dosya türü. İzin verilenler:{', '.join(IZIN_VERILEN_UZANTILAR)}")# 3. Dosya adını çakışmaları ve Türkçe karakter sorunlarını önlemek için UUID ile benzersiz yapbenzersiz_dosya_adi=f"{uuid4()}{uzanti.lower()}"hedef_konum= os.path.join(YUKLEME_KLASORU, benzersiz_dosya_adi)# 4. Dosyayı sunucu diskine kaydettry:withopen(hedef_konum,"wb")asbuffer:shutil.copyfileobj(file.file, buffer)finally:awaitfile.close()# 5. Veri tabanına dosya yolunu kaydetgorev.dosya_yolu =f"/static/uploads/{benzersiz_dosya_adi}"db.commit()return{"mesaj":"Dosya başarıyla yüklendi","dosya_yolu": gorev.dosya_yolu}
Kötü niyetli botların API uç noktalarımızı saniyede binlerce istekle yormasını (DDoS) veya şifre deneme saldırısı yapmasını (Brute-Force) engellemek için SlowAPI kütüphanesini entegre edeceğiz. SlowAPI, istemcileri IP adreslerine göre takip ederek belirlenen limit aşıldığında istekleri bloke eder.
pip install slowapimain.py
veya limiter.py)fromfastapiimportFastAPI,RequestfromslowapiimportLimiter,_rate_limit_exceeded_responderfromslowapi.utilimportget_remote_addressfromslowapi.errorsimportRateLimitExceeded# İstek yapan istemcinin IP adresini baz alan bir sınırlayıcı oluşturuyoruzlimiter= Limiter(key_func=get_remote_address)app= FastAPI()# Limit aşımında fırlatılacak istisnayı uygulamaya kaydediyoruzapp.state.limiter = limiterapp.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_responder)# ÖRNEK: Dakikada en fazla 5 istek atılabilen korumalı giriş rotası@app.post("/auth/login")@limiter.limit("5/minute")asyncdefgiris_yap(request: Request):# Bu rota aynı IP adresinden 1 dakika içinde 6. kez çağrılırsa# otomatik olarak HTTP 429 Too Many Requests yanıtı döner.return{"mesaj":"Giriş işlemi başarılı."}# ÖRNEK: Saniyede en fazla 2 istek atılabilen veri çekme rotası@app.get("/todo/tum-listem")@limiter.limit("2/second")asyncdeflistele(request: Request):return{"mesaj":"Görevler listelendi."}
Kullanıcıların şifreleri ele geçirilse bile hesaplarının korunması için Google Authenticator gibi uygulamalarla senkronize çalışan Zamana Duyarlı Tek Kullanımlık Şifre (TOTP) mimarisini kuracağız.
pip install pyotp qrcodemodels.py)Her kullanıcının kendine özel 2FA anahtarını (Secret Key) ve bu özelliğin aktif olup olmadığını tutacak alanları ekliyoruz.
# models.py içindeki Kullanicilar sınıfına ekleme yapın:classKullanicilar(Base):__tablename__="users"# ... önceki alanlar ...two_factor_secret= Column(String, nullable=True)# Kriptografik gizli anahtartwo_factor_enabled= Column(Boolean, default=False)# 2FA aktif mi?
two_factor.py)Kullanıcıya özel anahtar üretme, Google Authenticator uygulamasına taratabilmesi için QR kod şablonu oluşturma ve uygulamadaki 6 haneli kodu doğrulama işlemlerini yürütecek servis yapısı:
importpyotpclassTwoFactorService:@staticmethoddefyeni_anahtar_uret() -> str:# Google Authenticator için rastgele tab32 anahtarı üretirreturnpyotp.random_base32()@staticmethoddefqr_kod_url_uret(kullanici_adi: str,gizli_anahtar: str) -> str:# Kullanıcının Authenticator uygulamasında göreceği başlık şablonureturnpyotp.totp.TOTP(gizli_anahtar).provisioning_uri(name=kullanici_adi,issuer_name="TodoApp Şirketi")@staticmethoddefkodu_dogrula(gizli_anahtar: str,girilen_kod: str) -> bool:# Kullanıcının telefonundaki 6 haneli kodu doğrular (30 saniyelik zaman toleransı içerir)totp= pyotp.totp.TOTP(gizli_anahtar)returntotp.verify(girilen_kod)
auth_2fa.py)Kullanıcının panelinden 2FA'yı aktifleştirmesi ve giriş yaparken 2FA doğrulamasına tabi tutulması süreçleri:
fromfastapiimportAPIRouter,Depends,HTTPException,statusfromsqlalchemy.ormimportSessionfromdatabaseimportget_dbfromauthimportmevcut_kullaniciyi_getirfromtwo_factorimportTwoFactorServiceimportmodelsrouter= APIRouter(prefix="/2fa", tags=["İki Faktörlü Doğrulama"])# 1. 2FA ETKİLEŞTİRME TALEBİ (Kullanıcıya QR kod adresi döner)@router.post("/kurulum-baslat")asyncdefiki_faktorlu_kurulum(mevcut_kullanici: dict = Depends(mevcut_kullaniciyi_getir),db: Session = Depends(get_db)):kullanici= db.query(models.Kullanicilar).filter(models.Kullanicilar.id == mevcut_kullanici.get("id")).first()# Kullanıcıya özel benzersiz anahtar atanırgizli_anahtar= TwoFactorService.yeni_anahtar_uret()kullanici.two_factor_secret = gizli_anahtardb.commit()# Google Authenticator uygulamasının okuyabileceği link üretilirqr_url= TwoFactorService.qr_kod_url_uret(kullanici.kullanici_adi, gizli_anahtar)return{"mesaj":"Telefonunuzdaki Google Authenticator uygulamasından QR kodu taratın veya gizli anahtarı girin.","gizli_anahtar": gizli_anahtar,"qr_kod_baglantisi": qr_url# Arayüzde bu link qrcode kütüphanesi ile görsele dönüştürülür}# 2. 2FA AKTİFLEŞTİRME ONAYI (Kullanıcı kodu doğru girerse sistemde aktif edilir)@router.post("/kurulum-onayla")asyncdefiki_faktorlu_onayla(kod: str,mevcut_kullanici: dict = Depends(mevcut_kullaniciyi_getir),db: Session = Depends(get_db)):kullanici= db.query(models.Kullanicilar).filter(models.Kullanicilar.id == mevcut_kullanici.get("id")).first()ifnotkullanici.two_factor_secret:raiseHTTPException(status_code=400, detail="Önce kurulumu başlatmalısınız.")# Kod kontrolüifTwoFactorService.kodu_dogrula(kullanici.two_factor_secret, kod):kullanici.two_factor_enabled =Truedb.commit()return{"mesaj":"İki faktörlü kimlik doğrulama hesabınızda başarıyla aktif edildi."}raiseHTTPException(status_code=400, detail="Girdiğiniz 6 haneli kod hatalı veya süresi dolmuş.")# 3. GİRİŞ SÜRECİNE ENTEGRASYON (GÜNCELLEME MANTIĞI)# 6. Bölümdeki /token (Giriş) rotasına şu kontrol eklenir:# if kullanici.two_factor_enabled:# return {"status": "2fa_gerekli", "mesaj": "Lütfen 2FA kodunuzu gönderin."}
Bu bölümde, canlı ortamdaki (production) FastAPI uygulamamızı mikrofonksiyonel ve kurumsal düzeyde izlenebilir, hataya dayanıklı (fault-tolerant) ve asenkron işleyişe sahip profesyonel bir ekosisteme dönüştüreceğiz. Üç ana mimariyi sisteme entegre edeceğiz:
Asenkron Kuyruk Yönetimi: Uzun süren işlemleri ana istek-yanıt (request-response) döngüsünden ayıran Celery & Redis mimarisi.
Hata ve Log Yönetimi: Canlı sistemdeki hataları anlık olarak yakalayan Sentry ve yapılandırılmış log üreten Merkezi Günlükleme (Logging) entegrasyonu.
Metrik İzleme (Monitoring): API performansını, istek sayılarını ve yanıt sürelerini ölçüp görselleştiren Prometheus & Grafana altyapısı.
Kullanıcı veri tabanındaki tüm görevlerin özetini içeren bir Excel veya PDF raporu talep ettiğinde, bu işlem binlerce satır veri için birkaç saniye veya dakika sürebilir. İstemciyi HTTP isteğinde bekletmek yerine, işi arka plandaki bir işçi havuzuna (Celery Worker) devrederiz ve mesaj ileticisi (Message Broker) olarak Redis kullanırız.
pip install celery redis pandas openpyxlworker.py)Ana uygulamadan bağımsız çalışacak Celery istemcisini ve uzun süren örnek bir "Rapor Üretme" görevini tasarlıyoruz.
importosimporttimeimportpandasaspdfromceleryimportCelery# Redis bağlantı adresini çevre değişkenlerinden alıyoruzREDIS_URL= os.getenv("REDIS_URL","redis://localhost:6379/0")# Celery uygulamasını tanımlıyoruzcelery_app= Celery("tasks",broker=REDIS_URL,# Görev kuyruğunun tutulacağı yerbackend=REDIS_URL# Görev sonuçlarının saklanacağı yer)# Celery konfigürasyon ayarlarıcelery_app.conf.update(task_serializer="json",result_serializer="json",accept_content=["json"],timezone="Europe/Istanbul",enable_utc=True)@celery_app.task(name="tasks.excel_rapor_uret_task")defexcel_rapor_uret_task(kullanici_id: int,gorevler_listesi: list):"""Bu fonksiyon ana FastAPI thread'ini bloke etmedenarka plandaki Celery Worker üzerinde çalışır."""print(f"[Worker] Kullanıcı{kullanici_id}için rapor üretimi başladı...")# Ağır bir verisel işlem simülasyonu (3 saniye sürüyor)time.sleep(3)# Pandas ile veriyi Excel formatına dönüştürmedf= pd.DataFrame(gorevler_listesi)rapor_klasoru="static/reports"os.makedirs(rapor_klasoru, exist_ok=True)dosya_adi=f"kullanici_{kullanici_id}_rapor_{int(time.time())}.xlsx"hedef_yol= os.path.join(rapor_klasoru, dosya_adi)df.to_excel(hedef_yol, index=False)print(f"[Worker] Rapor başarıyla üretildi:{hedef_yol}")return{"durum":"Tamamlandı","dosya_yolu":f"/static/reports/{dosya_adi}"}
reports_api.py)
Kullanıcı rapor isteği attığında
görevi kuyruğa fırlatıp anında bir görev_id
teslim eden API uç noktası:
fromfastapiimportAPIRouter,Depends,statusfromsqlalchemy.ormimportSessionfromdatabaseimportget_dbfromauthimportmevcut_kullaniciyi_getirimportmodelsfromworkerimportexcel_rapor_uret_taskrouter= APIRouter(prefix="/raporlar", tags=["Asenkron Raporlama"])@router.post("/excel-tetikle", status_code=status.HTTP_202_ACCEPTED)asyncdefrapor_uretme_talebi(mevcut_kullanici: dict = Depends(mevcut_kullaniciyi_getir),db: Session = Depends(get_db)):# 1. Kullanıcının görevlerini veri tabanından çekgorevler= db.query(models.Görevler).filter(models.Görevler.sahip_id == mevcut_kullanici.get("id")).all()# 2. Veriyi Celery'nin serialize edebileceği basit bir liste/sözlük yapısına dönüştürseri_gorevler= [{"ID": g.id,"Başlık": g.baslik,"Açıklama": g.aciklama,"Durum": g.tamamlandi}forgingorevler]# 3. .delay() fonksiyonu ile işi Redis kuyruğuna gönderiyoruz (Anında yanıt döner)kuyruk_gorevi= excel_rapor_uret_task.delay(mevcut_kullanici.get("id"), seri_gorevler)return{"mesaj":"Raporlama işlemi arka planda başlatıldı.","görev_id": kuyruk_gorevi.id,"kontrol_adresi":f"/raporlar/durum/{kuyruk_gorevi.id}"}@router.get("/durum/{gorev_id}")asyncdefrapor_durumu_sorgula(gorev_id: str):# Kuyruğa gönderilen işin bitip bitmediğini kontrol eden uç noktais_sonucu= excel_rapor_uret_task.AsyncResult(gorev_id)ifis_sonucu.state =="PENDING":return{"durum":"Sırada Bekliyor / İşleniyor"}elifis_sonucu.state =="SUCCESS":return{"durum":"Başarılı","sonuc": is_sonucu.result}return{"durum": is_sonucu.state}
Celery Worker'ı ayağa kaldırmak için terminalden çalıştırılacak komut:
celery -A worker.celery_app worker --loglevel=info
Canlı ortamda (production)
kodların hata fırlatması durumunda, sunucu loglarına girip tek
tek satır aramak verimsizdir. Sentry
entegrasyonu sayesinde oluşan her HTTP
500
veya beklenmedik çökme hatası anında Sentry paneline tüm çağrı
yığınıyla (stack trace) birlikte düşer. Aynı zamanda yerel
izleme için Python'ın logging
modülünü yapılandırılmış (structured) formata getireceğiz.
pip install sentry-sdk[fastapi]logger_setup.py)importloggingimportsentry_sdkfromsentry_sdk.integrations.fastapiimportFastAPIIntegration# Sentry Canlı Sunucu Bağlantısı (DSN adresi Sentry panelinden alınır)SENTRY_DSN= os.getenv("SENTRY_DSN","")ifSENTRY_DSN:sentry_sdk.init(dsn=SENTRY_DSN,integrations=[FastAPIIntegration()],traces_sample_rate=1.0,# Performans analizi için tüm işlemleri örnekleprofiles_sample_rate=1.0,)# Standart Python log yapısını özelleştirmelogging.basicConfig(level=logging.INFO,format="%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s",handlers=[logging.FileHandler("app_production.log", encoding="utf-8"),# Logları dosyaya yazarlogging.StreamHandler()# Logları terminale yazar])logger= logging.getLogger("TodoAppLogger")
fromfastapiimportAPIRouter,HTTPExceptionfromlogger_setupimportloggerrouter= APIRouter(prefix="/sistem", tags=["Sistem Yönetimi"])@router.get("/kritik-islem")asyncdefkritik_islem():logger.info("Kritik işlem rotasına bir istek geldi.")try:# Hata simülasyonu: Sıfıra bölme hatasısonuc=10/0exceptZeroDivisionErrorase:# Log dosyasına detay yazılırlogger.error(f"Kritik işlem sırasında beklenmedik hata oluştu:{str(e)}", exc_info=True)# Sentry aktiftir, hata otomatik yakalanıp buluta fırlatılır.raiseHTTPException(status_code=500, detail="İç sunucu hatası oluştu.")
API'mizin sağlığını ölçmek amacıyla; hangi endpoints'e kaç istek geldiğini (Request Count) ve bu isteklerin yanıtlanmasının ne kadar sürdüğünü (Latency) takip etmek için Prometheus biçiminde metrik üreten bir ara katman yazılımı (Middleware) tasarlayacağız.
pip install prometheus-clientmetrics.py)importtimefromfastapiimportFastAPI,Requestfromprometheus_clientimportCounter,Histogram,generate_latest,CONTENT_TYPE_LATESTfromfastapi.responsesimportResponseapp= FastAPI()# Prometheus Metriklerinin Tanımlanması# 1. İstek Sayacı (Hangi metoda, hangi rotaya, hangi HTTP kodu dönmüş?)REQUEST_COUNT= Counter("fastapi_requests_total","Toplam API İstek Sayısı",labelnames=["method","endpoint","http_status"])# 2. İstek Süresi Ölçer (İşlem kaç saniye sürmüş?)REQUEST_LATENCY= Histogram("fastapi_request_duration_seconds","API İstek Yanıt Süresi Dağılımı",labelnames=["method","endpoint"])# Ara katman yazılımı (Middleware): Her isteğin başında ve sonunda çalışır@app.middleware("http")asyncdefprometheus_metrics_middleware(request: Request,call_next):baslangic_zamani= time.time()# İsteğin hangi endpoint'e gittiğini yakalaendpoint= request.url.pathmethod= request.method# İsteyi bir sonraki katmana/endpoint fonksiyonuna iletyanit=awaitcall_next(request)# Süreyi hesapla ve kaydetgecen_sure= time.time() - baslangic_zamanistatus_code= str(yanit.status_code)# Prometheus metriklerini güncelleREQUEST_COUNT.labels(method=method, endpoint=endpoint, http_status=status_code).inc()REQUEST_LATENCY.labels(method=method, endpoint=endpoint).observe(gecen_sure)returnyanit# Prometheus Sunucusunun Verileri Çekeceği (/metrics) Endpoint'i@app.get("/metrics")asyncdefmetrics():# Prometheus kazıyıcısı (Scraper) bu endpoint'i çağırarak ham metrik verilerini toplarreturnResponse(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)
Prometheus Kurulumu
(prometheus.yml):
Prometheus
sunucunuza uygulamanızın metrik endpoint'ini şu şekilde
tanıtırsınız:
scrape_configs:-job_name:'fastapi-app'scrape_interval:5sstatic_configs:-targets: ['localhost:8000']# FastAPI sunucunuzun adresi
Grafana Görselleştirme:
Grafana arayüzüne girip Data Sources kısmından Prometheus'u bağlayın.
Yeni bir Dashboard oluşturup bir grafik (Panel) ekleyin.
Metrik Sorgusu (PromQL):
rate(fastapi_requests_total[1m])
sorgusunu yazarak API'nizin dakika başına aldığı istek
yoğunluk grafiklerini canlı olarak izleyebilirsiniz.
Bu sistemlerin yerel bilgisayarınızda ya da canlı sunucuda (Render, AWS, DigitalOcean vb.) hiçbir "Bende çalışıyordu, sunucuda çalışmıyor" problemi olmadan ayağa kalkabilmesi için en kararlı yöntem Docker ve Docker Compose mimarisidir.
Aşağıda, geliştirdiğimiz tüm bu katmanları (FastAPI, Celery Worker, Redis ve Prometheus) tek bir hamleyle ayağa kaldıracak profesyonel Docker konfigürasyonlarını, çalıştırma komutlarını ve canlı ortamda en sık karşılaşacağınız konfigürasyon hatalarının çözümlerini hazırladım.
Projenizin kök (ana) dizininde aşağıdaki dosyaları sırasıyla oluşturun.
Dockerfile
(Uygulama İmaj Şablonu)Hem FastAPI web sunucumuz hem de Celery işçimiz (worker) aynı kod tabanını kullandığı için tek bir Dockerfile ikisine de hizmet edecektir.
# Hafif ve kararlı Python imajı seçiyoruzFROMpython:3.11-slim# Linux bağımlılıklarını ve PostgreSQL sürücüsü için gerekli paketleri kuruyoruzRUNapt-get update && apt-get install -y \build-essential \libpq-dev \gcc \&& rm -rf /var/lib/apt/lists/*# Çalışma dizini oluşturuyoruzWORKDIR/app# Bağımlılıkları kopyalayıp yüklüyoruzCOPYrequirements.txt .RUNpip install --no-cache-dir -r requirements.txt# Projenin tüm kodlarını ve klasörlerini (static, templates vb.) konteynere kopyalıyoruzCOPY. .# Logların terminale anlık düşmesi için Python çıktı tamponlamasını kapatıyoruzENVPYTHONUNBUFFERED=1
docker-compose.yml
(Çoklu Konteyner Orkestrasyonu)Tüm servislerin birbiriyle izole bir sanal ağ üzerinden konuşmasını ve doğru sıra ile açılmasını sağlayan orkestrasyon dosyası:
version:'3.8'services:# 1. MESAJ İLETİCİSİ (REDIS)redis_broker:image:redis:7-alpinecontainer_name: todo_redisports:-"6379:6379"restart: always# 2. BACKEND API SUNUCUSU (FASTAPI)web_api:build: .container_name: todo_fastapicommand: uvicornmain:app --host0.0.0.0--port8000--reloadports:-"8000:8000"environment:-REDIS_URL=redis://redis_broker:6379/0-DATABASE_URL=sqlite:///./todosapp.db# Canlıda PostgreSQL adresiyle ezilirvolumes:- .:/app# Kodda değişiklik yapıldığında konteynerin anlık güncellenmesi için (Hot-Reload)depends_on:- redis_broker# 3. ARKA PLAN İŞÇİSİ (CELERY WORKER)celery_worker:build: .container_name: todo_celerycommand: celery -A worker.celery_app worker --loglevel=infoenvironment:-REDIS_URL=redis://redis_broker:6379/0volumes:- .:/appdepends_on:- redis_broker- web_api# 4. METRİK TOPLAYICI (PROMETHEUS)prometheus:image:prom/prometheus:latestcontainer_name: todo_prometheusports:-"9090:9090"volumes:- ./prometheus.yml:/etc/prometheus/prometheus.ymldepends_on:- web_api
prometheus.yml
(Prometheus Konfigürasyonu)
Docker ağındaki web_api
servisinin /metrics
uç noktasını kazımak (scrape) için Docker Compose dizinine bu
dosyayı ekleyin:
global:scrape_interval:5sscrape_configs:-job_name:'fastapi-service'metrics_path:'/metrics'static_configs:# Docker Compose içindeki servis ismini direkt domain gibi kullanabiliyoruz-targets: ['web_api:8000']
Gerekli tüm dosyaları hazırladıktan sonra projenizi yerelde ya da sunucuda çalıştırmak için terminalden şu komut zincirini uygulayabilirsiniz:
Sistemi İlk Kez İnşa Etme ve Arka Planda Çalıştırma:
docker-compose up --build -dÇalışan Konteynerlerin Durumunu İzleme:
docker-compose psCelery İşçisinin veya FastAPI'nin Canlı Loglarını Takip Etme:
docker-compose logs -f web_apidocker-compose logs -f celery_worker
Sistemi Tamamen Durdurma (Verileri Kaybetmeden):
docker-compose downCanlı ortama (Render/Docker) geçerken en sık karşılaşılan, saatlerce vakit kaybettirebilecek mimari hatalar ve bunların nokta atışı çözümleri şunlardır:
ConnectionError:
Error 111 connecting to localhost:6379. Connection refused.
(Celery/Redis Hatası)Nedeni:
Celery kodunuzda Redis adresi olarak redis://localhost:6379/0
yazılı kalmıştır. Konteynerler izole çalıştığı için
Celery kendi "localhost" içinde Redis'i arar ve bulamaz.
Kesin Çözümü: Docker
Compose ağında servislerin isimleri birer DNS adresidir.
Kodunuzdaki Redis bağlantısını çevre değişkenine bağlayın
ve docker-compose.yml
içinde redis_broker
(servis adı neyse o) olarak geçin:
# worker.py içi doğrusuREDIS_URL= os.getenv("REDIS_URL","redis://redis_broker:6379/0")
PENDING
Durumunda KalıyorNedeni: FastAPI görevi Redis kuyruğuna başarıyla gönderiyor fakat Celery Worker servisiniz ya çalışmıyor ya da kod tabanındaki fonksiyon isimleri (Task Names) uyuşmuyor.
Kesin Çözümü:
Python'daki klasör yapısından dolayı Celery bazen fonksiyonu
bulamaz. worker.py
içinde task tanımlarken açıkça isim verin:
@celery_app.task(name="tasks.excel_rapor_uret_task")defexcel_rapor_uret_task(...):
FastAPI tarafında çağırırken de tam olarak bu isimle
kaydedildiğinden emin olun. Ayrıca docker-compose
logs todo_celery komutuyla işçinin kuyruğu
dinleyip dinlemediğini loglardan doğrulayın.
OperationalError:
(psycopg2.OperationalError) SSL error: certificate verify failed
(PostgreSQL Canlı Bağlantı Hatası)Nedeni: Render veya bulut tabanlı PostgreSQL veri tabanları dışarıdan gelen bağlantılarda güvenli SSL sertifikası zorunlu tutar. SQLAlchemy varsayılan ayarlarla bağlanmaya çalıştığında el sıkışma (handshake) başarısız olur.
Kesin Çözümü:
database.py
dosyanızdaki bağlantı url kontrolü alanına, eğer canlı
ortamdaysa SSL parametrelerini zorunlu tutacak şu parametreyi
ekleyin:
# database.py güncellenmiş motor kurulumuif"postgresql"inDATABASE_URL:engine= create_engine(DATABASE_URL,connect_args={"sslmode":"require"}# SSL modunu aktif ediyoruz)
127.0.0.1)
Görüp Herkesi Bloke EdiyorNedeni:
Uygulamanızın önüne Nginx, Apache veya Render'ın kendi yük
dengeleyicisi (Load Balancer) geçtiğinde, get_remote_address
fonksiyonu gerçek kullanıcının IP'si yerine proxy sunucusunun
IP'sini yakalar. Bu yüzden tek bir kullanıcı limit doldurduğunda
tüm dünya sistemden engellenir.
Kesin Çözümü:
SlowAPI'ye gerçek kullanıcı IP'sini HTTP başlıklarındaki
(Header) X-Forwarded-For
alanından okumasını söyleyen özel bir fonksiyon yazıp
bağlamalısınız:
# main.py veya limiter.py içi düzeltmedefgercek_kullanici_ip_yakala(request: Request) -> str:# Eğer önünde proxy varsa gerçek IP'yi, yoksa standart IP'yi döndürürforwarded_for= request.headers.get("X-Forwarded-For")ifforwarded_for:returnforwarded_for.split(",")[0].strip()returnrequest.client.host# Limiter'ı bu fonksiyonla başlatınlimiter= Limiter(key_func=gercek_kullanici_ip_yakala)
**********************************************************************************