良好的编程语言框架可以更容易、更快地生成高质量的产品. 好的框架甚至会让整个开发过程变得愉快. FastAPI是一个新的Python web框架，它功能强大，使用起来很愉快. 以下特性使Python FastAPI框架值得一试:

• 速度:FastAPI是最快的Python web框架之一. 事实上，它的速度与Node相当.js and Go. Check 这些FastAPI性能测试.
• The FastAPI documentation 是否详细且易于使用.
• 键入提示您的代码，并获得免费的数据验证和转换.
• 使用依赖注入轻松创建插件.

Python FastAPI教程:构建TODO应用程序

探索FastAPI背后的重要思想, 让我们构建一个TODO应用程序, 它会为用户设置待办事项列表. 我们的FastAPI示例应用程序将提供以下功能:

• Signup and Login
• Add new TODO item
• 列出所有待办事项
• 删除/更新待办事项

数据模型的SQLAlchemy

我们的应用程序只有两个模型:User和TODO. 在SQLAlchemy (Python的数据库工具包)的帮助下，我们可以这样表达我们的模型:

class User(Base):
   __tableame__ = "users"
   id = Column(Integer, primary_key=True, index=True)
   lname = Column(String)
   fname = Column(String)
   email = Column(字符串，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 . name ").id"))
   owner = relationship("User"， back_populates="todos")

一旦我们的模型准备好, 让我们为SQLAlchemy编写配置文件，这样它就知道如何与数据库建立连接.

