FastAPI中如何实现安全路由与生成访问Token？

我有一个基于Flask的后端REST API，计划迁移至FastAPI，但不清楚如何在FastAPI中实现安全路由以及创建访问Token。在Flask中，我借助flask_jwt_extended库的@jwt_required()装饰器实现安全路由，用create_access_token()函数生成JWT Token。请问FastAPI是否具备类似功能，或该如何完成此类功能的实现？

以下是我在Flask中的实现示例：

import hashlib
import traceback
from datetime import timedelta
from http import HTTPStatus
from flask import Flask, jsonify, request
from flask_jwt_extended import JWTManager, jwt_required, get_jwt_identity, create_access_token

app = Flask(__name__)
jwt = JWTManager(app)
app.config["JWT_SECRET_KEY"] = "very-secret1234567890"
app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(minutes=15)
app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=30)
host = "localhost"
port = 5000
test_password = "test_password"
db = [
    {
        "username": "test_user",
        "email": "test_email.gmail.com",
        "password": hashlib.sha256(test_password.encode()).hexdigest()
    }
]


@app.route('/login', methods=['POST'])
def login():
    try:
        json_data = request.get_json()
        email = json_data.get("email")
        password = json_data.get("password")
        if not email or not password:
            response = jsonify(error="'email' and 'password' are required")
            return response, HTTPStatus.BAD_REQUEST
        # Check if email exists in DB
        user_result = [user for user in db if user["email"].lower() == email.lower()]
        # Check if the password is correct
        encoded_password = hashlib.sha256(password.encode()).hexdigest()
        if not user_result or user_result[0]["password"] != encoded_password:
            response = jsonify(error="Wrong credentials")
            return response, HTTPStatus.BAD_REQUEST
        user = user_result[0]
        # Generate JWT token and return it
        access_token = create_access_token(identity=user["username"])
        response = jsonify(username=user["username"], token=access_token)
        return response, HTTPStatus.OK
    except Exception as e:
        print(f"Error: {e}")
        print(traceback.format_exc())
        response = jsonify(result={"error": "Server error"})
        return response, HTTPStatus.INTERNAL_SERVER_ERROR


@app.route('/secured_page', methods=['GET'])
@jwt_required()
def __create_participant():
    try:
        response = jsonify(message="You are logged in as {}".format(get_jwt_identity()))
        return response, HTTPStatus.OK
    except Exception as e:
        print(f"Error: {e}")
        print(traceback.format_exc())
        response = jsonify(result={"error": "Server error"})
        return response, HTTPStatus.INTERNAL_SERVER_ERROR


if __name__ == '__main__':
    app.run(host=host, port=port, debug=True)

FastAPI本身不内置JWT功能，但可以通过官方推荐的第三方库快速实现，核心依赖包括：

• python-jose[cryptography]：用于生成和验证JWT Token
• passlib[bcrypt]：（可选，推荐）用于更安全的密码哈希存储，也可保留原SHA256实现

1. 安装依赖 #

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt]

2. 完整代码实现 #

import hashlib
import traceback
from datetime import timedelta, datetime
from http import HTTPStatus
from typing import Optional

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from pydantic import BaseModel

# 配置项
SECRET_KEY = "very-secret1234567890"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 30

host = "localhost"
port = 5000
test_password = "test_password"
db = [
    {
        "username": "test_user",
        "email": "test_email.gmail.com",
        "password": hashlib.sha256(test_password.encode()).hexdigest()
    }
]

app = FastAPI()

# 定义OAuth2认证方案，指定Token获取路径
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

# Pydantic模型：用于请求/响应数据校验
class Token(BaseModel):
    access_token: str
    token_type: str
    username: str

class TokenData(BaseModel):
    username: Optional[str] = None

# 工具函数：从数据库获取用户
def get_user_by_email(db, email: str):
    user_result = [user for user in db if user["email"].lower() == email.lower()]
    return user_result[0] if user_result else None

def get_user_by_username(db, username: str):
    user_result = [user for user in db if user["username"] == username]
    return user_result[0] if user_result else None

# 工具函数：验证密码
def verify_password(plain_password: str, hashed_password: str):
    encoded_password = hashlib.sha256(plain_password.encode()).hexdigest()
    return encoded_password == hashed_password

# 工具函数：生成访问Token
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# 依赖函数：验证Token并获取当前用户
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if not username:
            raise credentials_exception
        user = get_user_by_username(db, username)
        if not user:
            raise credentials_exception
        return user
    except JWTError:
        raise credentials_exception

# 登录接口
@app.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    try:
        # 用form_data.username接收邮箱（OAuth2默认字段名，也可自定义请求模型）
        user = get_user_by_email(db, form_data.username)
        if not user or not verify_password(form_data.password, user["password"]):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Wrong credentials"
            )
        access_token = create_access_token(
            data={"sub": user["username"]},
            expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        )
        return {"access_token": access_token, "token_type": "bearer", "username": user["username"]}
    except Exception as e:
        print(f"Error: {e}")
        print(traceback.format_exc())
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
            detail="Server error"
        )

# 安全路由示例
@app.get("/secured_page")
async def secured_page(current_user: dict = Depends(get_current_user)):
    try:
        return {"message": f"You are logged in as {current_user['username']}"}
    except Exception as e:
        print(f"Error: {e}")
        print(traceback.format_exc())
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
            detail="Server error"
        )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host=host, port=port)

关键功能对应说明 #

1. 安全路由控制：用Depends(get_current_user)作为路由依赖，替代Flask的@jwt_required()装饰器，只有携带有效Token的请求才能访问。
2. Token生成：create_access_token函数对应Flask的同名函数，通过jwt.encode生成JWT。
3. 身份解析：get_current_user函数对应Flask的get_jwt_identity，负责解析Token并验证用户合法性。
4. 认证方案：OAuth2PasswordBearer对应Flask的JWTManager，定义Token的获取规则和路径。

额外提示 #

• 原代码使用SHA256存储密码，建议改用bcrypt等加盐哈希算法，passlib库可快速实现。
• FastAPI的Pydantic模型会自动校验请求/响应数据，比Flask手动处理更高效可靠。
• 运行FastAPI应用需用uvicorn替代Flask的app.run，这是官方推荐的ASGI服务器。

内容的提问来源于stack exchange，提问作者Vlad Efanov