Python ile Yüksek Performanslı Uygulamalar – Bir FastAPI Eğitimi

İyi programlama dili çerçeveleri, kaliteli ürünleri daha hızlı üretmeyi kolaylaştırır. Harika çerçeveler, tüm geliştirme deneyimini bile eğlenceli hale getirir. FastAPI, güçlü ve kullanımı keyifli yeni bir Python web çerçevesidir. Aşağıdaki özellikler FastAPI'yi denemeye değer kılar:

• Hız: FastAPI, en hızlı Python web çerçevelerinden biridir. Aslında, hızı Node.js ve Go ile eşittir. Bu performans testlerini kontrol edin.
• Ayrıntılı ve kullanımı kolay geliştirici belgeleri
• İpucu yazın ve ücretsiz veri doğrulama ve dönüştürme elde edin.
• Bağımlılık enjeksiyonunu kullanarak kolayca eklentiler oluşturun.

TODO Uygulaması Oluşturma

FastAPI'nin arkasındaki büyük fikirleri keşfetmek için, kullanıcıları için yapılacaklar listeleri oluşturan bir TODO uygulaması oluşturalım. Minik uygulamamız aşağıdaki özellikleri sağlayacaktır:

• Kaydol ve Giriş Yap
• Yeni YAPILACAKLAR öğesi ekle
• Tüm TODO'ların bir listesini alın
• YAPILACAKLAR öğesini silme/güncelleme

Veri Modelleri için SQLAlchemy

Uygulamamızın sadece iki modeli var: Kullanıcı ve YAPILACAKLAR. Python için veritabanı araç takımı olan SQLAlchemy'nin yardımıyla modellerimizi şu şekilde ifade edebiliriz:

 class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) lname = Column(String) fname = Column(String) email = Column(String, unique=True, index=True) todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan") class TODO(Base): __tablename__ = "todos" id = Column(Integer, primary_key=True, index=True) text = Column(String, index=True) completed = Column(Boolean, default=False) owner_id = Column(Integer, ForeignKey("users.id")) owner = relationship("User", back_populates="todos")

Modellerimiz hazır olduğunda SQLAlchemy için konfigürasyon dosyasını yazalım ki veritabanı ile nasıl bağlantı kurulacağını bilsin.

 import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL'] engine = create_engine( SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

Tip İpuçlarının Gücünü Ortaya Çıkarın

Herhangi bir API projesinin büyük bir kısmı, veri doğrulama ve dönüştürme gibi rutin şeylerle ilgilidir. İstek işleyicilerini yazmaya geçmeden önce bunu baştan ele alalım. FastAPI ile, pydantic modelleri kullanarak gelen/giden verilerimizin şemasını ifade ediyoruz ve ardından bu pydantic modelleri ipucu yazmak ve ücretsiz veri doğrulama ve dönüştürmenin keyfini çıkarmak için kullanıyoruz. Lütfen bu modellerin veritabanı iş akışımızla ilgili olmadığını ve yalnızca REST arayüzümüze giren ve çıkan verilerin şeklini belirttiğini unutmayın. Pydantik modeller yazmak için, Kullanıcı ve TODO bilgilerinin içeri ve dışarı akacağı tüm yolları düşünün.

Geleneksel olarak, TODO hizmetimize yeni bir kullanıcı kaydolur ve mevcut bir kullanıcı oturum açar. Bu etkileşimlerin her ikisi de Kullanıcı bilgileriyle ilgilenir, ancak verilerin şekli farklı olacaktır. Kayıt sırasında kullanıcılardan daha fazla bilgiye ve oturum açarken minimum (yalnızca e-posta ve şifre) bilgilere ihtiyacımız var. Bu, Kullanıcı bilgilerinin bu iki farklı şeklini ifade etmek için iki pydantic modele ihtiyacımız olduğu anlamına geliyor.

Ancak TODO uygulamamızda, JSON Web Belirteçleri (JWT) tabanlı oturum açma akışı için FastAPI'deki yerleşik OAuth2 desteğinden yararlanacağız. Kayıt bitiş noktamıza akacak verileri belirtmek için burada bir UserCreate şeması ve kayıt işleminin başarılı olması durumunda yanıt olarak geri dönecek bir UserBase şeması tanımlamamız yeterlidir.

 from pydantic import BaseModel from pydantic import EmailStr class UserBase(BaseModel): email: EmailStr class UserCreate(UserBase): lname: str fname: str password: str

Burada, soyadı, adı ve parolayı bir dize olarak işaretledik, ancak minimum uzunluk, maksimum uzunluk ve normal ifadeler gibi kontrolleri etkinleştiren pydantic kısıtlı dizeler kullanılarak daha da sıkılaştırılabilir.

TODO öğelerinin oluşturulmasını ve listelenmesini desteklemek için aşağıdaki şemayı tanımlarız:

 class TODOCreate(BaseModel): text: str completed: bool

