Wysoce wydajne aplikacje w Pythonie — samouczek FastAPI

Opublikowany: 2022-03-11

Dobre frameworki języka programowania ułatwiają szybsze wytwarzanie produktów wysokiej jakości. Świetne frameworki sprawiają, że całe doświadczenie programistyczne jest przyjemne. FastAPI to nowy framework sieciowy Pythona, który jest potężny i przyjemny w użyciu. Następujące funkcje sprawiają, że FastAPI warto wypróbować:

  • Szybkość: FastAPI to jedna z najszybszych frameworków internetowych w Pythonie. W rzeczywistości jego szybkość dorównuje Node.js i Go. Sprawdź te testy wydajności.
  • Szczegółowe i łatwe w użyciu dokumenty dla programistów
  • Wpisz podpowiedź do swojego kodu i uzyskaj bezpłatną weryfikację i konwersję danych.
  • Łatwe tworzenie wtyczek za pomocą wstrzykiwania zależności.

Tworzenie aplikacji TODO

Aby odkryć wielkie idee stojące za FastAPI, zbudujmy aplikację TODO, która tworzy listy rzeczy do zrobienia dla swoich użytkowników. Nasza mała aplikacja zapewni następujące funkcje:

  • Rejestracja i logowanie
  • Dodaj nowy element TODO
  • Uzyskaj listę wszystkich TODO
  • Usuń/zaktualizuj element TODO

SQLAlchemy dla modeli danych

Nasza aplikacja ma tylko dwa modele: User i TODO. Z pomocą SQLAlchemy, zestawu narzędzi bazodanowych dla Pythona, możemy wyrazić nasze modele w następujący sposób:

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

Gdy nasze modele będą gotowe, napiszmy plik konfiguracyjny dla SQLAlchemy, aby wiedziała, jak nawiązać połączenie z bazą danych.

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

Uwolnij moc wskazówek pisania

Znaczna część każdego projektu API dotyczy rutynowych rzeczy, takich jak walidacja danych i konwersja. Zajmijmy się tym z góry, zanim przejdziemy do pisania obsługi żądań. Dzięki FastAPI wyrażamy schemat naszych danych przychodzących/wychodzących za pomocą modeli pydantycznych, a następnie używamy tych modeli pydantic do wpisywania wskazówek i cieszenia się bezpłatną walidacją i konwersją danych. Należy pamiętać, że te modele nie są związane z naszym przepływem pracy bazy danych i określają tylko kształt danych wpływających i wychodzących z naszego interfejsu REST. Aby napisać pidantyczne modele, zastanów się, w jaki sposób informacje o użytkowniku i TODO będą napływać i wychodzić.

Tradycyjnie nowy użytkownik zarejestruje się w naszej usłudze TODO, a istniejący użytkownik zaloguje się. Obie te interakcje dotyczą informacji o użytkowniku, ale kształt danych będzie inny. Potrzebujemy więcej informacji od użytkowników podczas rejestracji i minimalnych (tylko adres e-mail i hasło) podczas logowania. Oznacza to, że potrzebujemy dwóch modeli pydantycznych, aby wyrazić te dwa różne kształty informacji o użytkowniku.

Jednak w naszej aplikacji TODO wykorzystamy wbudowaną obsługę protokołu OAuth2 w FastAPI do przepływu logowania opartego na tokenach internetowych JSON (JWT). Musimy tylko zdefiniować tutaj schemat UserCreate , aby określić dane, które wpłyną do naszego punktu końcowego rejestracji, oraz schemat UserBase , który zostanie zwrócony jako odpowiedź w przypadku pomyślnego zakończenia procesu rejestracji.

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

Tutaj oznaczyliśmy nazwisko, imię i hasło jako ciąg znaków, ale można je dodatkowo zaostrzyć, używając pydantycznych ograniczonych ciągów, które umożliwiają sprawdzanie, takich jak minimalna długość, maksymalna długość i wyrażenia regularne.

Aby wesprzeć tworzenie i wyświetlanie pozycji TODO, definiujemy następujący schemat:

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

Aby wesprzeć aktualizację istniejącego elementu TODO, definiujemy inny schemat:

 class TODOUpdate(TODOCreate): id: int

Dzięki temu kończymy definiowanie schematów dla wszystkich wymian danych. Teraz zwracamy naszą uwagę na programy obsługi żądań, w których te schematy będą używane do wykonywania wszystkich ciężkich zadań związanych z konwersją i walidacją danych za darmo.

Pozwól użytkownikom się zarejestrować

Najpierw zezwólmy użytkownikom na rejestrację, ponieważ wszystkie nasze usługi muszą być dostępne dla uwierzytelnionego użytkownika. Piszemy nasz pierwszy program obsługi żądań, korzystając ze schematu UserCreate i UserBase zdefiniowanego powyżej.

 @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

