FastAPI实战：从类型校验到JWT鉴权的生产级接口开发指南

1. 这不是又一个“Hello World”式FastAPI教程——它是一份能让你三天内独立交付接口服务的实战手记FastAPI这个词最近半年在后端开发圈子里出现的频率已经快赶上“Python”本身了。但你点开任意一个标着“FastAPI教程”的页面十有八九开头就是pip install fastapi uvicorn然后跑个uvicorn main:app --reload屏幕上跳出个“Hello, World”。接着——戛然而止。你合上网页心里只剩下一个问号然后呢我要写一个用户注册接口该加密码加密吗怎么连MySQL前端发来带token的请求我该怎么验跨域报错405是前端错了还是我漏配了什么这些真实项目里每天都在发生的、让新手卡住一整天的问题没人告诉你答案。这本不是FastAPI的错而是绝大多数“教程”根本没打算带你走出开发环境那扇门。我用FastAPI上线过6个生产级内部系统从日活300的HR审批流到并发2000的IoT设备数据中台踩过的坑比写的代码还多。这篇东西不讲抽象概念不堆术语只讲你打开PyCharm后接下来30分钟该敲哪几行、为什么这么敲、敲错会报什么错、报错后看哪一行日志最管用。它适合刚学完Python基础、想快速做出可用接口的人也适合从Django或Flask转过来、想搞清FastAPI到底“快”在哪、又“难”在哪的老手。如果你需要的是一个能直接复制粘贴、改几个变量就能跑通、还能应付真实业务需求的最小可行骨架那你来对地方了。下面所有内容都来自我去年重构三个遗留系统时的真实工作流——没有理论推导只有命令、配置、截图文字版和一句大实话“我当时就是这么干的结果成了。”2. FastAPI为何不是“又一个Web框架”它的设计哲学与真实优势拆解2.1 它不是靠“更快的异步”赢的而是靠“类型即契约”省下的沟通成本很多人第一次听说FastAPI是因为它“性能比Flask高”。这没错但只说对了10%。真正让它在团队协作中爆发价值的是类型注解驱动的自动文档与校验。我们来看一个真实场景前端同事发来一个接口需求文档写着“POST /api/v1/users参数name字符串必填、email邮箱格式、age整数18-99”。在Flask里你得手动写request.json.get(name)再自己判断if not name再用正则校验邮箱再写if not 18 age 99……写完还得在Swagger里手动补参数说明。而FastAPI只用这一段from pydantic import BaseModel, EmailStr from fastapi import FastAPI class UserCreate(BaseModel): name: str email: EmailStr age: int app FastAPI() app.post(/api/v1/users) def create_user(user: UserCreate): # user.name, user.email, user.age 已经是校验通过的干净数据 return {id: 123, name: user.name}你猜发生了什么自动校验如果前端传{name: , email: invalid, age: 15}FastAPI直接返回422错误并附带精确到字段的错误信息{detail: [{loc: [body, name], msg: field required, ...}]}自动文档访问/docsSwagger UI自动生成完整交互式文档每个参数类型、是否必填、示例值全都有前端点一下就能调试IDE智能提示你在create_user函数里敲user.PyCharm立刻弹出name、email、age三个属性不用翻文档查字段名。这背后不是魔法是Pydantic模型与Python 3.6类型注解的深度绑定。它把“接口契约”从口头约定、Word文档、Swagger手工维护变成了代码本身的一部分。一个新人接手项目看一眼UserCreate模型就知道这个接口要什么、不要什么、格式是什么——沟通成本直降70%。这才是FastAPI在企业级项目里站稳脚跟的根本原因远比UVLoop带来的那几十毫秒性能提升重要得多。2.2 “异步支持”不是噱头而是应对真实IO瓶颈的刚需方案FastAPI默认支持async/await但这绝不意味着你必须写异步代码。很多教程一上来就教async def结果新手写出一堆await asyncio.sleep(1)假装异步反而拖慢性能。真相是只有当你调用真正的IO阻塞操作时异步才有意义。比如查询MySQL数据库用aiomysql或asyncpg调用另一个HTTP API用httpx读写文件用aiofiles调用Redis用redis-py的async版本。而像JSON序列化、字符串拼接、内存计算这类CPU操作同步反而更快。我见过最典型的反模式是有人把所有路由都写成async def里面却全是json.loads()和hashlib.sha256()——这不仅没提速还增加了事件循环调度开销。正确的姿势是先写同步版本功能跑通找出真正耗时的IO操作比如数据库查询替换为异步驱动只在这些函数上加async/await其他逻辑保持同步。举个例子一个用户登录接口需要查数据库验证密码。同步写法是cursor.execute(SELECT * FROM users WHERE email%s, [email])这里IO阻塞换成await conn.fetchrow(SELECT * FROM users WHERE email$1, email)整个请求线程就释放了可以去处理其他请求。而密码校验bcrypt.checkpw()是CPU密集型必须用同步方式在async函数里直接调用即可。FastAPI的聪明之处在于它允许你混合使用同步和异步函数框架自动调度你不用操心线程池或事件循环——这才是工程落地的关键。2.3 CORS、中间件、依赖注入不是配置项而是架构分层的工具搜索热词里高频出现fastapi cors说明这是新手第一道坎。但CORS本身不是FastAPI的特性它是浏览器同源策略的产物。FastAPI提供CORSMiddleware本质是帮你往响应头里加Access-Control-Allow-Origin等字段。问题在于很多人配完中间件发现还是跨域失败。原因往往不在FastAPI而在前端发起的请求方式如果前端用fetch发Content-Type: application/json浏览器会先发一个OPTIONS预检请求如果后端没正确处理OPTIONS或者allow_origins没包含前端域名注意[*]在带凭证时无效就会失败。解决方案不是死磕中间件参数而是理解整个流程确认前端是否发送了credentials: true如带cookie若是则allow_origins必须指定具体域名不能用*且要加allow_credentialsTrueallow_methods建议明确列出[GET, POST, PUT, DELETE]避免[*]在某些代理下失效最关键一步检查Nginx/Apache反向代理是否转发了Origin头——很多线上问题根源在此。这引出了FastAPI另一个被低估的能力中间件与依赖注入的组合拳。比如你想记录每个请求的耗时传统做法是在每个路由里写start time.time()。FastAPI的依赖注入让你写一次全局生效from fastapi import Depends, Request import time async def log_time(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time print(fRequest {request.method} {request.url.path} took {process_time:.2f}s) return response app.middleware(http)(log_time) # 全局中间件而依赖注入更强大你可以把数据库连接、配置对象、认证服务都定义成可复用的依赖按需注入到路由中。比如# 依赖定义 async def get_db(): db DatabaseConnection() try: yield db finally: await db.close() # 路由使用 app.get(/users) def list_users(db: DatabaseConnection Depends(get_db)): return db.query(SELECT * FROM users)这实现了三层解耦路由只关心业务逻辑不关心DB怎么连DB连接管理独立封装测试时可轻松Mockget_db返回假数据。这才是现代Web框架该有的架构观而不是把所有东西揉进一个main.py里。3. 从零搭建一个可交付的FastAPI项目目录结构、核心配置与初始化流程3.1 拒绝单文件陷阱一个生产级项目的最小合理目录结构main.py里写完所有代码那是教程不是项目。真实项目必须考虑可维护性、可测试性、可部署性。我推荐的最小结构如下基于实际项目压缩删减了CI/CD等非核心部分my_fastapi_project/ ├── app/ # 应用主目录 │ ├── __init__.py │ ├── core/ # 核心配置与工具 │ │ ├── __init__.py │ │ ├── config.py # 配置加载环境变量、配置文件 │ │ ├── security.py # 认证、密码哈希、JWT生成 │ │ └── logger.py # 统一日志配置 │ ├── models/ # 数据库模型SQLModel或SQLAlchemy │ │ ├── __init__.py │ │ └── user.py # 用户模型定义 │ ├── schemas/ # Pydantic数据模型输入/输出契约 │ │ ├── __init__.py │ │ └── user.py # UserCreate, UserOut等 │ ├── api/ # API路由定义 │ │ ├── __init__.py │ │ ├── v1/ # 版本化路由 │ │ │ ├── __init__.py │ │ │ ├── users.py # /api/v1/users 相关路由 │ │ │ └── auth.py # /api/v1/auth 相关路由 │ ├── dependencies/ # 依赖注入定义 │ │ ├── __init__.py │ │ └── database.py # DB连接依赖 │ └── main.py # 应用入口创建FastAPI实例、挂载路由 ├── alembic/ # 数据库迁移SQLModel用此 ├── tests/ # 单元测试 ├── requirements.txt └── .env # 环境变量文件为什么这样分schemas/和models/分离schemas是API契约给前端看的models是数据库实体ORM映射两者字段可能不同如数据库存password_hashAPI只收password明文api/v1/版本化避免接口变更导致前端崩溃v2上线时v1仍可用core/放配置和工具所有模块都能导入from app.core.config import settings不用相对路径dependencies/显式声明依赖get_db这种关键服务集中管理便于统一监控和Mock。新手常犯的错是把main.py当垃圾桶所有东西往里塞。结果一个月后main.py变成2000行改一个字段要grep半小时。这个结构看似多了一堆文件夹但当你第3次修改用户注册逻辑时你会感谢当初花10分钟建好它。3.2 配置管理如何安全地读取环境变量与配置文件FastAPI教程里常写os.getenv(DATABASE_URL)这在本地开发没问题但上线后硬编码密钥是灾难。正确做法是分层配置 环境隔离。我用pydantic-settings原pydantic.BaseSettings升级版实现# app/core/config.py from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic import PostgresDsn from typing import Optional class Settings(BaseSettings): # 基础配置 PROJECT_NAME: str My FastAPI App API_V1_STR: str /api/v1 # 数据库配置 DATABASE_URL: PostgresDsn # JWT配置 SECRET_KEY: str ACCESS_TOKEN_EXPIRE_MINUTES: int 30 # CORS配置 BACKEND_CORS_ORIGINS: list[str] [http://localhost:3000] model_config SettingsConfigDict( case_sensitiveFalse, env_file.env, env_file_encodingutf-8 ) settings Settings()对应.env文件# .env DATABASE_URLpostgresqlasyncpg://user:passwordlocalhost:5432/mydb SECRET_KEYmy_super_secret_key_change_in_production BACKEND_CORS_ORIGINShttp://localhost:3000,https://myapp.com关键点解析PostgresDsn类型自动校验数据库URL格式错写成mysql://会启动时报错而不是运行时崩溃env_file.env指定配置文件case_sensitiveFalse让DATABASE_URL和database_url都有效model_config里的env_file_encoding解决中文注释乱码问题生产环境不放.env而是用Kubernetes ConfigMap或云服务商Secret Manager注入环境变量代码完全不变。提示永远不要把.env提交到Git在.gitignore里加一行*.env。我见过太多人把数据库密码推到GitHub公开仓库然后收到安全团队的紧急邮件。3.3 数据库集成用SQLModel实现零学习成本的ORM搜索热词里mysql安装教程、postgresql安装教程高频出现说明数据库连接是另一大痛点。FastAPI官方推荐SQLModelSQLAlchemy Pydantic因为它复用你已有的Pydantic知识无需新学ORM语法。步骤极简安装依赖pip install sqlmodel asyncpgPostgreSQL或pip install sqlmodel pymysqlMySQL定义模型app/models/user.pyfrom sqlmodel import SQLModel, Field, Relationship from typing import Optional, List class UserBase(SQLModel): email: str Field(uniqueTrue, indexTrue) name: str class User(UserBase, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) hashed_password: str class UserCreate(UserBase): password: str # API输入不存库 class UserOut(UserBase): id: int # API输出不含密码初始化数据库连接app/dependencies/database.pyfrom sqlmodel import create_engine, SQLModel from app.core.config import settings # 同步引擎用于创建表 engine create_engine(settings.DATABASE_URL.replace(asyncpg, )) # 异步引擎用于运行时查询 async_engine create_engine( settings.DATABASE_URL, echoTrue, # 开发时打印SQL上线关闭 pool_pre_pingTrue, # 连接前检测有效性 ) def init_db(): SQLModel.metadata.create_all(engine) # 创建表在main.py中初始化from app.dependencies.database import init_db app.on_event(startup) def on_startup(): init_db()为什么选SQLModel而非纯SQLAlchemyUser类同时是Pydantic模型可直接用于API输入/输出和SQLAlchemy模型可直接session.add()不用写两套类Field参数直接定义数据库约束uniqueTrue,indexTrue不用额外写__table_args__Relationship自动处理外键关联比手写ForeignKey直观得多。实测下来一个有3年Python经验的开发者2小时就能掌握SQLModel完成CRUD而学透SQLAlchemy Core可能要一周。3.4 接口开发实战从用户注册到JWT登录的完整链路现在把前面所有模块串起来写一个真实的用户注册→登录→获取个人信息流程。重点看如何把类型、依赖、配置、数据库全部拧成一股绳。第一步定义API输入输出模型app/schemas/user.pyfrom pydantic import EmailStr, BaseModel from typing import Optional class UserCreate(BaseModel): email: EmailStr name: str password: str class UserLogin(BaseModel): email: EmailStr password: str class UserOut(BaseModel): id: int email: EmailStr name: str # 不返回password_hash也不返回password第二步写认证工具app/core/security.pyfrom passlib.context import CryptContext from jose import JWTError, jwt from datetime import datetime, timedelta from app.core.config import settings pwd_context CryptContext(schemes[bcrypt], deprecatedauto) def verify_password(plain_password: str, hashed_password: str) - bool: return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) - str: return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: timedelta None) - str: to_encode data.copy() if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, settings.SECRET_KEY, algorithmHS256) return encoded_jwt第三步实现路由app/api/v1/users.pyfrom fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies.database import get_async_session from app.models.user import User from app.schemas.user import UserCreate, UserOut, UserLogin from app.core.security import get_password_hash, verify_password, create_access_token from sqlmodel import select router APIRouter() router.post(/register, response_modelUserOut) async def register_user( user_in: UserCreate, session: AsyncSession Depends(get_async_session) ): # 检查邮箱是否已存在 statement select(User).where(User.email user_in.email) result await session.execute(statement) existing_user result.scalar_one_or_none() if existing_user: raise HTTPException( status_codestatus.HTTP_400_BAD_REQUEST, detailEmail already registered ) # 创建新用户 user User( emailuser_in.email, nameuser_in.name, hashed_passwordget_password_hash(user_in.password) ) session.add(user) await session.commit() await session.refresh(user) return user router.post(/login) async def login_user( user_in: UserLogin, session: AsyncSession Depends(get_async_session) ): statement select(User).where(User.email user_in.email) result await session.execute(statement) user result.scalar_one_or_none() if not user or not verify_password(user_in.password, user.hashed_password): raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailIncorrect email or password ) access_token_expires timedelta(minutessettings.ACCESS_TOKEN_EXPIRE_MINUTES) access_token create_access_token( data{sub: user.email}, expires_deltaaccess_token_expires ) return {access_token: access_token, token_type: bearer}第四步挂载路由到主应用app/main.pyfrom fastapi import FastAPI from app.api.v1.users import router as users_router from app.api.v1.auth import router as auth_router app FastAPI( titleMy FastAPI App, descriptionA production-ready FastAPI service, version0.1.0, ) # 挂载API路由 app.include_router(users_router, prefix/api/v1, tags[users]) app.include_router(auth_router, prefix/api/v1, tags[auth]) # 添加CORS中间件 from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_originssettings.BACKEND_CORS_ORIGINS, allow_credentialsTrue, allow_methods[*], allow_headers[*], )现在运行uvicorn app.main:app --reload访问/docs你就能看到完整的Swagger文档点“Try it out”直接测试注册和登录。整个流程里你没写一行校验代码Pydantic搞定没手动拼SQLSQLModel搞定没处理跨域中间件搞定没管理JWTsecurity模块搞定。FastAPI的威力正在于把这些重复劳动用声明式的方式压缩到几行类型定义里。4. 避坑指南那些让新手卡住一整天的典型问题与排查技巧4.1 “ImportError: cannot import name XXX”——模块导入地狱的终极解法这是FastAPI新手最高频的报错形式千变万化cannot import name SQLModel,cannot import name AsyncSession,cannot import name Depends……根源只有一个Python路径混乱 包版本冲突。我总结出三步定位法确认当前Python解释器路径在终端执行which pythonMac/Linux或where pythonWindows确保你pip install的包装到了这个Python的site-packages里。常见陷阱用PyCharm创建虚拟环境但终端没激活pip install装到了系统Python里检查包版本兼容性FastAPI 0.110要求Pydantic 2.0而SQLModel 0.0.16要求SQLAlchemy 2.0。运行pip list | grep -E (fastapi|pydantic|sqlmodel|sqlalchemy)对照 官方兼容矩阵 不匹配就pip install sqlmodel0.0.16强制升级验证模块可导入性在Python交互式环境中逐行测试 import fastapi fastapi.__version__ 0.110.0 from fastapi import Depends # 成功则无输出 from sqlmodel import SQLModel # 失败则报错说明SQLModel没装对实操心得永远用pip install -e .项目根目录有setup.py或pip install -e ./path/to/app安装本地包而不是pip install .。前者创建符号链接代码改完立即生效后者打包安装改代码要重装。我曾因没用-e模式改了3小时代码没生效最后发现还在跑旧版本。4.2 “TypeError: object of type coroutine is not subscriptable”——异步函数调用的致命陷阱当你把一个async def函数当成普通函数调用时就会遇到这个错误。典型场景在同步路由里调用异步数据库函数。错误写法app.get(/users) def get_users(): users get_all_users() # get_all_users是async def这里返回coroutine对象 return users[0] # TypeErrorcoroutine不能索引正确写法有两种方案1推荐整个路由改为异步app.get(/users) async def get_users(): users await get_all_users() # 加await return users[0]方案2在同步函数里运行异步代码仅限必要场景import asyncio app.get(/users) def get_users(): users asyncio.run(get_all_users()) # 启动新事件循环 return users[0]但asyncio.run()在Web服务器里慎用Uvicorn本身就是一个事件循环asyncio.run()会创建新循环可能导致资源泄漏。所以原则是只要涉及IO操作整个调用链都走异步。FastAPI的Depends支持异步依赖get_async_session返回AsyncSession所有数据库操作都用await一气呵成。4.3 “CORS error: No Access-Control-Allow-Origin header”——跨域问题的七层排查清单前端报跨域别急着改FastAPI配置。按顺序检查这七层层级检查点命令/方法常见问题1. 浏览器控制台查看Network标签页点击报错请求看Response Headerscurl -I http://localhost:8000/api/v1/usersAccess-Control-Allow-Origin是否缺失2. FastAPI中间件检查allow_origins是否包含前端域名print(settings.BACKEND_CORS_ORIGINS)[*]在带credentials时无效3. Nginx/Apache检查反向代理配置是否透传Origin头curl -H Origin: http://localhost:3000 http://your-server/api/v1/users代理未加proxy_set_header Origin $http_origin;4. 前端请求检查fetch是否带credentials: include浏览器Network看Request Headerscredentials: include时后端必须配allow_credentialsTrue5. OPTIONS预检检查是否收到OPTIONS请求Uvicorn日志是否有OPTIONS /api/v1/users中间件未处理OPTIONS或allow_methods未包含OPTIONS6. Cookie域检查Set-Cookie的Domain是否匹配Response Headers看Set-Cookie: tokenxxx; Domainlocalhost前端域名是localhost:3000Cookie Domain应为localhost7. HTTPS混合检查前端是否HTTPS后端HTTP浏览器地址栏锁图标混合内容被浏览器拦截必须全站HTTPS我遇到过最隐蔽的问题前端用axios发请求withCredentials: true但后端allow_origins写了[http://localhost:3000]少了个s——http://vshttps://。浏览器严格匹配一个字符不对就跨域失败。所以永远用print()或日志输出request.headers.get(Origin)确认前端实际发来的Origin是什么。4.4 “OperationalError: server closed the connection unexpectedly”——数据库连接池的隐形杀手本地跑得好好的一上服务器就报数据库断连。这不是FastAPI的锅而是异步数据库连接池配置不当。asyncpg默认连接池大小是10如果并发请求超过10新请求会等待超时后报错。解决方案# app/dependencies/database.py from sqlalchemy.ext.asyncio import create_async_engine async_engine create_async_engine( settings.DATABASE_URL, echoFalse, pool_size20, # 连接池大小 max_overflow30, # 超出pool_size后最多创建的额外连接 pool_timeout30, # 获取连接超时秒数 pool_recycle3600, # 连接复用1小时后重建防长连接失效 )另外pool_pre_pingTrue很重要每次从连接池取连接前先发个SELECT 1检测是否存活。否则数据库重启后连接池里一堆僵尸连接第一个请求必然失败。注意事项pool_size不是越大越好。PostgreSQL默认max_connections100如果你的服务器有5个FastAPI实例每个pool_size20总连接数就100刚好打满。线上务必根据数据库配置调整留出余量给后台任务。4.5 “Pydantic validation error: value is not a valid email address”——前端传参格式的血泪教训Pydantic的EmailStr校验非常严格它要求邮箱必须符合RFC标准。但现实是前端传来的邮箱可能是userdomain缺顶级域名userdomain.com.末尾多点USERDOMAIN.COM大写EmailStr会拒绝前两种但接受第三种自动转小写。解决方案不是放宽校验而是前端规范 后端兜底前端用HTML5input typeemail浏览器自带基础校验后端在UserCreate模型里用自定义校验器from pydantic import validator class UserCreate(BaseModel): email: str name: str password: str validator(email) def email_must_contain_at(cls, v): if not in v: raise ValueError(Email must contain ) return v.lower() # 统一小写这样既保持校验又兼容现实数据。记住API契约是双向协议前端要守规矩后端也要有容错。5. 进阶实战如何将FastAPI与Vue3前后端分离项目无缝对接5.1 前端视角Vue3项目中调用FastAPI接口的最佳实践搜索热词里fastapi vue3、fastapi vue前后端分离高频出现说明这是主流架构。但很多教程只教后端不教前端怎么接。一个健壮的Vue3调用方案必须解决三件事请求封装、错误统一处理、Token自动注入。我用axiosPinia实现// stores/auth.ts import { defineStore } from pinia import axios from axios export const useAuthStore defineStore(auth, { state: () ({ token: localStorage.getItem(access_token) || , }), actions: { setToken(token: string) { this.token token localStorage.setItem(access_token, token) }, clearToken() { this.token localStorage.removeItem(access_token) } } }) // utils/request.ts import axios from axios import { useAuthStore } from /stores/auth const request axios.create({ baseURL: http://localhost:8000/api/v1, timeout: 10000, }) // 请求拦截器自动添加Token request.interceptors.request.use((config) { const authStore useAuthStore() if (authStore.token) { config.headers.Authorization Bearer ${authStore.token} } return config }) // 响应拦截器统一错误处理 request.interceptors.response.use( (response) response, (error) { if (error.response?.status 401) { // Token过期清空并跳转登录页 const authStore useAuthStore() authStore.clearToken() window.location.href /login } return Promise.reject(error) } ) export default request调用时!-- components/UserList.vue -- script setup langts import request from /utils/request const users ref([]) const loading ref(false) const fetchUsers async () { loading.value true try { const res await request.get(/users) users.value res.data } catch (error) { console.error(Failed to fetch users:, error) } finally { loading.value false } } /script关键点baseURL统一管理后端地址换环境只需改这里interceptors.request自动加Authorization头业务代码不用重复写interceptors.response捕获401自动登出避免每个API调用都写判断localStorage存Token比Vuex更持久刷新不丢失。5.2 后端加固JWT鉴权与权限分级的落地代码有了登录下一步是保护接口。FastAPI的Depends让鉴权变得极其简洁# app/api/v1/users.py from fastapi import Depends, HTTPException, status from jose import JWTError, jwt from app.core.config import settings from app.core.security import verify_token from app.schemas.user import UserOut # 依赖从Token解析用户 async def get_current_user( token: str Depends(oauth2_scheme) # oauth2_scheme是Security对象 ) - UserOut: credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailCould not validate credentials, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, settings.SECRET_KEY, algorithms[HS256]) email: str payload.get(sub) if email is None: raise credentials_exception except JWTError: raise credentials_exception # 这里应该查数据库获取用户简化为伪代码 user get_user_by_email(email) # 实际用SQLModel查 if user is None: raise credentials_exception return user # 用法在需要鉴权的路由上加 router.get(/me, response_modelUserOut) def read_users_me(current_user: UserOut Depends(get_current_user)): return current_user # 权限分级管理员专用接口 def get_current_active_admin( current_user: UserOut Depends(get_current_user) ) - UserOut: if not current_user.is_admin: # 假设User模型有