Applications hautes performances avec Python - Un didacticiel FastAPI

De bons cadres de langage de programmation facilitent la production plus rapide de produits de qualité. De grands frameworks rendent même toute l'expérience de développement agréable. FastAPI est un nouveau framework Web Python puissant et agréable à utiliser. Les fonctionnalités suivantes font que FastAPI vaut la peine d'être essayé :

• Vitesse : FastAPI est l'un des frameworks Web Python les plus rapides. En fait, sa vitesse est à égalité avec Node.js et Go. Vérifiez ces tests de performance.
• Documents de développement détaillés et faciles à utiliser
• Tapez votre code et obtenez une validation et une conversion gratuites des données.
• Créez facilement des plugins en utilisant l'injection de dépendances.

Construire une application TODO

Pour explorer les grandes idées derrière FastAPI, créons une application TODO, qui établit des listes de tâches pour ses utilisateurs. Notre petite application fournira les fonctionnalités suivantes :

• Inscription et connexion
• Ajouter un nouvel élément TODO
• Obtenir une liste de tous les TODO
• Supprimer/Mettre à jour un élément TODO

SQLAlchemy pour les modèles de données

Notre application n'a que deux modèles : Utilisateur et TODO. Avec l'aide de SQLAlchemy, la boîte à outils de base de données pour Python, nous pouvons exprimer nos modèles comme ceci :

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

Une fois nos modèles prêts, écrivons le fichier de configuration de SQLAlchemy afin qu'il sache comment établir une connexion avec la base de données.

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

Libérez la puissance des indices de saisie

Une partie importante de tout projet d'API concerne les tâches de routine telles que la validation et la conversion des données. Abordons-le dès le départ avant de passer à l'écriture des gestionnaires de requêtes. Avec FastAPI, nous exprimons le schéma de nos données entrantes/sortantes à l'aide de modèles pydantiques, puis utilisons ces modèles pydantiques pour taper des indices et profiter de la validation et de la conversion gratuites des données. Veuillez noter que ces modèles ne sont pas liés à notre flux de travail de base de données et spécifient uniquement la forme des données qui entrent et sortent de notre interface REST. Pour écrire des modèles pydantic, pensez à toutes les façons dont les informations User et TODO vont entrer et sortir.

Traditionnellement, un nouvel utilisateur s'inscrira à notre service TODO et un utilisateur existant se connectera. Ces deux interactions traitent des informations de l'utilisateur, mais la forme des données sera différente. Nous avons besoin de plus d'informations de la part des utilisateurs lors de l'inscription et d'un minimum (uniquement e-mail et mot de passe) lors de la connexion. Cela signifie que nous avons besoin de deux modèles pydantic pour exprimer ces deux formes différentes d'informations utilisateur.

Dans notre application TODO, cependant, nous tirerons parti de la prise en charge intégrée d'OAuth2 dans FastAPI pour un flux de connexion basé sur JSON Web Tokens (JWT). Nous avons juste besoin de définir un schéma UserCreate ici pour spécifier les données qui seront transmises à notre point de terminaison d'inscription et un schéma UserBase à renvoyer en réponse si le processus d'inscription réussit.

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

Ici, nous avons marqué le nom, le prénom et le mot de passe en tant que chaîne, mais cela peut être encore renforcé en utilisant des chaînes contraintes pydantic qui permettent des vérifications telles que la longueur minimale, la longueur maximale et les regex.

Pour prendre en charge la création et la liste des éléments TODO, nous définissons le schéma suivant :

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

Pour prendre en charge la mise à jour d'un élément TODO existant, nous définissons un autre schéma :

 class TODOUpdate(TODOCreate): id: int

Avec cela, nous en avons fini avec la définition des schémas pour tous les échanges de données. Nous tournons maintenant notre attention vers les gestionnaires de requêtes où ces schémas seront utilisés pour effectuer gratuitement tout le gros du travail de conversion et de validation des données.

Laisser les utilisateurs s'inscrire

Tout d'abord, permettons aux utilisateurs de s'inscrire, car tous nos services doivent être accessibles par un utilisateur authentifié. Nous écrivons notre premier gestionnaire de requêtes en utilisant les UserCreate et UserBase définis ci-dessus.

 @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

Il se passe beaucoup de choses dans ce court morceau de code. Nous avons utilisé un décorateur pour spécifier le verbe HTTP, l'URI et le schéma des réponses réussies. Afin de s'assurer que l'utilisateur a soumis les bonnes données, nous avons tapé un indice dans le corps de la requête avec un schéma UserCreate défini précédemment. La méthode définit un autre paramètre pour obtenir un handle sur la base de données : il s'agit de l'injection de dépendances en action et elle est abordée plus loin dans ce didacticiel.