Mevcut bir TODO öğesinin güncellenmesini desteklemek için başka bir şema tanımlıyoruz:

 class TODOUpdate(TODOCreate): id: int

Bununla, tüm veri alışverişleri için şema tanımlamayı bitirdik. Şimdi dikkatimizi, veri dönüştürme ve doğrulamanın tüm ağır yükünü ücretsiz olarak yapmak için bu şemaların kullanılacağı istek işleyicilerine çeviriyoruz.

Kullanıcıların Kaydolmasına İzin Verin

Öncelikle, tüm hizmetlerimize kimliği doğrulanmış bir kullanıcı tarafından erişilmesi gerektiğinden, kullanıcıların kaydolmasına izin verelim. İlk istek işleyicimizi yukarıda tanımlanan UserCreate ve UserBase şemasını kullanarak yazıyoruz.

 @app.post("/api/users", response_model=schemas.User) def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)): """add new user""" user = crud.get_user_by_email(db, user_data.email) if user: raise HTTPException(status_code=409, detail="Email already registered.") signedup_user = crud.create_user(db, user_data) return signedup_user

Bu kısa kod parçasında çok şey oluyor. HTTP fiilini, URI'yi ve başarılı yanıtların şemasını belirtmek için bir dekoratör kullandık. Kullanıcının doğru verileri gönderdiğinden emin olmak için daha önce tanımlanmış bir UserCreate şemasıyla istek gövdesine ipucu yazdık. Yöntem, veritabanı üzerinde bir tanıtıcı almak için başka bir parametre tanımlar; bu, eylem halindeki bağımlılık enjeksiyonudur ve bu öğreticide daha sonra tartışılacaktır.

API'mizin Güvenliğini Sağlama

Uygulamamızda aşağıdaki güvenlik özelliklerini istiyoruz:

• Şifre karma
• JWT tabanlı kimlik doğrulama

Parola hash için Passlib kullanabiliriz. Parola karmasını işleyen ve bir parolanın doğru olup olmadığını kontrol eden işlevleri tanımlayalım.

 from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def authenticate_user(db, email: str, password: str): user = crud.get_user_by_email(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user

JWT tabanlı kimlik doğrulamayı etkinleştirmek için, kullanıcı kimlik bilgilerini almak için JWT'ler oluşturmamız ve bunların kodunu çözmemiz gerekir. Bu işlevselliği sağlamak için aşağıdaki işlevleri tanımlıyoruz.

 # install PyJWT import jwt from fastapi.security import OAuth2PasswordBearer SECRET_KEY = os.environ['SECRET_KEY'] ALGORITHM = os.environ['ALGORITHM'] def create_access_token(*, data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def decode_access_token(db, token): credentials_exception = HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get("sub") if email is None: raise credentials_exception token_data = schemas.TokenData(email=email) except PyJWTError: raise credentials_exception user = crud.get_user_by_email(db, email=token_data.email) if user is None: raise credentials_exception return user

Başarılı Girişte Jeton Ver

Şimdi bir Login uç noktası tanımlayacağız ve OAuth2 şifre akışını uygulayacağız. Bu uç nokta bir e-posta ve şifre alacaktır. Kimlik bilgilerini veritabanına göre kontrol edeceğiz ve başarılı olursa kullanıcıya bir JSON web belirteci vereceğiz.

Kimlik bilgilerini almak için FastAPI'nin güvenlik yardımcı programlarının bir parçası olan OAuth2PasswordRequestForm kullanacağız.

 @app.post("/api/token", response_model=schemas.Token) def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()): """generate access token for valid credentials""" user = authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token(data={"sub": user.email}, expires_delta=access_token_expires) return {"access_token": access_token, "token_type": "bearer"}

DB'ye Erişmek ve Uç Noktaları Korumak için Dependency Injection'ı Kullanma

Başarılı bir oturum açtıktan sonra bir kullanıcıya JWT sağlayan oturum açma uç noktasını kurduk. Kullanıcı bu belirteci yerel depolamaya kaydedebilir ve onu bir Yetkilendirme başlığı olarak arka uçumuza gösterebilir. Yalnızca oturum açmış kullanıcılardan erişim bekleyen uç noktalar, belirtecin kodunu çözebilir ve istekte bulunanın kim olduğunu öğrenebilir. Bu tür bir çalışma belirli bir uç noktaya bağlı değildir, bunun yerine korunan tüm uç noktalarda kullanılan paylaşılan mantıktır. Belirteç kod çözme mantığını herhangi bir istek işleyicide kullanılabilecek bir bağımlılık olarak ayarlamak en iyisidir.

