Leistungsstarke Apps mit Python – Ein FastAPI-Tutorial

Veröffentlicht: 2022-03-11

Gute Programmiersprachen-Frameworks machen es einfach, Qualitätsprodukte schneller zu produzieren. Großartige Frameworks machen sogar die gesamte Entwicklungserfahrung angenehm. FastAPI ist ein neues Python-Webframework, das leistungsstark und angenehm zu verwenden ist. Die folgenden Funktionen machen FastAPI einen Versuch wert:

  • Geschwindigkeit: FastAPI ist eines der schnellsten Python-Webframeworks. Tatsächlich ist seine Geschwindigkeit mit Node.js und Go vergleichbar. Überprüfen Sie diese Leistungstests.
  • Detaillierte und benutzerfreundliche Entwicklerdokumentation
  • Geben Sie Ihren Code hint ein und erhalten Sie eine kostenlose Datenvalidierung und -konvertierung.
  • Erstellen Sie Plugins einfach mit Abhängigkeitsinjektion.

Erstellen einer TODO-App

Um die großen Ideen hinter FastAPI zu erkunden, bauen wir eine TODO-App, die Aufgabenlisten für ihre Benutzer erstellt. Unsere kleine App bietet die folgenden Funktionen:

  • Anmeldung und Login
  • Neues TODO-Element hinzufügen
  • Holen Sie sich eine Liste aller TODOs
  • Ein TODO-Element löschen/aktualisieren

SQLAlchemy für Datenmodelle

Unsere App hat nur zwei Modelle: User und TODO. Mit Hilfe von SQLAlchemy, dem Datenbank-Toolkit für Python, können wir unsere Modelle wie folgt ausdrücken:

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

Sobald unsere Modelle fertig sind, schreiben wir die Konfigurationsdatei für SQLAlchemy, damit es weiß, wie es eine Verbindung mit der Datenbank herstellt.

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

Entfesseln Sie die Kraft der Typhinweise

Ein beträchtlicher Teil jedes API-Projekts betrifft Routineaufgaben wie Datenvalidierung und -konvertierung. Lassen Sie uns das im Vorfeld angehen, bevor wir mit dem Schreiben von Request-Handlern fortfahren. Mit FastAPI drücken wir das Schema unserer eingehenden/ausgehenden Daten mit pydantischen Modellen aus und verwenden diese pydantischen Modelle dann, um Hinweise einzugeben und eine kostenlose Datenvalidierung und -konvertierung zu genießen. Bitte beachten Sie, dass diese Modelle nicht mit unserem Datenbank-Workflow zusammenhängen und nur die Form der Daten angeben, die in unsere REST-Schnittstelle ein- und ausfließen. Um pydantische Modelle zu schreiben, denken Sie darüber nach, wie die Benutzer- und TODO-Informationen ein- und ausfließen.

Traditionell meldet sich ein neuer Benutzer für unseren TODO-Service an und ein bestehender Benutzer meldet sich an. Bei beiden Interaktionen geht es um Benutzerinformationen, aber die Form der Daten ist unterschiedlich. Wir benötigen mehr Informationen von Benutzern während der Anmeldung und minimal (nur E-Mail und Passwort) beim Anmelden. Das bedeutet, dass wir zwei pydantische Modelle benötigen, um diese beiden unterschiedlichen Formen von Benutzerinformationen auszudrücken.

In unserer TODO-App werden wir jedoch die integrierte OAuth2-Unterstützung in FastAPI für einen auf JSON Web Tokens (JWT) basierenden Anmeldefluss nutzen. Wir müssen hier nur ein UserCreate Schema definieren, um Daten anzugeben, die in unseren Anmeldeendpunkt fließen, und ein UserBase Schema, das als Antwort zurückgegeben wird, falls der Anmeldevorgang erfolgreich ist.

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

Hier haben wir Nachname, Vorname und Passwort als Zeichenfolge markiert, aber es kann weiter gestrafft werden, indem pydantische eingeschränkte Zeichenfolgen verwendet werden, die Prüfungen wie Mindestlänge, Höchstlänge und reguläre Ausdrücke ermöglichen.

Um die Erstellung und Auflistung von TODO-Elementen zu unterstützen, definieren wir das folgende Schema:

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

Um die Aktualisierung eines bestehenden TODO-Elements zu unterstützen, definieren wir ein weiteres Schema:

 class TODOUpdate(TODOCreate): id: int

Damit sind wir mit der Definition von Schemata für den gesamten Datenaustausch fertig. Wir richten unsere Aufmerksamkeit nun auf Request-Handler, bei denen diese Schemas verwendet werden, um die ganze schwere Arbeit der Datenkonvertierung und -validierung kostenlos zu erledigen.

Lassen Sie Benutzer sich registrieren

Lassen Sie uns zunächst Benutzern erlauben, sich anzumelden, da alle unsere Dienste von einem authentifizierten Benutzer aufgerufen werden müssen. Wir schreiben unseren ersten Request-Handler unter Verwendung des oben definierten UserCreate und UserBase Schemas.

 @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