Sécuriser notre API

Nous voulons les fonctionnalités de sécurité suivantes dans notre application :

• Hachage du mot de passe
• Authentification basée sur JWT

Pour le hachage de mot de passe, nous pouvons utiliser Passlib. Définissons les fonctions qui gèrent le hachage du mot de passe et vérifient si un mot de passe est correct.

 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

Pour activer l'authentification basée sur JWT, nous devons générer des JWT et les décoder pour obtenir les informations d'identification de l'utilisateur. Nous définissons les fonctions suivantes pour fournir cette fonctionnalité.

 # 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

Émettre des jetons lors d'une connexion réussie

Nous allons maintenant définir un point de terminaison de connexion et implémenter le flux de mot de passe OAuth2. Ce terminal recevra un e-mail et un mot de passe. Nous vérifierons les informations d'identification par rapport à la base de données et, en cas de succès, émettrons un jeton Web JSON à l'utilisateur.

Pour recevoir les informations d'identification, nous utiliserons OAuth2PasswordRequestForm , qui fait partie des utilitaires de sécurité de 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"}

Utilisation de l'injection de dépendance pour accéder à la base de données et protéger les terminaux

Nous avons configuré le point de terminaison de connexion qui fournit un JWT à un utilisateur lors d'une connexion réussie. L'utilisateur peut enregistrer ce jeton dans le stockage local et le montrer à notre serveur principal en tant qu'en-tête d'autorisation. Les points de terminaison qui attendent un accès uniquement des utilisateurs connectés peuvent décoder le jeton et découvrir qui est le demandeur. Ce type de travail n'est pas lié à un point de terminaison particulier, mais plutôt à une logique partagée utilisée dans tous les points de terminaison protégés. Il est préférable de configurer la logique de décodage de jeton en tant que dépendance pouvant être utilisée dans n'importe quel gestionnaire de requêtes.

Dans FastAPI-speak, nos fonctions d'opération de chemin (gestionnaires de requêtes) dépendraient alors de get_current_user . La dépendance get_current_user doit avoir une connexion à la base de données et se connecter à la logique OAuth2PasswordBearer de OAuth2PasswordBearer pour obtenir un jeton. Nous allons résoudre ce problème en faisant dépendre get_current_user d'autres fonctions. De cette façon, nous pouvons définir des chaînes de dépendance, qui est un concept très puissant.

 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

Les utilisateurs connectés peuvent CRUD TODO

Avant d'écrire les fonctions d'opération de chemin pour TODO Créer, Lire, Mettre à jour, Supprimer (CRUD), nous définissons les fonctions d'assistance suivantes pour effectuer le CRUD réel sur la base de données.

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

Ces fonctions au niveau de la base de données seront utilisées dans les points de terminaison REST suivants :

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

Écrire des tests

Écrivons quelques tests pour notre API TODO. FastAPI fournit une classe TestClient basée sur la populaire bibliothèque Requests, et nous pouvons exécuter les tests avec Pytest.

Pour nous assurer que seuls les utilisateurs connectés peuvent créer un TODO, nous pouvons écrire quelque chose comme ceci :

 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

Le test suivant vérifie notre point de terminaison de connexion et génère un JWT s'il est présenté avec des informations d'identification de connexion valides.

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

En résumé

Nous avons terminé la mise en œuvre d'une application TODO très simple à l'aide de FastAPI. À présent, vous avez vu la puissance des indications de type mises à profit pour définir la forme des données entrantes et sortantes via notre interface REST. Nous définissons les schémas à un seul endroit et laissons à FastAPI le soin d'appliquer la validation et la conversion des données. L'autre caractéristique remarquable est l'injection de dépendances. Nous avons utilisé ce concept pour empaqueter la logique partagée d'obtention d'une connexion à la base de données, décodant le JWT pour obtenir l'utilisateur actuellement connecté et implémentant OAuth2 simple avec mot de passe et porteur. Nous avons également vu comment les dépendances peuvent être enchaînées.

Nous pouvons facilement appliquer ce concept pour ajouter des fonctionnalités telles que l'accès basé sur les rôles. De plus, nous écrivons du code concis et puissant sans apprendre les particularités d'un framework. En termes simples, FastAPI est une collection d'outils puissants que vous n'avez pas à apprendre car ils ne sont que du Python moderne. S'amuser.