Высокопроизводительные приложения с Python — руководство по FastAPI

Опубликовано: 2022-03-11

Хорошие языковые фреймворки позволяют быстрее производить качественные продукты. Отличные фреймворки даже делают процесс разработки приятным. FastAPI — это новая веб-инфраструктура Python, мощная и приятная в использовании. Следующие функции делают FastAPI достойным внимания:

Скорость: FastAPI — одна из самых быстрых веб-платформ Python. Фактически, его скорость не уступает Node.js и Go. Проверьте эти тесты производительности.
Подробные и простые в использовании документы для разработчиков
Введите свой код и получите бесплатную проверку и преобразование данных.
Легко создавайте плагины, используя внедрение зависимостей.

Создание приложения TODO

Чтобы изучить большие идеи, лежащие в основе FastAPI, давайте создадим приложение TODO, которое создает списки дел для своих пользователей. Наше крошечное приложение предоставит следующие возможности:

Зарегистрироваться и войти
Добавить новый элемент TODO
Получить список всех TODO
Удалить/обновить элемент TODO

SQLAlchemy для моделей данных

В нашем приложении всего две модели: User и TODO. С помощью SQLAlchemy, набора инструментов базы данных для Python, мы можем выразить наши модели следующим образом:

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

Как только наши модели будут готовы, давайте напишем файл конфигурации для SQLAlchemy, чтобы он знал, как установить соединение с базой данных.

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

Раскройте всю мощь текстовых подсказок

Значительная часть любого проекта API касается рутинных вещей, таких как проверка и преобразование данных. Давайте займемся этим заранее, прежде чем мы перейдем к написанию обработчиков запросов. С помощью FastAPI мы выражаем схему наших входящих/исходящих данных с помощью моделей pydantic, а затем используем эти модели pydantic для ввода подсказок и бесплатной проверки и преобразования данных. Обратите внимание, что эти модели не связаны с нашим рабочим процессом базы данных и только определяют форму данных, которые втекают и выходят из нашего интерфейса REST. Чтобы написать модели pydantic, подумайте обо всех способах поступления и вывода информации о пользователе и TODO.

Традиционно новый пользователь регистрируется в нашей службе TODO, а существующий пользователь входит в систему. Оба этих взаимодействия имеют дело с информацией о пользователе, но форма данных будет разной. Нам нужно больше информации от пользователей во время регистрации и минимум (только адрес электронной почты и пароль) при входе в систему. Это означает, что нам нужны две пидантические модели для выражения этих двух разных форм информации о пользователе.

Однако в нашем приложении TODO мы будем использовать встроенную поддержку OAuth2 в FastAPI для потока входа на основе веб-токенов JSON (JWT). Нам просто нужно определить здесь схему UserCreate , чтобы указать данные, которые будут поступать в нашу конечную точку регистрации, и схему UserBase , которая будет возвращаться в качестве ответа в случае успешного процесса регистрации.

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

Здесь мы пометили фамилию, имя и пароль как строку, но ее можно еще больше сузить, используя строки с ограничениями pydantic, которые позволяют выполнять такие проверки, как минимальная длина, максимальная длина и регулярные выражения.

Для поддержки создания и перечисления элементов TODO мы определяем следующую схему:

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

Для поддержки обновления существующего элемента TODO мы определяем другую схему:

 class TODOUpdate(TODOCreate): id: int

На этом мы закончили определение схем для всех обменов данными. Теперь мы обратим внимание на обработчики запросов, в которых эти схемы будут использоваться для бесплатного выполнения всей тяжелой работы по преобразованию и проверке данных.

Разрешить пользователям регистрироваться

Во-первых, давайте разрешим пользователям регистрироваться, так как все наши службы должны быть доступны аутентифицированному пользователю. Мы пишем наш первый обработчик запросов, используя схемы UserCreate и UserBase определенные выше.

 @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

