使用 Python 的高性能应用程序 – FastAPI 教程

已发表: 2022-03-11

良好的编程语言框架可以轻松更快地生产出优质产品。伟大的框架甚至使整个开发体验变得愉快。 FastAPI 是一个新的 Python Web 框架，功能强大且使用起来很愉快。以下特性使 FastAPI 值得一试：

速度：FastAPI 是最快的 Python Web 框架之一。事实上，它的速度与 Node.js 和 Go 不相上下。检查这些性能测试。
详细且易于使用的开发人员文档
键入提示您的代码并获得免费的数据验证和转换。
使用依赖注入轻松创建插件。

构建一个 TODO 应用程序

为了探索 FastAPI 背后的重要理念，让我们构建一个 TODO 应用程序，它为用户设置待办事项列表。我们的微型应用程序将提供以下功能：

注册和登录
添加新的 TODO 项
获取所有 TODO 的列表
删除/更新 TODO 项目

数据模型的 SQLAlchemy

我们的应用只有两个模型：User 和 TODO。在 Python 的数据库工具包 SQLAlchemy 的帮助下，我们可以像这样表达我们的模型：

 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 模型，请考虑 User 和 TODO 信息流入和流出的所有方式。

传统上，新用户将注册我们的 TODO 服务，而现有用户将登录。这两种交互都处理用户信息，但数据的形式会有所不同。我们在注册时需要更多用户信息，在登录时需要最少的信息（只有电子邮件和密码）。这意味着我们需要两个 pydantic 模型来表达这两种不同形式的用户信息。

然而，在我们的 TODO 应用程序中，我们将利用 FastAPI 中内置的 OAuth2 支持来实现基于 JSON Web 令牌 (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 Web 令牌。

为了接收凭证，我们将使用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 创建、读取、更新、删除（CRUD）编写路径操作函数之前，我们定义了以下帮助函数来在 db 上执行实际的 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()

这些 db 级函数将用于以下 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 提供了一个基于流行的 Requests 库的TestClient类，我们可以使用 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()

总结一下

我们已经使用 FastAPI 实现了一个非常简单的 TODO 应用程序。到目前为止，您已经看到类型提示的强大功能在通过我们的 REST 接口定义传入和传出数据的形状时得到了很好的利用。我们在一处定义模式并将其留给 FastAPI 来应用数据验证和转换。另一个值得注意的特性是依赖注入。我们用这个概念封装了获取数据库连接的共享逻辑，解码JWT获取当前登录的用户，实现简单的OAuth2密码和承载。我们还看到了如何将依赖项链接在一起。

我们可以轻松地应用此概念来添加基于角色的访问等功能。此外，我们正在编写简洁而强大的代码，而无需学习框架的特性。简而言之，FastAPI 是一组强大的工具，您无需学习，因为它们只是现代 Python。玩得开心。