FastAPI konuşmasında, yol işlemi işlevlerimiz (istek işleyicileri) get_current_user öğesine bağlı olacaktır. get_current_user bağımlılığının veritabanıyla bir bağlantısı olması ve bir belirteç elde etmek için FastAPI'nin OAuth2PasswordBearer mantığına bağlanması gerekir. Bu sorunu get_current_user diğer fonksiyonlara bağımlı hale getirerek çözeceğiz. Bu şekilde çok güçlü bir kavram olan bağımlılık zincirlerini tanımlayabiliriz.

 def get_db(): """provide db session to path operation functions""" try: db = SessionLocal() yield db finally: db.close() def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): return decode_access_token(db, token) @app.get("/api/me", response_model=schemas.User) def read_logged_in_user(current_user: models.User = Depends(get_current_user)): """return user settings for current user""" return current_user

Oturum Açmış Kullanıcılar TODO'ları CRUD Yapabilir

TODO Create, Read, Update, Delete (CRUD) için path operasyon fonksiyonlarını yazmadan önce, db üzerinde gerçek CRUD gerçekleştirmek için aşağıdaki yardımcı fonksiyonları tanımlıyoruz.

 def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate): todo = models.TODO(text=todo_data.text, completed=todo_data.completed) todo.owner = current_user db.add(todo) db.commit() db.refresh(todo) return todo def update_todo(db: Session, todo_data: schemas.TODOUpdate): todo = db.query(models.TODO).filter(models.TODO.id == id).first() todo.text = todo_data.text todo.completed = todo.completed db.commit() db.refresh(todo) return todo def delete_todo(db: Session, id: int): todo = db.query(models.TODO).filter(models.TODO.id == id).first() db.delete(todo) db.commit() def get_user_todos(db: Session, userid: int): return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

Bu db düzeyi işlevler, aşağıdaki REST uç noktalarında kullanılacaktır:

 @app.get("/api/mytodos", response_model=List[schemas.TODO]) def get_own_todos(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """return a list of TODOs owned by current user""" todos = crud.get_user_todos(db, current_user.id) return todos @app.post("/api/todos", response_model=schemas.TODO) def add_a_todo(todo_data: schemas.TODOCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """add a TODO""" todo = crud.create_meal(db, current_user, meal_data) return todo @app.put("/api/todos/{todo_id}", response_model=schemas.TODO) def update_a_todo(todo_id: int, todo_data: schemas.TODOUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """update and return TODO for given id""" todo = crud.get_todo(db, todo_id) updated_todo = crud.update_todo(db, todo_id, todo_data) return updated_todo @app.delete("/api/todos/{todo_id}") def delete_a_meal(todo_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """delete TODO of given id""" crud.delete_meal(db, todo_id) return {"detail": "TODO Deleted"}

Test Yaz

TODO API'miz için birkaç test yazalım. FastAPI, popüler İstekler kitaplığına dayalı bir TestClient sınıfı sağlar ve testleri Pytest ile çalıştırabiliriz.

Yalnızca oturum açmış kullanıcıların YAPILACAKLAR oluşturabileceğinden emin olmak için şöyle bir şey yazabiliriz:

 from starlette.testclient import TestClient from .main import app client = TestClient(app) def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False) response = client.post("/api/todos", data=todo) assert response.status_code == 401

Aşağıdaki test, oturum açma uç noktamızı kontrol eder ve geçerli oturum açma kimlik bilgileriyle sunulursa bir JWT oluşturur.

 def test_user_can_obtain_auth_token(): response = client.post("/api/token", data=good_credentials) assert response.status_code == 200 assert 'access_token' in response.json() assert 'token_type' in response.json()

Özetlemek

FastAPI kullanarak çok basit bir YAPILACAKLAR uygulamasını uygulamayı bitirdik. Şimdiye kadar, REST arayüzümüz aracılığıyla gelen ve giden verilerin şeklini tanımlamada iyi bir şekilde kullanılan tür ipuçlarının gücünü gördünüz. Şemaları tek bir yerde tanımlıyoruz ve veri doğrulama ve dönüştürmeyi uygulamak için FastAPI'ye bırakıyoruz. Dikkate değer diğer bir özellik de bağımlılık enjeksiyonudur. Bu konsepti, bir veritabanı bağlantısı elde etmenin, şu anda oturum açmış olan kullanıcıyı almak için JWT'nin kodunu çözmenin ve parola ve taşıyıcı ile basit OAuth2'nin uygulanmasının paylaşılan mantığını paketlemek için kullandık. Ayrıca bağımlılıkların nasıl zincirlenebileceğini de gördük.

Rol tabanlı erişim gibi özellikler eklemek için bu konsepti kolayca uygulayabiliriz. Ayrıca, bir çerçevenin özelliklerini öğrenmeden kısa ve güçlü kod yazıyoruz. Basit bir deyişle, FastAPI, yalnızca modern Python oldukları için öğrenmeniz gerekmeyen güçlü araçlardan oluşan bir koleksiyondur. İyi eğlenceler.