import os
从sqlalchemy导入create_engine
from sqlalchemy.ext.声明式导入declarative_base
from sqlalchemy.Orm导入sessionmaker
SQLALCHEMY_DATABASE_URL = os.环境(“SQLALCHEMY_DATABASE_URL”)
Engine = create_engine(
   SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

释放类型提示的力量

任何API项目中都有相当一部分涉及数据验证和转换等日常工作. 在开始编写请求处理程序之前，让我们先解决这个问题. With FastAPI, 我们使用pydantic模型表示传入/传出数据的模式，然后使用这些pydantic模型键入提示并享受免费的数据验证和转换. 请注意，这些模型与我们的数据库工作流无关，它们只指定进出REST接口的数据的形状. 写pydantic模型, 考虑用户和待办事项信息进出的所有方式.

传统上，新用户将注册我们的TODO服务，现有用户将登录. 这两种交互都处理用户信息，但数据的形状将有所不同. 我们需要更多的信息，从用户在注册和最小(只有电子邮件和密码)登录时. 这意味着我们需要两个解析模型来表达这两种不同形状的用户信息.

In our TODO app, however, 我们将利用FastAPI中内置的OAuth2支持来实现基于JSON Web Tokens (JWT)的登录流. 我们只需要定义a UserCreate 模式指定将流入我们的注册端点的数据 UserBase 模式，以便在注册过程成功时作为响应返回.

从pydantic导入BaseModel
从pydantic导入EmailStr
类用户群(BaseModel):
   email: EmailStr
类UserCreate(群):
   lname: str
   fname: str
   password: str

Here, we marked last name, first name, 和密码作为一个字符串, 但它可以通过使用pydantic进一步收紧 constrained strings 这将启用最小长度、最大长度和正则表达式等检查.

为了支持TODO项目的创建和列表，我们定义了以下模式:

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

为了支持现有TODO项的更新，我们定义了另一个模式:

类TODOUpdate (TODOCreate):
   id: int

至此，我们就完成了为所有数据交换定义模式的工作. 现在我们将注意力转向请求处理程序，这些模式将用于免费完成所有繁重的数据转换和验证工作.

Let Users Sign Up

First, 让我们允许用户注册, 因为我们所有的服务都需要经过身份验证的用户访问. 类编写第一个请求处理程序 UserCreate and UserBase schema defined above.

@app.邮报》(“/ api /用户”,response_model =模式.User)
defsignup (user_data: schema.UserCreate, db: Session = Depends(get_db)):
   """add new user"""
   user = crud.user_data get_user_by_email (db.email)
   if user:
   	提高textbox (status_code = 409,
   	                    detail="邮箱已注册.")
   signedup_user = crud.user_data create_user (db)
   return signedup_user

在这一小段代码中有很多内容. 我们使用了一个修饰符来指定HTTP谓词、URI和成功响应的模式. 以确保用户提交了正确的数据, 我们已经用先前定义的提示键入了请求体 UserCreate schema. 该方法定义了用于获取数据库句柄的另一个参数——这是实际的依赖注入，将在本FastAPI教程后面讨论.

Securing Our API

我们需要以下内容 security features 在我们的FastAPI示例中:

• Password hashing
• JWT-based身份验证

对于密码散列，我们可以使用Passlib. 让我们定义处理密码散列和检查密码是否正确的函数.

from passlib.导入CryptContext
 
pwd_context = CryptContext(schemes=["bcrypt"]， deprecated="auto")
 
Def verify_password(plain_password, hashed_password):
   return pwd_context.验证(plain_password hashed_password)
 
 
def get_password_hash(密码):
   return pwd_context.hash(password)
 
defauthenticate_user (db, email: str, password: str):
   user = crud.get_user_by_email(数据库、电子邮件)
   if not user:
   	return False
   如果不是，verify_password(password, user。.hashed_password):
   	return False
   return user

启用基于jwt的身份验证, 我们需要生成jwt并对其进行解码以获得用户凭据. 我们定义了以下函数来提供此功能.

# install PyJWT
import jwt
from fastapi.security导入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.更新({}“经验值”:到期)
   encoded_jwt = jwt.编码(to_encode, SECRET_KEY, algorithm= algorithm)
   return encoded_jwt
Def decode_access_token(db, token):
   credentials_exception = HTTPException(
   	status_code = HTTP_401_UNAUTHORIZED,
   	detail="无法验证凭据"，
   	头={“WWW-Authenticate”:“持票人”},
   )
   try:
   	payload = jwt.解码(token, SECRET_KEY, algorithms=[ALGORITHM])
   	email: str = payload.get("sub")
   	if email is None:
       	提高credentials_exception
   	token_data = schemas.TokenData(邮件=电子邮件)
   except PyJWTError:
   	提高credentials_exception
   user = crud.get_user_by_email(数据库、电子邮件= token_data.email)
   if user is None:
   	提高credentials_exception
   return user

成功登录时发出令牌

现在，我们将定义一个Login端点并实现OAuth2密码流. 该端点将接收电子邮件和密码. 我们将根据数据库检查凭据, and on success, 向用户发出JSON web令牌.

为了获得证书，我们将使用 OAuth2PasswordRequestForm，它是FastAPI安全实用程序的一部分.

@app.邮报》(“/ api /令牌”,response_model =模式.Token)
def login_for_access_token(db: Session = Depends(get_db)，
                      	form_data: OAuth2PasswordRequestForm = Depends()):
   """为有效凭证生成访问令牌"""
   User = authenticate_user(db, form_data . User.username, form_data.password)
   if not user:
   	raise HTTPException(
       	status_code = HTTP_401_UNAUTHORIZED,
       	detail="不正确的电子邮件或密码"，
       	头={“WWW-Authenticate”:“持票人”},
   	)
   access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
   Access_token = create_access_token(data={"sub":用户.email},
                                  	expires_delta = access_token_expires)
   返回{"access_token": access_token， "token_type": "bearer"}

使用依赖注入来访问数据库和保护端点

我们已经设置了登录端点，在成功登录时向用户提供JWT. 用户可以将此令牌保存在本地存储中，并将其作为授权头显示给我们的后端. 只期望登录用户访问的端点可以解码令牌并找出请求者是谁. 这种工作与特定的端点无关, 相反，它是在所有受保护端点中使用的共享逻辑. 最好将令牌解码逻辑设置为可以在任何请求处理程序中使用的依赖项.

在fastapi中，我们的路径操作函数(请求处理程序)将依赖于 get_current_user. The get_current_user 依赖项需要连接到数据库，并挂钩到FastAPI OAuth2PasswordBearer 获取令牌的逻辑. 我们会解决这个问题 get_current_user 依赖于其他函数. 这样，我们就可以定义依赖链，这是一个非常强大的概念.

def get_db():
   为路径操作函数提供db会话
   try:
   	db = SessionLocal()
   	yield db
   finally:
   	db.close()
get_current_user(db: Session = Depends(get_db)，
                	token: str = Depends(oauth2_scheme):
   返回decode_access_token(db, token)
@app.get (" / api /我”,response_model =模式.User)
(current_user:模型.User = Depends(get_current_user)):
   """返回当前用户设置"""
   return current_user

登录用户可以删除待办事项

在我们为TODO Create编写路径操作函数之前, Read, Update, Delete (CRUD), 我们定义了以下帮助函数来对数据库执行实际的CRUD.

def create_todo(db: Session, current_user: models.User, todo_data:模式.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
db: Session, todo_data: schema.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 =[模式列表.TODO])
current_user:模型.User = Depends(get_current_user)，
             	db: Session = Depends(get_db)):
   """返回当前用户"""拥有的TODOs列表"""
   todos = crud.current_user get_user_todos (db.id)
   return todos
@app.邮报》(“/ api /行动计划”,response_model =模式.TODO)
定义add_a_todo(todo_data:模式.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 /待办事项/ {todo_id}”,response_model =模式.TODO)
函数update_a_todo(todo_id: int)
             	todo_data: schemas.TODOUpdate,
             	current_user: models.User = Depends(get_current_user)，
             	db: Session = Depends(get_db)):
   更新并返回给定id的TODO
   todo = crud.todo_id get_todo (db)
   updated_todo = crud.Update_todo (db, todo_id, todo_data)
   return updated_todo
@app.删除(“/ api /行动计划/ {todo_id}”)
Def delete_a_meal(todo_id: int)
             	current_user: models.User = Depends(get_current_user)，
             	db: Session = Depends(get_db)):
   “”“删除指定id的TODO”“”
   crud.todo_id delete_meal (db)
   返回{"detail": "TODO已删除"}

Write Tests

让我们为TODO API编写一些测试. FastAPI provides a TestClient 类，它基于流行的Requests库，我们可以使用Pytest运行测试.

为了确保只有登录的用户才能创建TODO，我们可以这样写:

from starlette.testclient导入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.邮报》(“/ api /行动计划”,数据= todo)
assert response.status_code == 401

下面的测试检查我们的登录端点，并在提供有效登录凭据的情况下生成JWT.

def test_user_can_obtain_auth_token ():
  response = client.邮报》(“/ api /令牌”,数据= good_credentials)
  assert response.status_code == 200
  在响应中断言'access_token'.json()
  在响应中断言'token_type'.json()

Summing It Up

我们已经使用FastAPI实现了一个非常简单的TODO应用程序. By now, 您已经看到了类型提示在定义通过REST接口传入和传出数据的形状方面的强大作用. 我们在一个地方定义模式，让FastAPI来应用数据验证和转换. 另一个值得注意的特性是依赖注入. 我们使用这个概念来打包获取数据库连接的共享逻辑, 解码JWT以获取当前登录的用户, 并实现简单的OAuth2与密码和承载. 我们还看到了依赖关系是如何链接在一起的.

我们可以很容易地应用这个概念来添加诸如基于角色的访问之类的特性. FastAPI的性能和文档是首屈一指的，使其易于掌握和使用. 我们正在编写简洁而强大的代码，而无需学习框架的特性. In simple words, FastAPI是一个强大的工具集合，你不需要学习，因为它们只是现代的Python. Have fun.