Aplicații de înaltă performanță cu Python – Un tutorial FastAPI

Publicat: 2022-03-11

Cadrele bune de limbaj de programare facilitează producerea mai rapidă a produselor de calitate. Cadrele grozave chiar fac ca întreaga experiență de dezvoltare să fie plăcută. FastAPI este un nou cadru web Python care este puternic și plăcut de utilizat. Următoarele caracteristici fac ca FastAPI să merite încercat:

  • Viteză: FastAPI este unul dintre cele mai rapide cadre web Python. De fapt, viteza sa este la egalitate cu Node.js și Go. Verificați aceste teste de performanță.
  • Documente pentru dezvoltatori detaliate și ușor de utilizat
  • Introduceți indiciu codul și obțineți validarea și conversia datelor gratuite.
  • Creați cu ușurință pluginuri folosind injecția de dependență.

Crearea unei aplicații TODO

Pentru a explora ideile mari din spatele FastAPI, să construim o aplicație TODO, care creează liste de activități pentru utilizatorii săi. Mica noastră aplicație va oferi următoarele caracteristici:

  • Înregistrare și autentificare
  • Adăugați un nou articol TODO
  • Obțineți o listă cu toate TODO
  • Șterge/Actualizează un articol TODO

SQLAlchemy pentru modele de date

Aplicația noastră are doar două modele: User și TODO. Cu ajutorul SQLAlchemy, setul de instrumente pentru baze de date pentru Python, ne putem exprima modelele astfel:

 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")

Odată ce modelele noastre sunt gata, să scriem fișierul de configurare pentru SQLAlchemy, astfel încât acesta să știe cum să stabilească o conexiune cu baza de date.

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

Dezlănțuiți puterea sugestiilor de tip

O parte considerabilă a oricărui proiect API se referă la chestii de rutină, cum ar fi validarea și conversia datelor. Să o abordăm înainte de a trece la scrierea de gestionare a cererilor. Cu FastAPI, exprimăm schema datelor noastre de intrare/ieșire folosind modele pidantice și apoi folosim aceste modele pidantice pentru a introduce indiciu și pentru a ne bucura de validarea și conversia gratuită a datelor. Vă rugăm să rețineți că aceste modele nu sunt legate de fluxul de lucru al bazei de date și specifică doar forma datelor care curg în și din interfața noastră REST. Pentru a scrie modele pidantice, gândiți-vă la toate modurile în care informațiile despre utilizator și TODO vor intra și ieși.

În mod tradițional, un utilizator nou se va înscrie la serviciul nostru TODO și un utilizator existent se va autentifica. Ambele interacțiuni se referă la informațiile despre utilizator, dar forma datelor va fi diferită. Avem nevoie de mai multe informații de la utilizatori în timpul înscrierii și minime (doar e-mail și parolă) atunci când ne conectăm. Aceasta înseamnă că avem nevoie de două modele pydantice pentru a exprima aceste două forme diferite de informații despre utilizator.

Cu toate acestea, în aplicația noastră TODO, vom folosi suportul OAuth2 încorporat în FastAPI pentru un flux de conectare bazat pe JSON Web Tokens (JWT). Trebuie doar să definim aici o schemă UserCreate pentru a specifica datele care vor curge în punctul final de înregistrare și o schemă UserBase care va fi returnată ca răspuns în cazul în care procesul de înregistrare are succes.

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

Aici, am marcat numele de familie, prenumele și parola ca șir, dar poate fi întărit și mai mult prin utilizarea șirurilor de caractere constrânse pidantice care permit verificări precum lungimea minimă, lungimea maximă și regexe.

Pentru a sprijini crearea și listarea articolelor TODO, definim următoarea schemă:

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

Pentru a sprijini actualizarea unui articol TODO existent, definim o altă schemă:

 class TODOUpdate(TODOCreate): id: int

Cu aceasta, am terminat cu definirea schemelor pentru toate schimburile de date. Acum ne îndreptăm atenția către gestionatorii de solicitări, unde aceste scheme vor fi folosite pentru a face toate sarcinile grele de conversie și validare a datelor în mod gratuit.

Permiteți utilizatorilor să se înregistreze

În primul rând, să permitem utilizatorilor să se înscrie, deoarece toate serviciile noastre trebuie să fie accesate de un utilizator autentificat. Scriem primul nostru handler de solicitări folosind schema UserCreate și UserBase definită mai sus.

 @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