In diesem kurzen Stück Code passiert eine Menge. Wir haben einen Decorator verwendet, um das HTTP-Verb, den URI und das Schema erfolgreicher Antworten anzugeben. Um sicherzustellen, dass der Benutzer die richtigen Daten übermittelt hat, haben wir den Anfragetext mit einem zuvor definierten UserCreate Schema als Hinweis eingegeben. Die Methode definiert einen weiteren Parameter, um einen Zugriff auf die Datenbank zu erhalten – dies ist die Abhängigkeitsinjektion in Aktion und wird später in diesem Tutorial erläutert.

Sicherung unserer API

Wir wollen die folgenden Sicherheitsfunktionen in unserer App:

  • Passwort-Hashing
  • JWT-basierte Authentifizierung

Für das Passwort-Hashing können wir Passlib verwenden. Lassen Sie uns Funktionen definieren, die das Hashen von Passwörtern handhaben und prüfen, ob ein Passwort korrekt ist.

 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

Um die JWT-basierte Authentifizierung zu aktivieren, müssen wir JWTs generieren und sie decodieren, um Benutzeranmeldeinformationen zu erhalten. Wir definieren die folgenden Funktionen, um diese Funktionalität bereitzustellen.

 # 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

Tokens bei erfolgreicher Anmeldung ausgeben

Jetzt definieren wir einen Login-Endpunkt und implementieren den OAuth2-Passwortablauf. Dieser Endpunkt erhält eine E-Mail und ein Kennwort. Wir werden die Anmeldeinformationen mit der Datenbank vergleichen und bei Erfolg ein JSON-Web-Token für den Benutzer ausstellen.

Um die Anmeldeinformationen zu erhalten, verwenden wir OAuth2PasswordRequestForm , das Teil der Sicherheitsdienstprogramme von FastAPI ist.

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

Verwenden von Dependency Injection für den Zugriff auf die Datenbank und den Schutz von Endpunkten

Wir haben den Anmeldeendpunkt eingerichtet, der einem Benutzer bei erfolgreicher Anmeldung ein JWT bereitstellt. Der Benutzer kann dieses Token im lokalen Speicher speichern und es unserem Backend als Autorisierungsheader anzeigen. Die Endpunkte, die Zugriff nur von angemeldeten Benutzern erwarten, können das Token entschlüsseln und herausfinden, wer der Anforderer ist. Diese Art von Arbeit ist nicht an einen bestimmten Endpunkt gebunden, sondern es handelt sich um eine gemeinsame Logik, die in allen geschützten Endpunkten verwendet wird. Am besten richten Sie die Token-Decodierungslogik als Abhängigkeit ein, die in jedem Request-Handler verwendet werden kann.

In der FastAPI-Sprache würden unsere Pfadoperationsfunktionen (Request-Handler) dann von get_current_user abhängen. Die get_current_user Abhängigkeit muss eine Verbindung zur Datenbank haben und sich in die OAuth2PasswordBearer -Logik der FastAPI einklinken, um ein Token zu erhalten. Wir lösen dieses Problem, indem wir get_current_user von anderen Funktionen abhängig machen. Auf diese Weise können wir Abhängigkeitsketten definieren, was ein sehr mächtiges Konzept ist.

 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

Angemeldete Benutzer können TODOs CRUD

Bevor wir die Pfadoperationsfunktionen für TODO Create, Read, Update, Delete (CRUD) schreiben, definieren wir die folgenden Hilfsfunktionen, um das tatsächliche CRUD auf der Datenbank auszuführen.

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

Diese Funktionen auf Datenbankebene werden in den folgenden REST-Endpunkten verwendet:

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

Tests schreiben

Lassen Sie uns ein paar Tests für unsere TODO-API schreiben. FastAPI stellt eine TestClient -Klasse bereit, die auf der beliebten Requests-Bibliothek basiert, und wir können die Tests mit Pytest ausführen.

Um sicherzustellen, dass nur eingeloggte Benutzer ein TODO erstellen können, können wir so etwas schreiben:

 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

Der folgende Test überprüft unseren Anmeldeendpunkt und generiert ein JWT, wenn gültige Anmeldedaten vorliegen.

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

Zusammenfassend

Wir sind mit der Implementierung einer sehr einfachen TODO-App mit FastAPI fertig. Inzwischen haben Sie gesehen, wie gut Typhinweise bei der Definition der Form eingehender und ausgehender Daten über unsere REST-Schnittstelle eingesetzt werden. Wir definieren die Schemas an einem Ort und überlassen es FastAPI, die Datenvalidierung und -konvertierung anzuwenden. Die andere bemerkenswerte Funktion ist die Abhängigkeitsinjektion. Wir haben dieses Konzept verwendet, um die gemeinsam genutzte Logik zum Abrufen einer Datenbankverbindung, zum Decodieren des JWT zum Abrufen des aktuell angemeldeten Benutzers und zum Implementieren von einfachem OAuth2 mit Kennwort und Träger zu packen. Wir haben auch gesehen, wie Abhängigkeiten miteinander verkettet werden können.

Wir können dieses Konzept einfach anwenden, um Funktionen wie den rollenbasierten Zugriff hinzuzufügen. Außerdem schreiben wir prägnanten und leistungsstarken Code, ohne die Besonderheiten eines Frameworks zu lernen. In einfachen Worten, FastAPI ist eine Sammlung leistungsstarker Tools, die Sie nicht lernen müssen, da es sich nur um modernes Python handelt. Habe Spaß.