作者都是各自领域经过审查的专家,并撰写他们有经验的主题. 我们所有的内容都经过同行评审,并由同一领域的Toptal专家验证.
Zubair有三年的Python开发经验,使用Django、Flask和FastAPI. He works in avionics and aerospace.
良好的编程语言框架可以更容易、更快地生成高质量的产品. 好的框架甚至会让整个开发过程变得愉快. FastAPI是一个新的Python web框架,它功能强大,使用起来很愉快. 以下特性使Python FastAPI框架值得一试:
To explore the big ideas behind FastAPI, let’s build a TODO app, which sets up to-do lists for its users. 我们的FastAPI示例应用程序将提供以下功能:
Our app has just two models: User and TODO. 在SQLAlchemy (Python的数据库工具包)的帮助下,我们可以这样表达我们的模型:
class User(Base):
__tablename__ = "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.id"))
owner = relationship("User", back_populates="todos")
Once our models are ready, 让我们为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项目中都有相当一部分涉及数据验证和转换等日常工作. 在开始编写请求处理程序之前,让我们先解决这个问题. With FastAPI, 我们使用pydantic模型表示传入/传出数据的模式,然后使用这些pydantic模型键入提示并享受免费的数据验证和转换. 请注意,这些模型与我们的数据库工作流无关,它们只指定进出REST接口的数据的形状. To write pydantic models, 考虑用户和待办事项信息进出的所有方式.
传统上,新用户将注册我们的TODO服务,现有用户将登录. 这两种交互都处理用户信息,但数据的形状将有所不同. 我们需要更多的信息,从用户在注册和最小(只有电子邮件和密码)登录时. 这意味着我们需要两个解析模型来表达这两种不同形状的用户信息.
In our TODO app, however, 我们将利用FastAPI中内置的OAuth2支持来实现基于JSON Web Tokens (JWT)的登录流. We just need to define a UserCreate
模式指定将流入我们的注册端点的数据 UserBase
模式,以便在注册过程成功时作为响应返回.
from pydantic import BaseModel
from pydantic import EmailStr
class UserBase(BaseModel):
email: EmailStr
class UserCreate(UserBase):
lname: str
fname: str
password: str
Here, we marked last name, first name, and password as a string, 但它可以通过使用pydantic进一步收紧 constrained strings 这将启用最小长度、最大长度和正则表达式等检查.
为了支持TODO项目的创建和列表,我们定义了以下模式:
class TODOCreate(BaseModel):
text: str
completed: bool
为了支持现有TODO项的更新,我们定义了另一个模式:
class TODOUpdate(TODOCreate):
id: int
至此,我们就完成了为所有数据交换定义模式的工作. 现在我们将注意力转向请求处理程序,这些模式将用于免费完成所有繁重的数据转换和验证工作.
First, let’s allow users to sign up, 因为我们所有的服务都需要经过身份验证的用户访问. We write our first request handler using the UserCreate
and UserBase
schema defined above.
@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
schema. 该方法定义了用于获取数据库句柄的另一个参数——这是实际的依赖注入,将在本FastAPI教程后面讨论.
We want the following security features in our FastAPI example:
For password hashing, we can use 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)
defauthenticate_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
To enable JWT-based authentication, 我们需要生成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.编码(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.解码(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
现在,我们将定义一个Login端点并实现OAuth2密码流. 该端点将接收电子邮件和密码. 我们将根据数据库检查凭据, and on success, issue a JSON web token to the user.
为了获得证书,我们将使用 OAuth2PasswordRequestForm
,它是FastAPI安全实用程序的一部分.
@app.post("/api/token", response_model=schemas.Token)
def login_for_access_token(db: Session = Depends(get_db),
form_data: OAuth2PasswordRequestForm = Depends()):
"""为有效凭证生成访问令牌"""
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":用户.email},
expires_delta=access_token_expires)
返回{"access_token": access_token, "token_type": "bearer"}
我们已经设置了登录端点,在成功登录时向用户提供JWT. 用户可以将此令牌保存在本地存储中,并将其作为授权头显示给我们的后端. 只期望登录用户访问的端点可以解码令牌并找出请求者是谁. 这种工作与特定的端点无关, 相反,它是在所有受保护端点中使用的共享逻辑. 最好将令牌解码逻辑设置为可以在任何请求处理程序中使用的依赖项.
在fastapi中,我们的路径操作函数(请求处理程序)将依赖于 get_current_user
. The get_current_user
依赖项需要连接到数据库,并挂钩到FastAPI OAuth2PasswordBearer
logic to obtain a token. We will resolve this issue by making get_current_user
depend on other functions. 这样,我们就可以定义依赖链,这是一个非常强大的概念.
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)):
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
在我们为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
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])
def get_own_todos(current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""返回当前用户"""拥有的TODOs列表"""
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 /待办事项/ {todo_id}”,response_model =模式.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"}
Let’s write a few tests for our TODO API. FastAPI provides a 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()
我们已经使用FastAPI实现了一个非常简单的TODO应用程序. By now, 您已经看到了类型提示在定义通过REST接口传入和传出数据的形状方面的强大作用. 我们在一个地方定义模式,让FastAPI来应用数据验证和转换. 另一个值得注意的特性是依赖注入. 我们使用这个概念来打包获取数据库连接的共享逻辑, 解码JWT以获取当前登录的用户, 并实现简单的OAuth2与密码和承载. 我们还看到了依赖关系是如何链接在一起的.
我们可以很容易地应用这个概念来添加诸如基于角色的访问之类的特性. FastAPI的性能和文档是首屈一指的,使其易于掌握和使用. 我们正在编写简洁而强大的代码,而无需学习框架的特性. In simple words, FastAPI是一个强大的工具集合,你不需要学习,因为它们只是现代的Python. Have fun.
FastAPI是一个Python框架和一组工具,它使开发人员能够使用REST接口调用常用函数来实现应用程序. 它可以通过REST API访问,以调用应用程序的公共构建块. 在本例中,作者使用FastAPI创建帐户、登录和身份验证.
像所有REST接口一样,FastAPI是从代码中调用的. 它提供了传入数据的类型提示等特性, dependency injection, 还有身份验证,这样你就不用自己写函数了.
While an open-source framework, FastAPI is fully production-ready, with excellent documentation, support, and an easy-to-use interface. 它可以用来构建和运行与用其他脚本语言编写的应用程序一样快的应用程序.
如果接口定义良好且底层代码已经过优化,则可以. 文档声称FastAPI的性能和Node一样好.js and Go.
Kamra Kalan, Punjab, Pakistan
Member since March 23, 2020
Zubair有三年的Python开发经验,使用Django、Flask和FastAPI. He works in avionics and aerospace.
World-class articles, delivered weekly.
World-class articles, delivered weekly.
Join the Toptal® community.