Good programming language frameworks make it easy to produce quality products faster. 好的框架甚至会让整个开发过程变得愉快. FastAPI是一个新的Python web框架，它功能强大，使用起来很愉快. 以下特性使Python FastAPI框架值得一试:

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

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

探索FastAPI背后的重要思想, let’s build a TODO app, 它会为用户设置待办事项列表. 我们的FastAPI示例应用程序将提供以下功能:

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

数据模型的SQLAlchemy

我们的应用程序只有两个模型:User和TODO. With the help of SQLAlchemy, the database toolkit for Python, we can express our models like this:

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

一旦我们的模型准备好, let’s write the configuration file for SQLAlchemy so that it knows how to establish a connection with the database.

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

释放类型提示的力量

A sizable part of any API project concerns the routine stuff like data validation and conversion. 在开始编写请求处理程序之前，让我们先解决这个问题. With FastAPI, we express the schema of our incoming/outgoing data using pydantic models and then use these pydantic models to type hint and enjoy free data validation and conversion. Please note these models are not related to our database workflow and only specify the shape of data that’s flowing in and out of our REST interface. 写pydantic模型, 考虑用户和待办事项信息进出的所有方式.

Traditionally, a new user will sign up for our TODO service and an existing user will log in. Both of these interactions deal with User information, but the shape of data will be different. We need more information from users during signup and minimal (only email and password) when logging in. This means we need two pydantic models to express these two different shapes of User info.

In our TODO app, however, we will leverage the built-in OAuth2 support in FastAPI for a JSON Web Tokens (JWT)-based login flow. 我们只需要定义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 这将启用最小长度、最大长度和正则表达式等检查.

To support the creation and listing of TODO items, we define the following schema:

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

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

类TODOUpdate (TODOCreate):
   id: int

至此，我们就完成了为所有数据交换定义模式的工作. We now turn our attention to request handlers where these schemas will be used to do all the heavy lifting of data conversion and validation for free.

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

在这一小段代码中有很多内容. We have used a decorator to specify the HTTP verb, the URI, and the schema of successful responses. 以确保用户提交了正确的数据, 我们已经用先前定义的提示键入了请求体 UserCreate schema. The method defines another parameter for getting a handle on the database—this is dependency injection in action and is discussed later in this FastAPI tutorial.

Securing Our API

We want the following security features 在我们的FastAPI示例中:

• Password hashing
• JWT-based身份验证

对于密码散列，我们可以使用Passlib. Let’s define functions that handle password hashing and checking if a password is correct.

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

成功登录时发出令牌

Now, we will define a Login endpoint and implement the OAuth2 password flow. 该端点将接收电子邮件和密码. 我们将根据数据库检查凭据, 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"}

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

We have set up the login endpoint that provides a JWT to a user upon successful login. The user can save this token in local storage and show it to our back end as an Authorization header. The endpoints that expect access only from logged-in users can decode the token and find out who the requester is. 这种工作与特定的端点无关, 相反，它是在所有受保护端点中使用的共享逻辑. It’s best to set up the token-decoding logic as a dependency that can be used in any request handler.

In FastAPI-speak, our path operation functions (request handlers) would then depend on get_current_user. The get_current_user dependency needs to have a connection to the database and to hook into the FastAPI’s 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.get_todo(db, todo_id)
   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 class that’s based on the popular Requests library, and we can run the tests with Pytest.

To make sure only logged-in users can create a TODO, we can write something like this:

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

The following test checks our login endpoint and generates a JWT if presented with valid login credentials.

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, you’ve seen the power of type hints put to good use in defining the shape of incoming and outgoing data through our REST interface. We define the schemas at one place and leave it to FastAPI to apply data validation and conversion. 另一个值得注意的特性是依赖注入. We used this concept to package the shared logic of obtaining a database connection, 解码JWT以获取当前登录的用户, 并实现简单的OAuth2与密码和承载. 我们还看到了依赖关系是如何链接在一起的.

我们可以很容易地应用这个概念来添加诸如基于角色的访问之类的特性. FastAPI performance and documentation are second to none, making it easy to master and use. We are writing concise and powerful code without learning the peculiarities of a framework. In simple words, FastAPI is a collection of powerful tools that you don’t have to learn because they are just modern Python. Have fun.