W tym krótkim kodzie dużo się dzieje. Użyliśmy dekoratora do określenia czasownika HTTP, identyfikatora URI i schematu pomyślnych odpowiedzi. Aby upewnić się, że użytkownik przesłał właściwe dane, w treści żądania wpisaliśmy podpowiedź ze zdefiniowanym wcześniej schematem UserCreate . Metoda definiuje inny parametr do uzyskania dojścia do bazy danych — jest to wstrzykiwanie zależności w akcji i jest omówione w dalszej części tego samouczka.

Zabezpieczanie naszego API

W naszej aplikacji chcemy mieć następujące funkcje bezpieczeństwa:

  • Haszowanie hasła
  • Uwierzytelnianie oparte na JWT

Do haszowania haseł możemy użyć Passlib. Zdefiniujmy funkcje obsługujące haszowanie hasła i sprawdzanie poprawności hasła.

 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

Aby włączyć uwierzytelnianie oparte na JWT, musimy wygenerować tokeny JWT, a także je zdekodować, aby uzyskać poświadczenia użytkownika. Aby zapewnić tę funkcjonalność, definiujemy następujące funkcje.

 # 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

Wydaj tokeny przy pomyślnym logowaniu

Teraz zdefiniujemy punkt końcowy logowania i zaimplementujemy przepływ hasła OAuth2. Ten punkt końcowy otrzyma wiadomość e-mail i hasło. Sprawdzimy poświadczenia w bazie danych i po pomyślnym zakończeniu wystawimy użytkownikowi token sieciowy JSON.

Aby otrzymać dane uwierzytelniające, użyjemy OAuth2PasswordRequestForm , który jest częścią narzędzi bezpieczeństwa 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"}

Korzystanie z wstrzykiwania zależności w celu uzyskania dostępu do bazy danych i ochrony punktów końcowych

Skonfigurowaliśmy punkt końcowy logowania, który udostępnia użytkownikowi token JWT po pomyślnym zalogowaniu. Użytkownik może zapisać ten token w magazynie lokalnym i pokazać go w naszym zapleczu jako nagłówek autoryzacji. Punkty końcowe, które oczekują dostępu tylko od zalogowanych użytkowników, mogą zdekodować token i dowiedzieć się, kto jest zleceniodawcą. Ten rodzaj pracy nie jest powiązany z konkretnym punktem końcowym, a raczej jest to wspólna logika wykorzystywana we wszystkich chronionych punktach końcowych. Najlepiej skonfigurować logikę dekodowania tokenów jako zależność, której można użyć w dowolnej obsłudze żądań.

W FastAPI-speak nasze funkcje operacji na ścieżce (programy obsługi żądań) będą wtedy zależeć od get_current_user . Zależność get_current_user musi mieć połączenie z bazą danych i podłączyć się do logiki OAuth2PasswordBearer , aby uzyskać token. Rozwiążemy ten problem, uzależniając get_current_user od innych funkcji. W ten sposób możemy zdefiniować łańcuchy zależności, co jest bardzo potężną koncepcją.

 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

Zalogowani użytkownicy mogą CRUD TODOs

Zanim napiszemy funkcje operacji na ścieżce dla TODO Create, Read, Update, Delete (CRUD), zdefiniujemy następujące funkcje pomocnicze do wykonania rzeczywistego CRUD na 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()

Te funkcje na poziomie bazy danych będą używane w następujących punktach końcowych 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"}

Napisz testy

Napiszmy kilka testów dla naszego API TODO. FastAPI udostępnia klasę TestClient opartą na popularnej bibliotece Requests, a testy możemy uruchamiać za pomocą Pytest.

Aby upewnić się, że tylko zalogowani użytkownicy mogą tworzyć TODO, możemy napisać coś takiego:

 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

Poniższy test sprawdza nasz punkt końcowy logowania i generuje token JWT, jeśli zostanie przedstawiony z prawidłowymi poświadczeniami logowania.

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

Podsumowanie

Zakończyliśmy implementację bardzo prostej aplikacji TODO przy użyciu FastAPI. Do tej pory widziałeś, jak dobrze można wykorzystać wskazówki dotyczące typów w definiowaniu kształtu danych przychodzących i wychodzących za pośrednictwem naszego interfejsu REST. Definiujemy schematy w jednym miejscu i pozostawiamy FastAPI zastosowanie walidacji i konwersji danych. Inną godną uwagi funkcją jest wstrzykiwanie zależności. Wykorzystaliśmy tę koncepcję do spakowania współdzielonej logiki uzyskiwania połączenia z bazą danych, dekodowania JWT w celu uzyskania aktualnie zalogowanego użytkownika i implementacji prostego OAuth2 z hasłem i nośnikiem. Zobaczyliśmy też, jak można łączyć zależności.

Możemy łatwo zastosować tę koncepcję, aby dodać funkcje, takie jak dostęp oparty na rolach. Poza tym piszemy zwięzły i potężny kod bez poznawania specyfiki frameworka. Krótko mówiąc, FastAPI to zbiór potężnych narzędzi, których nie musisz się uczyć, ponieważ są po prostu nowoczesnym Pythonem. Baw się dobrze.