新闻中心
FastAPI 多种认证方式(任选其一)实现指南

本教程详细阐述了如何在 fastapi 中实现多种认证机制(如 basic auth 和 jwt auth),并允许客户端任选其一进行认证。核心方法是修改各个认证依赖项,使其在认证失败时返回 `none` 而非立即抛出异常,从而使一个组合认证依赖能够基于“或”逻辑判断任一认证是否成功,最终实现灵活的多重认证支持。
在构建现代 Web API 时,支持多种认证方式以满足不同客户端或场景的需求是常见的实践。例如,您可能希望同时支持传统的 HTTP Basic 认证和基于令牌的 JWT 认证。然而,在 FastAPI 中直接将多个认证依赖项组合使用时,默认行为可能会导致问题:FastAPI 会强制所有依赖项都通过验证,如果其中一个依赖项在解析阶段抛出 HTTPException,那么后续的依赖项将不会被执行,从而无法实现“任选其一”的逻辑。
理解 FastAPI 依赖注入的默认行为
FastAPI 的依赖注入系统在处理多个 Depends 依赖时,遵循严格的顺序和错误处理机制。当一个依赖项(例如,一个认证函数)内部抛出 HTTPException 时,FastAPI 会立即捕获该异常并返回相应的 HTTP 响应,而不会继续执行后续的依赖项或路由处理函数。这意味着,如果您尝试通过一个自定义的组合依赖函数来包装多个认证依赖,并期望它们能像逻辑“或”一样工作,那么默认情况下是行不通的。
例如,以下尝试实现“或”逻辑的组合认证依赖:
# 假设 jwt_logged_user 和 basic_logged_user 在认证失败时会抛出 HTTPException
def auth_user(jwt_auth: Any = Depends(jwt_logged_user),
basic_auth: Any = Depends(basic_logged_user)):
if jwt_auth or basic_auth:
return jwt_auth or basic_auth
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid Credentials')如果 jwt_logged_user 认证失败并抛出异常,那么 basic_auth 依赖将永远不会被解析,auth_user 函数体也永远不会执行,从而无法检查 Basic Auth 是否有效。
核心解决方案:优雅地处理认证失败
要实现“任选其一”的认证逻辑,关键在于修改各个独立的认证依赖项,使其在认证失败时不再立即抛出 HTTPException,而是返回一个指示失败的值(例如 None)。这样,组合认证依赖就可以接收到所有独立认证的结果,并根据这些结果进行逻辑判断。
1. 修改 HTTP Basic 认证依赖
对于 HTTP Basic 认证,FastAPI 提供了 HTTPBasic 类。通过设置 auto_error=False,可以禁用其在认证失败时自动抛出 HTTPException 的行为。
首先,更新 HTTPBasic 实例:
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from typing import Annotated, Optional
import secrets
# ... 其他导入 ...
# 将 auto_error 设置为 False
security = HTTPBasic(auto_error=False)
def basic_logged_user(credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)]):
"""
处理 HTTP Basic 认证。
如果认证失败,返回 None 而非抛出异常。
"""
if credentials is None:
# 没有提供 Basic 认证凭据,或者凭据格式不正确 (由 auto_error=False 处理)
return None
current_username_bytes = credentials.username.encode("utf8")
correct_username_bytes = settings.SESSION_LOGIN_USER.encode("utf8")
is_correct_username = secrets.compare_digest(
current_username_bytes, correct_username_bytes
)
current_password_bytes = credentials.password.encode("utf8")
correct_password_bytes = settings.SESSION_LOGIN_PASS.encode("utf8")
is_correct_password = secrets.compare_digest(
current_password_bytes, correct_password_bytes
)
if not (is_correct_username and is_correct_password):
# 凭据不匹配,返回 None
return None
# 认证成功,返回用户名
return credentials.username关键点:
美图云修
商业级AI影像处理工具
50
查看详情
- security = HTTPBasic(auto_error=False):禁用 HTTPBasic 在凭据缺失或格式错误时自动抛出 401 异常。
- credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)]:声明 credentials 为 Optional 类型,因为 security 在 auto_error=False 时可能返回 None。
- if credentials is None::检查是否提供了凭据。
- if not (is_correct_username and is_correct_password): return None:在自定义逻辑中,如果凭据无效,也返回 None。
2. 修改 JWT 认证依赖
对于 JWT 认证,通常会使用 OAuth2PasswordBearer。同样,通过设置 auto_error=False 来禁用其自动错误处理。此外,您的令牌验证函数 (utils.verify_token) 也应该被修改,以在验证失败时返回 None 或被 try-except 块包裹。
假设 utils.OAuth2_scheme 是 OAuth2PasswordBearer 的实例,并且 utils.verify_token 函数负责验证 JWT。
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from typing import Annotated, Optional
from jose import JWTError, jwt # 假设您使用 jose 库
from pydantic import BaseModel
# ... 其他导入 ...
# 假设 utils 模块包含 OAuth2_scheme 和 verify_token
# utils.OAuth2_scheme 应该被初始化为 auto_error=False
# 例如:
# class TokenData(BaseModel):
# username: Optional[str] = None
# class Utils:
# OAuth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
# SECRET_KEY = "your-secret-key" # 替换为实际的密钥
# ALGORITHM = "HS256"
# def verify_token(self, token: str) -> dict:
# try:
# payload = jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
# username: str = payload.get("sub")
# if username is None:
# raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload")
# return {"username": username}
# except JWTError:
# raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials")
# utils = Utils()
def jwt_logged_user(token: Annotated[Optional[str], Depends(utils.OAuth2_scheme)],
db: Session = Depends(db_session)):
"""
处理 JWT 认证。
如果认证失败,返回 None 而非抛出异常。
"""
if token is None:
# 没有提供 JWT 令牌,或者令牌格式不正确 (由 auto_error=False 处理)
return None
try:
# 尝试验证令牌,并从 payload 中获取用户名
# 假设 util
s.verify_token 在验证失败时会抛出 HTTPException 或 JWTError
payload = utils.verify_token(token) # 假设返回一个包含用户信息的字典
username = payload.get("username") # 根据实际 payload 结构调整
if username is None:
return None # 令牌有效但没有找到用户名信息
# 查询数据库以验证用户是否存在
user = db.query(User).filter(User.username == username).first()
if user is None:
return None # 用户不存在
# 认证成功,返回用户名
return user.username
except HTTPException:
# 捕获 verify_token 内部抛出的 HTTPException
return None
except JWTError: # 捕获 jose 库的 JWT 错误
return None
except Exception:
# 捕获其他未知错误
return None关键点:
- utils.OAuth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False):确保 OAuth2PasswordBearer 实例设置了 auto_error=False。
- token: Annotated[Optional[str], Depends(utils.OAuth2_scheme)]:声明 token 为 Optional 类型。
- if token is None::检查是否提供了令牌。
- try...except 块:包裹 utils.verify_token 调用,捕获可能抛出的异常,并在捕获到异常时返回 None。
- return user.username:认证成功后返回用户名,与 Basic 认证的返回类型保持一致。
3. 实现组合认证逻辑
现在,当 basic_logged_user 和 jwt_logged_user 认证失败时都返回 None,我们可以创建一个新的组合依赖项来检查任一认证是否成功。
from fastapi import HTTPException, status, Depends
from typing import Annotated, Optional, Union
# ... 其他导入 ...
def auth_user(jwt_username: Annotated[Optional[str], Depends(jwt_logged_user)],
basic_username: Annotated[Optional[str], Depends(basic_logged_user)]) -> str:
"""
组合认证依赖,允许通过 JWT 或 Basic Auth 中的任一种进行认证。
如果任一认证成功,返回对应的用户名。
如果两种认证都失败,则抛出 401 异常。
"""
if jwt_username:
# JWT 认证成功
return jwt_username
if basic_username:
# Basic Auth 认证成功
return basic_username
# 如果两种认证都失败,则抛出未授权异常
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail='Invalid Credentials',
headers={"WWW-Authenticate": "Basic, Bearer"}, # 提示支持的认证方式
)
# 路由使用组合认证依赖
@router.get("/users/")
async def get_users(db: Session = Depends(db_session),
logged_username: Annotated[str, Depends(auth_user)]):
"""
获取用户列表,需要通过 JWT 或 Basic Auth 认证。
"""
# 假设您需要根据用户名进行某些操作,例如权限检查
print(f"Authenticated user: {logged_username}")
query_users = db.query(User).all()
return query_users关键点:
- jwt_username: Annotated[Optional[str], Depends(jwt_logged_user)] 和 basic_username: Annotated[Optional[str], Depends(basic_logged_user)]:接收两个子认证依赖的结果,它们现在可以是 None。
- if jwt_username: return jwt_username:如果 JWT 认证成功(返回了用户名),则直接返回该用户名。
- if basic_username: return basic_username:如果 JWT 认证失败但 Basic Auth 成功,则返回 Basic Auth 的用户名。
- raise HTTPException(...):只有当所有子认证都失败时,才由 auth_user 统一抛出 401 异常。这样确保了只有在没有任何有效凭据的情况下才拒绝访问。
- headers={"WWW-Authenticate": "Basic, Bearer"}:在 401 响应头中明确告知客户端支持的认证方式。
完整示例代码结构
为了更好地理解,以下是一个整合了上述修改的 FastAPI 应用片段:
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials, OAuth2PasswordBearer
from sqlalchemy.orm import Session
from typing import Annotated, Optional, Union
import secrets
from jose import JWTError, jwt # 假设使用 jose 库
# 假设这些是您的配置和模型
class Settings:
SESSION_LOGIN_USER = "admin"
SESSION_LOGIN_PASS = "securepassword"
JWT_SECRET_KEY = "your-super-secret-jwt-key" # 替换为实际的密钥
JWT_ALGORITHM = "HS256"
settings = Settings()
# 假设 db_session 是您的数据库会话依赖
# 假设 User 是您的 SQLAlchemy 用户模型
class User: # 简化示例
def __init__(self, username):
self.username = username
def __repr__(self):
return f"<User {self.username}>"
class MockDBSession: # 模拟数据库会话
def query(self, model):
return self
def filter(self, *args, **kwargs):
return self
def first(self):
return User("testuser") # 模拟返回一个用户
def all(self):
return [User("user1"), User("user2")]
def get_db():
db = MockDBSession()
try:
yield db
finally:
pass # 实际应用中会关闭会话
db_session = Depends(get_db)
# JWT 相关的工具类
class Utils:
OAuth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", auto_error=False)
SECRET_KEY = settings.JWT_SECRET_KEY
ALGORITHM = settings.JWT_ALGORITHM
def verify_token(self, token: str) -> dict:
try:
payload = jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise JWTError("Invalid token payload: no subject")
return {"username": username}
except JWTError as e:
# 在这里可以记录错误,但不再抛出 HTTPException
raise e # 重新抛出 JWTError,由 jwt_logged_user 捕获
except Exception as e:
raise e
utils = Utils()
router = APIRouter()
# --- 独立认证依赖项 ---
# Basic Auth
security = HTTPBasic(auto_error=False)
def basic_logged_user(credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)]) -> Optional[str]:
if credentials is None:
return None
current_username_bytes = credentials.username.encode("utf8")
correct_username_bytes = settings.SESSION_LOGIN_USER.encode("utf8")
is_correct_username = secrets.compare_digest(
current_username_bytes, correct_username_bytes
)
current_password_bytes = credentials.password.encode("utf8")
correct_password_bytes = settings.SESSION_LOGIN_PASS.encode("utf8")
is_correct_password = secrets.compare_digest(
current_password_bytes, correct_password_bytes
)
if not (is_correct_username and is_correct_password):
return None
return credentials.username
# JWT Auth
def jwt_logged_user(token: Annotated[Optional[str], Depends(utils.OAuth2_scheme)],
db: Session = Depends(db_session)) -> Optional[str]:
if token is None:
return None
try:
payload = utils.verify_token(token)
username = payload.get("username")
if username is None:
return None
user = db.query(User).filter(User.username == username).first()
if user is None:
return None
return user.username
except JWTError:
return None
except Exception:
return None
# --- 组合认证依赖项 ---
def auth_user(jwt_username: Annotated[Optional[str], Depends(jwt_logged_user)],
basic_username: Annotated[Optional[str], Depends(basic_logged_user)]) -> str:
if jwt_username:
return jwt_username
if basic_username:
return basic_username
raise HTTPException(
status_code=status.以上就是FastAPI 多种认证方式(任选其一)实现指南的详细内容,更多请关注其它相关文章!
# 而非
# 成都创意营销推广
# 营销推广怎么做到
# 兰州皋兰县网站建设平台
# 网络推广公司怎么做营销
# 淮北seo公司怎么选
# seo1加密线路1
# 越秀seo优化公司排名
# 承德抖音搜索seo
# 咸宁网站建设大全推广
# 忻州网站建设网络推广
# 自定义
# 客户端
# 两种
# word
# 多个
# 美图
# 文档
# 您的
# 令牌
# 抛出
# asic
# red
# 路由
# ai
# session
# 工具
# go
相关栏目:
【
科技资讯46185 】
【
网络学院92790 】
相关推荐:
网站内容防复制粘贴的实现策略与局限性
b站怎么取消点赞_b站点赞取消操作方法
React/Next.js中实现列表项的动态移动与状态管理:兼论唯一键的重要性
电脑IP地址怎么查 查看本机IP地址的几种方法
如何使用Node.js csv 包按条件移除含空字段的CSV记录
向日葵客户端怎么进行远程CentOS控制_向日葵客户端远程CentOS控制操作教程
聚水潭ERP登录页面入口 聚水潭ERP官网登录界面
正确连接J*aScript到HTML实现可点击图片与自定义事件处理
React项目中导航栏Logo自适应布局:避免裁剪与布局溢出
J*aScript生成器_j*ascript异步迭代
Safari自带网页翻译功能怎么用 无需插件轻松看懂外文网站【方法】
护手霜蹭到袖口上了如何清洗? 怎样避免留下一圈油印?
J*aScript中高效管理与清空动态列表:避免循环陷阱
Linux如何构建多环境配置管理_Linux多环境配置方案
离线运行Go语言之旅:本地部署与GOPATH配置指南
2025AO3夸克浏览器通道_AO3手机HTTPS安全入口分享
b站怎么看视频的弹幕数量_b站弹幕数量查看方法
微信怎么把收藏的内容分类管理 微信收藏内容标签分类方法
Golang如何实现简单的Web表单_Golang表单提交与验证处理方法
拼多多购物车商品数量无法修改如何处理 拼多多购物车操作优化方法
在Typer应用中优雅地处理和重组任意命令行参数
vivo手机参数配置怎么增强信号_vivo手机参数配置信号增强方法
机构:以往存储涨价周期小米利润率实际上有所改善 能转嫁给消费者等
谷歌浏览器如何快速清除某个网站的数据_Chrome网站缓存清理方法
TikTok网页版直接登录 TikTok网页端官方平台入口
谷歌google账号怎么注册账号 谷歌账号注册官方流程
如何使用spryker/configurable-bundles-products-resource-relationship模块解决复杂产品捆绑关系难题
在FastAPI中利用lifespan与依赖注入高效管理Redis连接池
Excel文件在线转换快速入口 Excel在线格式转换网站
零跑汽车11月交付量达70327台 实现连续9个月正增长
zookeeper 都有哪些功能?
MAC如何安全彻底地删除文件_MAC使用终端命令确保文件无法被恢复
Django模型中自动计算可用余额的实现方法
在J*aScript中复现SciPy的B样条拟合与求值:关键考量
C++ vector二维数组定义_C++ vector of vector用法
漫蛙网页登录入口 漫蛙漫画官方授权网址
J*a里如何实现订单支付与库存同步功能_支付库存同步项目开发方法说明
《刺客信条4:黑旗》重制版新细节曝光:无缝加载 地图更细致!
composer的"require-dev"部分是用来做什么的?
火锅吃太多会怎样 火锅吃太多会上火吗
ExcelARRAYTOTEXT函数怎么自定义分隔符输出数组文本_ARRAYTOTEXT实现动态生成SQL语句
如何高效处理PHP中的Excel数据导入导出?PortPHP/Spreadsheet助你轻松搞定!
夸克浏览器图书入口 夸克手机浏览器阅读入口
PySpark中从现有列右侧提取可变长度字符创建新列的教程
腾讯视频怎么使用多账号家庭管理_腾讯视频家庭多账号统一管理与权限分配教程
HuggingFaceEmbeddings中向量嵌入维度调整的限制与理解
AO3中文官网链接_AO3网页版稳定镜像站
解决Tabulator日期时间排序问题的专业指南
WordPress插件开发:正确注册卸载钩子与避免常见陷阱
在Go Martini框架中高效服务动态生成图像的实践指南


2025-11-25
浏览次数:次
返回列表
s.verify_token 在验证失败时会抛出 HTTPException 或 JWTError
payload = utils.verify_token(token) # 假设返回一个包含用户信息的字典
username = payload.get("username") # 根据实际 payload 结构调整
if username is None:
return None # 令牌有效但没有找到用户名信息
# 查询数据库以验证用户是否存在
user = db.query(User).filter(User.username == username).first()
if user is None:
return None # 用户不存在
# 认证成功,返回用户名
return user.username
except HTTPException:
# 捕获 verify_token 内部抛出的 HTTPException
return None
except JWTError: # 捕获 jose 库的 JWT 错误
return None
except Exception:
# 捕获其他未知错误
return None