В этом коротком фрагменте кода происходит многое. Мы использовали декоратор, чтобы указать глагол HTTP, URI и схему успешных ответов. Чтобы убедиться, что пользователь отправил правильные данные, мы набрали тело запроса с ранее определенной схемой UserCreate . Этот метод определяет еще один параметр для получения дескриптора базы данных — это внедрение зависимостей в действии, которое обсуждается далее в этом руководстве.

Защита нашего API

Нам нужны следующие функции безопасности в нашем приложении:

Хэширование пароля
Аутентификация на основе JWT

Для хеширования паролей мы можем использовать Passlib. Давайте определим функции, которые обрабатывают хеширование паролей и проверяют правильность пароля.

 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, нам нужно сгенерировать JWT, а также декодировать их, чтобы получить учетные данные пользователя. Мы определяем следующие функции, чтобы обеспечить эту функциональность.

 # 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

Выдача токенов при успешном входе в систему

Теперь мы определим конечную точку входа и реализуем поток паролей OAuth2. Эта конечная точка получит адрес электронной почты и пароль. Мы проверим учетные данные по базе данных и в случае успеха выдадим пользователю веб-токен JSON.

Чтобы получить учетные данные, мы будем использовать OAuth2PasswordRequestForm , который является частью утилит безопасности 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"}

Использование внедрения зависимостей для доступа к БД и защиты конечных точек

Мы настроили конечную точку входа, которая предоставляет пользователю JWT после успешного входа в систему. Пользователь может сохранить этот токен в локальном хранилище и показать его нашей серверной части в качестве заголовка авторизации. Конечные точки, которые ожидают доступа только от вошедших в систему пользователей, могут декодировать токен и выяснить, кто запрашивающий. Этот вид работы не привязан к конкретной конечной точке, это скорее общая логика, используемая во всех защищенных конечных точках. Лучше всего настроить логику декодирования токена как зависимость, которую можно использовать в любом обработчике запросов.

Говоря языком FastAPI, наши функции обработки пути (обработчики запросов) будут зависеть от get_current_user . Зависимость get_current_user должна иметь подключение к базе данных и подключаться к логике FastAPI OAuth2PasswordBearer для получения токена. Мы решим эту проблему, сделав get_current_user зависимой от других функций. Таким образом, мы можем определить цепочки зависимостей, что является очень мощной концепцией.

 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

Вошедшие пользователи могут CRUD TODO

Прежде чем мы напишем функции операций пути для TODO Create, Read, Update, Delete (CRUD), мы определяем следующие вспомогательные функции для выполнения фактического CRUD в базе данных.

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

Эти функции уровня базы данных будут использоваться в следующих конечных точках 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"}

Пишите тесты

Давайте напишем несколько тестов для нашего TODO API. FastAPI предоставляет класс TestClient , основанный на популярной библиотеке Requests, и мы можем запускать тесты с помощью Pytest.

Чтобы убедиться, что только авторизованные пользователи могут создавать TODO, мы можем написать что-то вроде этого:

 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

Следующий тест проверяет нашу конечную точку входа и генерирует JWT, если представлены действительные учетные данные для входа.

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

Подводя итоги

Мы закончили реализацию очень простого приложения TODO с использованием FastAPI. К этому моменту вы уже видели, как хорошо используются подсказки типов для определения формы входящих и исходящих данных через наш интерфейс REST. Мы определяем схемы в одном месте и оставляем на FastAPI проверку и преобразование данных. Другая заслуживающая внимания особенность — внедрение зависимостей. Мы использовали эту концепцию для упаковки общей логики получения подключения к базе данных, декодирования JWT для получения текущего пользователя, вошедшего в систему, и реализации простого OAuth2 с паролем и носителем. Мы также увидели, как зависимости могут быть объединены в цепочку.

Мы можем легко применить эту концепцию для добавления таких функций, как доступ на основе ролей. Кроме того, мы пишем лаконичный и мощный код, не изучая особенности фреймворка. Проще говоря, FastAPI — это набор мощных инструментов, которые вам не нужно изучать, потому что это просто современный Python. Повеселись.