Se întâmplă multe în această scurtă bucată de cod. Am folosit un decorator pentru a specifica verbul HTTP, URI-ul și schema răspunsurilor de succes. Pentru a ne asigura că utilizatorul a trimis datele corecte, am introdus indiciu în corpul cererii cu o schemă UserCreate definită mai devreme. Metoda definește un alt parametru pentru obținerea unui handle în baza de date - aceasta este injecția de dependență în acțiune și este discutată mai târziu în acest tutorial.

Securizarea API-ului nostru

Dorim următoarele caracteristici de securitate în aplicația noastră:

  • Hashing parole
  • Autentificare bazată pe JWT

Pentru hashing parole, putem folosi Passlib. Să definim funcții care se ocupă de hashingul parolei și de a verifica dacă o parolă este corectă.

 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

Pentru a activa autentificarea bazată pe JWT, trebuie să generăm JWT-uri și să le decodăm pentru a obține acreditările utilizatorului. Definim următoarele funcții pentru a oferi această funcționalitate.

 # 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

Emiteți jetoane la conectarea cu succes

Acum, vom defini un punct final de conectare și vom implementa fluxul de parole OAuth2. Acest punct final va primi un e-mail și o parolă. Vom verifica acreditările în baza de date și, în caz de succes, vom emite utilizatorului un token web JSON.

Pentru a primi acreditările, vom folosi OAuth2PasswordRequestForm , care face parte din utilitățile de securitate FastAPI.

 @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"}

Utilizarea Dependency Injection pentru a accesa DB și a proteja punctele finale

Am configurat punctul final de conectare care oferă un JWT unui utilizator după conectarea cu succes. Utilizatorul poate salva acest token în stocarea locală și îl poate afișa back-end-ului nostru ca antet de autorizare. Punctele finale care așteaptă acces numai de la utilizatorii conectați pot decoda simbolul și pot afla cine este solicitantul. Acest tip de lucru nu este legat de un anumit punct final, ci mai degrabă este o logică comună utilizată în toate punctele finale protejate. Cel mai bine este să configurați logica de decodare a simbolurilor ca o dependență care poate fi utilizată în orice handler de solicitare.

În limbajul FastAPI, funcțiile noastre de operare a căii (solicitanți) ar depinde apoi de get_current_user . Dependența get_current_user trebuie să aibă o conexiune la baza de date și să se conecteze la logica OAuth2PasswordBearer a OAuth2PasswordBearer pentru a obține un token. Vom rezolva această problemă făcând ca get_current_user să depindă de alte funcții. În acest fel, putem defini lanțuri de dependență, care este un concept foarte puternic.

 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

Utilizatorii autentificati pot CRUD TODO

Înainte de a scrie funcțiile de operare a căii pentru TODO Create, Read, Update, Delete (CRUD), definim următoarele funcții de ajutor pentru a efectua CRUD real pe db.

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

Aceste funcții la nivel db vor fi utilizate în următoarele puncte finale REST:

 @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"}

Scrie teste

Să scriem câteva teste pentru API-ul nostru TODO. FastAPI oferă o clasă TestClient care se bazează pe populara bibliotecă Requests și putem rula testele cu Pytest.

Pentru a ne asigura că numai utilizatorii autentificați pot crea o TODO, putem scrie ceva de genul acesta:

 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

Următorul test verifică punctul nostru de conectare și generează un JWT dacă este prezentat cu acreditări de conectare valide.

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

Rezumând

Am terminat de implementat o aplicație TODO foarte simplă folosind FastAPI. Până acum, ați văzut puterea indicii de tip folosită în mod adecvat în definirea formei datelor de intrare și de ieșire prin interfața noastră REST. Definim schemele într-un singur loc și lăsăm pe FastAPI să aplice validarea și conversia datelor. Cealaltă caracteristică demnă de remarcat este injecția de dependență. Am folosit acest concept pentru a împacheta logica partajată a obținerii unei conexiuni la baza de date, decodând JWT pentru a obține utilizatorul conectat în prezent și implementând OAuth2 simplu cu parolă și purtător. De asemenea, am văzut cum dependențele pot fi înlănțuite.

Putem aplica cu ușurință acest concept pentru a adăuga funcții precum accesul bazat pe roluri. În plus, scriem cod concis și puternic, fără a învăța particularitățile unui cadru. Cu cuvinte simple, FastAPI este o colecție de instrumente puternice pe care nu trebuie să le înveți, deoarece sunt doar Python modern. A se distra.