热门标签
热门文章
- 1[思维模式-17]：《复盘》-5- “行”篇 - 操作复盘- 项目复盘_复盘第五章项目复盘
- 2如何用python画一朵玫瑰花,用python画一朵玫瑰代码_玫瑰花python
- 3【Docker】在Windows操作系统安装Docker前配置环境_windows docker
- 4量化交易与人工智能：Python库的应用与效用_python量化交易库
- 5M1 Macbook Air使用VSCode配置C++环境_m1 cli vscode是什么版本
- 60day漏洞介绍
- 7基于YOLOV5的道路损伤（GRDDC‘2020）检测_道路缺陷d00
- 8如何搭建基于容器的工业互联网PaaS平台
- 9[转]SLA
- 10数据结构基础之图（中）：最小生成树算法_已知无向带权图,bfs序列和bfs生成树图解
当前位置: article > 正文
Python 高性能Web开发框架FastAPI精通指南_fastapi 写前端
作者：花生_TL007 | 2024-05-28 16:00:14
赞
踩
fastapi 写前端
文章目录
Web 开发应用模式
前后端不分离
核心概念：
- 前端页面看到的效果都是由后端控制，由后端渲染页面或重定向，也就是后端需要控制前端的展示，前端与后端的耦合度很高；
应用场景：
- 适用于小型项目或简单应用，如个人博客、公司网站等；
前后端不分离的应用模式比较适合纯网页应用，但是当后端对接App时，App可能并不需要后端返回一个HTML网页，而仅仅是数据本身，所以后端原本返回网页的接口不再适用于前端App应用，为了对接App后端还需再开发一套接口。
前后端分离
核心概念：
- 后端仅返回前端所需的数据，不再渲染HTML页面，前端通过Ajax调用后端的Restful API接口并使用JSON数据进行交互，实现前后端应用的解耦合；
应用场景：
- 适用于大型项目或复杂应用，需要前后端独立开发和维护；
在前后端分离的应用模式中，后端仅返回前端所需的数据，不再渲染HTML页面，不再控制前端的效果。至于前端用户看到什么效果，从后端请求的数据如何加载到前端中，都由前端自己决定，网页有网页的处理方式，App有App的处理方式，但无论哪种前端，所需的数据基本相同，后端仅需开发一套逻辑对外提供数据即可；
Web API 开发规范
API接口示例图
应用程序编程API接口（Application Programming Interface）
- 作为应用程序与外部世界交互的桥梁，其重要性不言而喻。
- 它可以是函数、类方法、URL地址或网络地址，但其核心目的始终是为客户端提供一个操作数据的入口。
- 当客户端调用这个入口时，应用程序会执行相应的代码操作，以满足客户端的需求。
关键问题
- 日常工作中我们既会调用他人编写的API接口，也会为他人提供API接口；
- 因此，如何编写清晰、易于维护的API接口成为了一个关键问题；
- 为了解决这一问题，开发团队需要遵循一套明确的API编写规范。
REST，全称为Representational State Transfer
- 是一种软件架构风格，尤其适用于前后端分离的应用模式。
- RESTful API强调面向资源开发，使用不同的HTTP方法（如GET、POST、PUT、DELETE）来操作资源，使得API的设计更加直观和易于理解。
RESTful风格
- URL路径代表了要操作的数据资源，而HTTP方法则表达了对这些资源的操作方式。
- GET方法用于获取资源
- POST方法用于创建资源
- PUT方法用于更新资源
- DELETE方法用于删除资源
- 这种设计风格使得API的使用更加直观和易于理解，降低了前后端之间的合作成本。
其他
- 除了RESTful风格外，RPC（Remote Procedure Call）也是一种常见的API实现规范。
- RPC允许客户端像调用本地方法一样调用远程服务器上的方法，从而简化了分布式系统的开发。
- 然而，RPC的缺点是可能会引入更多的复杂性和性能开销。
综上所述，无论是选择RESTful还是RPC作为API的实现规范，都需要根据具体的应用场景和团队需求来做出决策。
但无论选择哪种规范，都需要确保API的设计清晰、易于维护，并遵循一套明确的编写规范，以确保团队内部形成共识并减少个人习惯差异引起的混乱。
以webService为例通俗解释
非Rest设计，以往我们都会这么写： http://localhost:8080/admin/getUser （查询用户） http://localhost:8080/admin/addUser （新增用户） http://localhost:8080/admin/updateUser （更新用户） http://localhost:8080/admin/deleteUser （删除用户） 总结：以不同的URL（主要为使用动词）进行不同的操作。 Rest架构： GET http://localhost:8080/admin/user （查询用户） POST http://localhost:8080/admin/user （新增用户） PUT http://localhost:8080/admin/user （更新用户） DELETE http://localhost:8080/admin/user （删除用户） 总结：URL只指定资源，以HTTP方法动词进行不同的操作。用HTTP STATUS/CODE定义操作结果。
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
RESTful规范是一种通用的规范，不限制语言和开发框架的使用，也与技术无关；
事实上，我们可以使用任何一门语言，任何一个框架都可以实现符合restful规范的API接口。
Python Web 开发短板
- 性能略有挑战：相较于C/C++和Java等语言，Python在性能上略显逊色，这在一定程度上限制了它在处理大型、高并发Web应用方面的首选地位。
- 多线程处理受限：由于全局解释器锁（GIL）的存在，Python的多线程处理在性能上并不理想。GIL确保了同一时刻只有一个线程可以执行Python字节码，这在一定程度上限制了Python在并发处理上的能力，尤其是在CPU密集型任务中。
- 运行时错误的风险较高：**由于Python是一种动态类型语言，它允许在运行时进行类型检查和错误发现。**这意味着许多潜在的错误和类型不匹配问题可能不会在编写代码时立即显现，而是在程序运行时才被发现。这在Web开发中尤其是一个隐患，因为错误的类型可能导致页面崩溃、用户体验下降，甚至可能暴露敏感信息或导致安全问题。
- 缺乏成熟的异步库：尽管Python 3支持原生异步I/O，但在异步编程领域，与Node.js等语言相比，**Python的成熟异步库较少。**这增加了构建高性能异步应用的挑战，开发者往往需依赖第三方库或实现自定义异步功能。这提升了开发的复杂性和风险，突显了Python在异步编程生态方面仍有待进步。
- 部署过程较为繁琐：相较于PHP、Java等语言，Python的Web应用部署过程显得更为繁琐，这无疑为开发者带来了更高的技术挑战和更高的入门门槛。
尽管Python Web开发在某些方面存在短板，如性能上的挑战和部署流程的复杂性，但随着技术的不断进步，像FastAPI这样的现代化框架的涌现，为Python Web开发领域带来了显著的变革。
它们不仅优化了性能，还简化了部署流程，有效地解决了先前存在的许多问题，为开发者提供了更加高效、便捷的Web开发体验。
三大流行框架比较
Django、Flask、FastAPI是当下Python中最流行的Web开发框架，各自都拥有独特且迷人的魅力；
背景
- Django：作为Python Web框架中的老牌劲旅，自2005年诞生以来，Django凭借其“为完美者而生”的设计哲学，赢得了广泛的认可和应用。它提供了一套全面的解决方案，包括ORM、模板引擎、表单处理等，为用户带来了一站式的开发体验。
- Flask：自2010年问世以来，Flask凭借其微框架的身份，为开发者带来了更多的灵活性和简洁性。与Django不同，Flask并不提供全包式的功能，但它通过易于扩展的机制，使用户能够轻松增加所需的功能，满足各种定制化需求。
- FastAPI：作为后来居上的新兴框架，FastAPI于2018年崭露头角，它专注于提供高性能和快速开发的API服务。通过利用现代Python特性，如异步处理和类型提示，FastAPI实现了这一目标，成为了现代Web开发中的一股强劲力量。
设计
- Django：作为一个**全包式框架，Django致力于快速构建具有复杂数据模型的大型应用；**它内置了丰富的功能，如用户认证、内容管理等，为开发者提供了全面而强大的支持。
- Flask：是一个注重核心Web功能的轻量级框架，如路由和请求处理。它鼓励开发者使用扩展来灵活添加其他功能，如数据库操作、用户认证等，非常适合小到中型项目或作为微服务架构中的一部分。
- FastAPI：**专注于快速、高效的API开发，通过支持异步编程和自动数据验证，它显著提高了开发效率和性能。**此外，FastAPI还具备自动生成API文档的能力，并充分利用了现代Python特性（如类型提示），为开发者带来了更加便捷和高效的开发体验。
性能
- Django：虽然不是最快的框架，但对于大多数Web应用来说，其性能已经足够。
- Flask：性能比Django稍好，但在默认配置下，它不支持异步编程。
- FastAPI：在性能方面，FastAPI通常比Flask和Django表现得更好，这得益于其对异步编程的原生支持和高效的请求解析。
用途
- Django：该框架尤为适合那些需求快速开发的大型应用，特别是在构建内容管理系统（CMS）、电商平台以及社交网络等场景中，它能够提供丰富的内置功能，助力项目高效推进。
- Flask：对于小到中型项目、Web开发初学者、或是想要构建微服务架构以及作为某些特定应用的一部分来说，Flask框架无疑是一个理想的选择。其灵活性和简洁性使得它在这些场景中表现出色。
- FastAPI：对于那些需要构建高性能API的应用来说，FastAPI无疑是首选。无论是机器学习模型接口、高频交易平台，还是实时数据处理应用，FastAPI都能够凭借其高性能和快速开发的特点，满足这些场景下的严苛需求。
| / | FastAPI | Flask | Django |
|---|---|---|---|
| 发布 | 2018 | 2010 | 2005 |
| 设计理念 | 高性能API框架 | 小而精的微框架 | 大而全的全包式框架 |
| 特性 | 异步支持、自动数据验证、自动生成文档 | 轻量级、扩展性强 | ORM、认证系统、内置管理后台 |
| 性能 | 极高（与Go比肩） | 中等 | 中等 |
| 类型提示 | 原生支持 | 通过扩展支持 | 通过扩展支持 |
| 异步编程 | 原生支持 | 通过扩展部分支持 | Django 3.1+部分支持 |
| 自动文档 | 原生支持 | 通过扩展实现 | 通过扩展实现 |
| 适用场景 | 高性能API、机器学习接口、实时数据处理 | 小到中型项目、微服务 | 大型应用、CMS、电商平台 |
选择哪个框架，需综合考虑项目的具体需求、团队的技术栈以及期望的开发速度。
在现代API开发领域，FastAPI凭借其卓越的性能和高效的开发效率脱颖而出，成为众多开发者的首选；
而Flask则以其灵活性和简洁性，轻松应对多种类型的项目需求；
对于那些需要快速开发大型应用且希望框架提供多数必要功能的开发者来说，Django无疑是理想的选择；
综上所述，选择框架应基于项目的实际需求和技术团队的偏好，以确保项目的顺利进行和高效完成。
FastAPI 热度
| 框架 | Star | 开源地址 |
|---|---|---|
| django | 73.9k | https://github.com/django/django |
| flask | 64.9K | https://github.com/pallets/flask |
| fastapi | 64.3K | https://github.com/tiangolo/fastapi |
FastAPI 简介
FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，基于 Python 3.6+ 和标准 Python 类型提示。
FastAPI 的设计目标是让 API 开发变得更快、更简单，并且与当前 Python 开发的标准习惯更加契合。
FastAPI 的核心特性：
- 极高性能：使用了Starlette作为其底层框架，并基于Pydantic完成数据验证，使得可与NodeJS和Go并肩的极高性能；
- 类型提示：利用 Python 的类型提示功能来自动生成 API 文档（如 OpenAPI），并且进行请求和响应数据的验证。
- 高效编码：提高功能开发速度约 200％ 至 300％。
- 更少 bug：减少约 40％ 的人为（开发者）导致错误。
- 易于测试：支持使用 Pytest 等测试框架进行 API 的单元测试。
- 异步支持：支持异步编程，使得在高并发场景下性能更佳。
- 依赖注入：支持依赖注入，使得中间件、身份验证、数据库连接等变得更加容易管理。
- 标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI(以前被称为 Swagger) 和 JSON Schema。
FastAPI 依赖
依赖1：Python 3.6 + （负责编码开发）
- FastAPI充分利用了Python 3.6+版本的现代特性，尤其是类型提示功能，从而显著提升了开发效率。
- 这种类型提示不仅使代码更加清晰易读，还有助于静态分析工具准确检测潜在错误，进一步优化整体性能。
依赖2：Starlette（负责Web部分）
- FastAPI的卓越性能是建立在Starlette框架的坚实基础之上的。
- Starlette，作为一个轻量级且专为速度和性能打造的ASGI框架/工具集，为FastAPI提供了强大的异步请求处理能力。
- FastAPI自设计之初就全面支持异步编程，优化了底层架构和功能实现，使其在处理大量并发请求时具备极高的效率。
- 异步I/O操作避免线程阻塞，智能释放线程处理其他任务，实现服务器资源最大化利用，从而在高并发场景下表现卓越，提升服务器响应速度。
- 这种能力不仅确保了FastAPI在处理请求时的流畅性和高效性，更使得它在众多API框架中脱颖而出，成为了高性能特性的典范。
依赖3：Pydantic（负责数据验证部分）
- FastAPI巧妙地整合了Pydantic库，利用其强大的数据验证和管理设置功能。
- Pydantic基于Python的类型提示，为数据验证提供了快速且可靠的支持。
- 得益于Pydantic的高效数据处理能力，FastAPI能够迅速解析请求并验证数据的准确性，从而在确保性能的同时，极大地提升了开发效率和用户体验。
依赖4：第三方ORM（数据持久层）模块
- FastAPI没有内置ORM（对象关系映射）是因为它本身是一个轻量级的框架，专注于提供REST接口的优化和自动生成openapi文档等功能。
- 可以借助SQLAlchemy模块完成ORM（高版本支持异步）
- 可以借助Tortoise-orm模块完成ORM（原生支持异步）
FastAPI 安装
# fastapi框架
pip install fastapi
# 还会需要一个ASGI服务器，这里我选择的是Uvicorn
pip install uvicorn
- 1
- 2
- 3
- 4
- 5
FastAPI 快速启动
main.py文件
强调一下，FastAPI是完全支持异步的；
如果路由函数中，需要异步操作时（如数据库操作、调用外部API、耗时任务）等，需要使用await声明，并且函数需要使用async声明；
from fastapi import FastAPI # FastAPI 是一个为你的API接口提供了所有功能的Python类 app = FastAPI() # 这个实例将是创建你所有API接口的主要交互对象，app并不是固定变量名称 @app.get("/user") async def get_all_user(): return {"message": "获取所有用户"} @app.get("/user/{user_id}") async def get_user(): return {"message": "获取一个用户"} @app.post("/user") async def root(): return {"message": "新增一个用户"} @app.put("/user/{user_id}") async def update_user(): return {"message": "更新一个用户"} @app.delete("/user/{user_id}") def delete_user(): return {"message": "删除一个用户"}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
启动方式
| 参数 | 作用 |
|---|---|
| app | 运行的 py 文件:FastAPI 实例对象 |
| host | 访问url，默认 127.0.0.1 |
| port | 访问端口，默认 8000 |
| reload | 热更新，为true时有内容修改自动重启服务器 |
| debug | 同reload |
| reload_dirs | 设置需要 reload 的目录，List[str] 类型 |
| log_level | 设置日志级别，默认 info |
# 方式一 # 在命令行中执行uvicorn命令 # uvicorn：是一个ASGI服务器和工具，用于运行Python web应用程序 # main：是你实例化FastAPI()得到的app所在的py文件名称 # app： 是你实例化FastAPI()得到的app # 可以选择添加更多的命令 # --reload：告诉 Uvicorn 在检测到源文件发生变化时自动重新加载应用 # --host：指定服务器监听的IP地址 # --port：指定服务器监听的端口号 # --workers：指定用于处理请求的工作进程数 # --debug：是否启动调试模式 uvicorn main:app --port=8080 --reload # 方式二 # 在你实例化FastAPI()得到的app所在的py文件中使用__main__函数 if __name__ == '__main__': # app:这是要运行的 ASGI 应用的名称或实例 # host:指定服务器监听的 IP 地址 # port:指定服务器监听的端口号 # reload:如果设置为 True，则当源文件发生变化时，服务器会自动重新加载应用 # reload_dirs:一个目录列表，当这些目录中的文件发生变化时，服务器将重新加载应用 uvicorn.run(f"main:app", host="0.0.0.0", port=8080, reload=True) # INFO: Will watch for changes in these directories: ['/Users/...'] # INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) # INFO: Started reloader process [27454] using StatReload # INFO: Started server process [27456] # INFO: Waiting for application startup. # INFO: Application startup complete.
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
一旦FastAPI应用启动，它将自动生成Swagger UI界面，提供详尽的API文档，并允许用户直观探索API的所有路由和操作。
用户可点击API端点查看详细说明、请求参数和响应体，并在界面上直接测试API，提高API易用性和用户体验。
访问地址：http://127.0.0.1:8080/docs#/
FastAPI 使用流程：
- 安装FastAPI模块：使用pip命令安装FastAPI库，例如：
pip install fastapi。- 导入FastAPI模块：在Python脚本中导入FastAPI模块，例如：
from fastapi import FastAPI。- 创建FastAPI实例：使用导入的FastAPI类创建一个实例，例如：
app = FastAPI()。- 定义路由和视图函数：在FastAPI实例中定义路由和对应的视图函数，用于处理HTTP请求和返回响应。
- 运行FastAPI应用：使用Web服务器（如Uvicorn）来运行FastAPI应用，例如：
uvicorn main:app --reload。
FastAPI 路径参数
路径操作装饰器
fastapi支持各种请求方式
import uvicorn from fastapi import FastAPI app = FastAPI() @app.get("/get") def get(): return {"method": "get 方法"} @app.post("/post") def post(): return {"method": "post 方法"} @app.put("/put") def put(): return {"method": "put 方法"} @app.patch("/patch") def patch(): return {"method": "patch 方法"} @app.delete("/delete") def delete(): return {"method": "delete 方法"} @app.options("/options") def options(): return {"method": "options 方法"} @app.head("/head") def head(): return {"method": "head 方法"} @app.trace("/trace") def trace(): return {"method": "trace 方法"} if __name__ == '__main__': uvicorn.run(f"main:app", host="0.0.0.0", port=8080, reload=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
请求参数的docs交互文档1
请求参数的docs交互文档2
简单示例
# 步骤1：导入FastAPI
from fastapi import FastAPI
# 步骤2：创建一个FastAPI实例
app = FastAPI()
# 步骤3：创建一个路径操作：定义一个路径操作装饰器
@app.get("/")
async def root(): # 步骤4：定义路径操作函数
return {"message": "Hello World"} # 步骤5：返回内容
# 运行：uvicorn main:app —reload
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
说明：
- **路径：**get括号里引号里面的部分，这里是
/- **操作：**http方法，包括get，post等请求方式，这里是get；
@app.get("/")叫路径操作装饰器，它的作用是告诉下方的函数对应的路径'/’加上get操作；async def root()叫路径操作函数，表示当FastAPI接收一个使用GET方法访问URL/的请求时会调用这个函数；- 返回的内容可以是dict，list，或者Pydantic模型；
路径参数
FastAPI中的路径参数是URL中用于标识资源的变量，通过花括号在路由路径中定义，并在视图函数中接收；
例如/users/{user_id}中的{user_id}就是路径参数用于获取特定用户信息。
使用路径参数能提高API的明确性和灵活性，使URL更具描述性并允许客户端通过更改URL请求不同资源，无需修改API底层逻辑。
可以使用标准的 Python 类型标注为函数中的路径参数声明类型
@app.get("/items01/{item_id}")
async def read_item(item_id):
return {"item_id": item_id}
- 1
- 2
- 3
没有限制item_id字段的类型，则什么数据都可以传递
带类型的参数路径
# 指定参数类型，仅可接受“整数”或可以转换整数的字符串
@app.get("/items02/{item_id}")
async def read_item(item_id: int): # item_id字段仅接受int类型
return {"item_id": item_id}
- 1
- 2
- 3
- 4
限制item_id字段为int类型，则只接受int类型的数据或可以转换为int的字符串数据
数据校验
如果上述视图函数（items02）通过浏览器访问则会看到一个清晰可读的 HTTP 错误：
因为路径参数item_id传入的值是一二三，它不是且无法转换为int类型；
如果提供的是float类型的数据，也会出现同样的错误；
所以，通过Python的标准类型声明，FastAPI提供了数据校验的功能。
另外，请注意，FastAPI清晰直观的反馈错误详情，这对调试工作非常有用。
{
"detail": [
{
"type": "int_parsing",
"loc": [
"path",
"item_id"
],
"msg": "Input should be a valid integer, unable to parse string as an integer", "input": "一二三",
"url": "https://errors.pydantic.dev/2.6/v/int_parsing"
}
]
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
路由顺序
在构建视图时，经常会发现一些略微尴尬的问题，比如有些情况下路径是固定的；
例如，/user/{user_id}，用于获取指定用户的数据；
然后，/user/me，用于获取当前用户的数据；
问题，这样能够执行吗？由于路径操作是按照顺序以此执行的，/user/me将无法触发；
需要确保**/user/me声明在前，/user/{user_id}**声明在后；
@app.get("/user/{user_id}")
async def get_user(user_id: int):
return {"user": f"id={user_id}的用户信息"}
@app.get("/user/me")
async def get_me():
return {"user": "当前登录的用户信息"}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
函数位置调换前
函数位置调换后
枚举值参数
如果你有一个接收路径参数的路径操作，但你希望预先设定可能的有效参数值，则可以使用标准的 Python Enum 类型
class Gender(str, Enum):
m = "男"
w = "女"
o = "未知生物"
@app.get("/gender/{gender}")
async def get_model(gender: Gender):
if gender is gender.m:
return {"message": "大帅哥"}
if gender is gender.w:
return {"message": "小仙女"}
return {"message": "不可思议"}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
预设值
包含路径的参数
假设你有一个路径操作，它的路径为**/files/{file_path}。
但是你需要 file_path 自身也包含路径，比如 home/johndoe/myfile.txt。
因此，该文件的URL将类似于这样：/files/home/johndoe/myfile.txt**。
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
return {"file_path": file_path}
- 1
- 2
- 3
- 4
- 5
FastAPI 查询参数
声明不属于路径参数的其他函数参数时，它们将被自动解释为”查询字符串”参数
查询参数是URL的查询字符串部分，用于传递额外的信息或筛选条件。
它们以
?开头，后面跟着参数名和参数值，多个参数之间用&连接。例如，在以下URL中：
http://127.0.0.1:8000/items/?skip=0&limit=10
查询参数为：
- **skip：**对应的值为0
- **limit：**对应的值为10
# 利用faker生成虚拟数据
from faker import Faker
city_names = []
for i in range(50):
fake = Faker("zh-CN")
city_name = fake.city_name()
if city_name in city_names: continue
city_names.append(city_name)
@app.get("/city")
async def read_item(skip: int = 0, limit: int = 10):
return city_names[skip: skip + limit]
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
默认值
由于查询参数不是路径的固定部分，因此它们是可以是可选的，并且可以有默认值。
在上面的示例中，它们具有skip=0和limit=10的默认值
因此，访问下面两个URL，请求的资源是相同的；
http://127.0.0.1:8000/items/http://127.0.0.1:8000/items/?skip=0&limit=10但是，如果访问的下面这个URL，请求的资源则会不同：
http://127.0.0.1:8000/items/?skip=20
必选和可选参数
当为非路径参数声明了默认值，则该参数就是可选参数，否则就是必选参数；
@app.get("/items/{item_id}")
async def read_item(item_id: str, name: str, q: Union[str, None] = None):
data = {"item_id": item_id, "name": name}
if q:
data["q"] = q
return data
return data
- 1
- 2
- 3
- 4
- 5
- 6
- 7
在这个示例中，函数参数name是必选的，而函数参数q是可选的，并且默认值为None；
多个路径和查询参数
无论是路径参数还是查询参数都是可以声明多个的，也可以同时混用；
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
user_id: int,
item_id: str,
q: Union[str, None] = None,
short: bool = False
):
item = {"user_id": user_id, "item_id": item_id}
if q:
item.update({"q": q})
if not short:
item.update(
{"description": "short is false"}
)
return item
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
FastAPI 路由蓝图
项目结构
在main文件（主FastAPI实例文件）同级目录下创建N个不同子app应用
apps/app01/urls.py
from fastapi import APIRouter
shop = APIRouter()
@shop.get("/food")
def shop_food():
return {"shop": "food"}
@shop.get("/bed")
def shop_food():
return {"shop": "bed"}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
apps/app02/urls.py
from fastapi import APIRouter
user = APIRouter()
@user.post("/login")
def user_login():
return {"user": "login"}
@user.post("/reg")
def user_reg():
return {"user": "reg"}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
main.py
import uvicorn
from fastapi import FastAPI
from apps.app01.urls import shop
from apps.app02.urls import user
# 主应用路由（FastAPI实例）
app = FastAPI()
# 子应用路由（APIRouter实例）
app.include_router(router=shop, prefix="/shop", tags=["购物中心接口"])
app.include_router(router=user, prefix="/user", tags=["用户中心接口"])
if __name__ == '__main__':
uvicorn.run(f"main:app", port=8080, reload=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
Swagger/OpenAPI交互文档
路由蓝图下的docs文档下有两个应用
在 FastAPI 中，
include_router是一个函数，用于将一个或多个路由器（router）包含（或挂载）到父路由器或主应用中。这允许你构建更大的 API，同时保持代码的模块化和可读性。
include_router函数的主要参数如下：
- router: 这是你要包含的路由器实例。你可以传递一个或多个路由器实例。
- prefix: 一个字符串，用于为路由器中的所有路由添加一个前缀。例如，如果你的路由器包含
/items/路由，并且你为其设置了一个prefix为/api，则最终路由为/api/items/。- tags: 一个字符串或一个字符串列表，用于标记路由器中的所有路由。这些标签将出现在自动生成的 API 文档中，用于对路由进行分组和描述。
- responses: 一个字典，用于为路由器中的所有路由设置默认的响应模型。键是 HTTP 状态码，值是 Pydantic 模型。
- dependencies: 一个列表，用于为路由器中的所有路由添加依赖项。这些依赖项通常是身份验证、权限或其他中间件函数。
- add_exception_handlers: 一个布尔值，用于确定是否应为路由器中的所有路由添加默认的异常处理程序。
- add_route_operations: 一个布尔值，用于确定是否应为路由器中的所有路由添加默认的 HTTP 方法操作。
FastAPI 字段工厂
Query
Query是用于定义查询字符串参数的工厂类。它的主要参数有：
- default: 参数的默认值。如果请求中没有提供该参数，将使用此默认值。
- description: 参数的描述，通常用于文档和API接口描述。
- max_length: 参数的最大长度。如果提供的参数超过这个长度，将引发一个验证错误。
- min_length: 参数的最小长度。如果提供的参数短于这个长度，将引发一个验证错误。
- regex: 用于验证参数的正则表达式。
- alias: 参数的别名，允许在请求中使用不同的名称来传递此参数。
- ge: 大于或等于某个值。
- gt: 大于某个值。
- le: 小于或等于某个值。
- lt: 小于某个值。
- multiple: 允许参数在查询字符串中出现多次，返回一个列表。
- dependency: 用于创建更复杂的依赖关系，通常与
Depends结合使用。
from fastapi import FastAPI, Query from typing import List app = FastAPI() -------------------- # 使用Query工厂类限制查询参数字段 @app.get("/items/") async def read_items( # default=None 代表非必选参数，默认值为None值 q: str = Query(None, description="用于筛选项目的搜索查询", max_length=50, min_length=3, alias="搜索查询"), size: int = Query(10, description="返回的数据条数", ge=1, le=100), skip: int = Query(0, description="跳过的数据条数", ge=0), multiple_param: List[str] = Query(None, description="物品清单", multiple=True) ): """ 模拟根据查询参数从数据库中读取项目 """ # 这里只是模拟了响应，实际应用中应该根据参数查询数据库 results = { "query": q, "size": size, "skip": skip, "multiple_param": multiple_param } return results # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
Field
Field通常用于 Pydantic 模型中定义字段，但在 FastAPI 中也可以用于查询参数。它的参数与Query类似，但主要用于模型字段的描述和验证。
- default: 字段的默认值。
- description: 字段的描述。
- max_length: 字段的最大长度。
- min_length: 字段的最小长度。
- pattern: 用于验证字段的正则表达式。
- alias: 字段的别名。
- title: 字段的标题，通常用于文档和模型描述。
- ge: 大于或等于某个值。
- gt: 大于某个值。
- le: 小于或等于某个值。
- lt: 小于某个值。
from fastapi import FastAPI # 确保你已经安装了fastapi和pydantic库 from pydantic import BaseModel, Field app = FastAPI() class Item(BaseModel): # default=... 代表必填 name: str = Field(default=..., title="物品名称", max_length=100, min_length=3, description="必须仅包含字母", pattern=r'^[a-zA-Z]+$') quantity: int = Field(default=..., ge=0, description="物品数量，长度必须为非负数") price: float = Field(default=..., gt=0, le=1000, description="物品价格，必须大于0且小于等于1000") description: str = Field(default=None, max_length=500, min_length=10, description="物品的详细描述") # 使用pydantic模块限制查询参数字段 @app.post("/items/") async def create_item(item: Item): """ 模拟根据提供的数据创建新物品保存到数据库 """ # 在实际应用中，这里可能会将item保存到数据库 return {"message": "Item created successfully", "item": item.dict()} # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
Path
Path用于定义路径参数。它的参数包括：
- default: 参数的默认值。
- description: 参数的描述。
- max_length: 参数的最大长度。
- min_length: 参数的最小长度。
- pattern: 用于验证参数的正则表达式。
- alias: 参数的别名。
- ge: 大于或等于某个值。
- gt: 大于某个值。
- le: 小于或等于某个值。
- lt: 小于某个值。
from fastapi import FastAPI, Path app = FastAPI() # 使用Path工厂类限制路径参数 @app.get("/items/{item_id}") async def read_item( # Path只能够限制路径参数，不可以限制查询参数 # item_id 是一个必需的整数参数 item_id: int = Path(default=..., description="限制路径参数", max_value=100, min_value=1, ge=1, gt=0, alias="item-id"), ): """ 模拟根据物品ID和可选子物品ID读取项目。 """ results = { "item_id": item_id, } # 在实际应用中，这里可能会根据参数查询数据库并返回结果 return results # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
Form
Form是一个工厂类，用于处理表单数据，通常用于处理POST请求中的表单数据。
- **default：**默认值。如果表单中没有包含该字段，将使用此默认值。
- **description：**参数的描述，用于API文档生成。
- **alias：**参数的别名。允许你指定一个不同于字段名的名称来接收表单数据。
- **max_length：**字符串字段的最大长度。
- **min_length：**字符串字段的最小长度。
- **gt：**字段值必须大于指定的值。
- **ge：**字段值必须大于等于指定的值。
- **lt：**字段值必须小于指定的值。
- **le：**字段值必须小于等于指定的值。
- **pattrn：**字段值必须匹配指定的正则表达式。
from fastapi import FastAPI, Form app = FastAPI() # 使用Form工厂类，发起HTTP请求的表单数据 @app.post("/submit-form/") async def submit_form( name: str = Form(default=..., max_length=50, min_length=3, description="姓名"), email: str = Form(default=..., max_length=100, description="邮箱"), password: str = Form(default=..., min_length=8, description="密码", pattern=r"^[a-zA-z].*"), topics: List[str] = Form(default=None, description="兴趣", max_items=5, min_items=1) ): """ 模拟提交一份包含姓名、电子邮件、密码和主题的表单数据 """ # 处理表单数据... return { "name": name, "email": email, "password": password, # 注意：实际应用中不应返回密码 "topics": topics } # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
Cookie
- **default：**默认值。如果请求的Cookie中没有找到指定的键，则使用此默认值。
- **alias：**别名。允许使用不同的名字来引用同一个Cookie。
- **description：**描述。用于API文档生成，描述该Cookie的用途或含义。
- **encode：**是否对值进行URL编码。默认为
True，通常不需要修改。- **decode：**是否对值进行URL解码。默认为
True，通常不需要修改。请注意，encode和decode参数在FastAPI的官方文档中并不直接作为
Cookie依赖项的参数。通常情况下，FastAPI会自动处理URL编码和解码，所以通常不需要显式设置这些参数。
from fastapi import FastAPI, Cookie app = FastAPI() @app.get("/get-cookie") async def get_cookie( cookie_id: str = Cookie(default=None, description="用户ID"), ): """ 模拟从Cookie中获取user_id信息 """ return { "user_id": cookie_id, } # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
Header
- 含义如Cookie一致
from typing import Union from fastapi import FastAPI, Header app = FastAPI() @app.get("/header/") def header(user_agent: Optional[str] = Header(None, convert_underscores=True), x_token: List[str] = Header(None)): """ 有些HTTP代理和服务器是不允许在请求头中带有下划线的，所以Header类提供类convert_underscores属性让设置 :param user_agent: convert_underscores=True 会把 user_agent 转换成 user-agent :param x_token: 包含多个token的列表 :return: """ return {"User-Agent": user_agent, "x-token": x_token} # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
UploadFile
UploadFile是一个类，用于处理文件上传，它提供了比File更多的功能，例如直接访问文件内容、保存文件到磁盘等。
- filename: 上传文件的名称。
- content_type: 上传文件的内容类型。
- file: 文件对象本身。
- headers: 文件的相关 HTTP 头信息。
FastAPI 请求模型
path:
- 参数含义：定义了一个路由的路径，即客户端发送请求时需要访问的 URL 路径。
- 示例中值：
/items，意味着当客户端发送 POST 请求到/items路径时，会触发下面的函数处理该请求。tags:
- 参数含义：用于对 API 文档中的端点进行分组。通过给端点分配标签，可以在 API 文档中更方便地组织和管理 API。
- 示例中值：
["这是items的测试接口"]，意味着在生成的 API 文档中，这个端点会出现在一个名为“这是items的测试接口”的标签下。summary:
- 参数含义：为 API 文档中的端点提供一个简短的概述。这通常是一个简短的描述，用于说明该端点的功能。
- 示例中值：
"关于items测试的概述"，为 API 文档中的这个端点提供了一个简短的描述。description:
- 参数含义：为 API 文档中的端点提供更详细的描述。这通常用于解释端点的用途、请求参数、预期行为等。
- 示例中值：
"关于items测试的请求详情描述"，为 API 文档中的这个端点提供了更详细的描述。response_description:
- 参数含义：为 API 文档中的端点提供关于响应的详细描述。这有助于用户了解他们可以期待从端点获得什么样的响应。
- 示例中值：
"关于items测试的响应详情描述"，为 API 文档中的这个端点提供了关于响应的详细描述。这些参数都是用于生成 API 文档的，它们帮助开发者提供关于 API 端点的清晰和有用的信息，同时也帮助 API 的使用者理解和使用这些端点。
import uvicorn from fastapi import FastAPI app = FastAPI() # 暂且说明“请求”时用的参数 # tags 交互文档的标题 # summary 概述 # description 请求详情描述 # response_description 响应详情描述 @app.post( path="/items", tags=["这是items的测试接口"], summary="关于items测试的概述", description="关于items测试的请求详情描述", response_description="关于items测试的响应详情描述" ) def items(): return {"items": "items 数据"} if __name__ == '__main__': uvicorn.run(f"05 路径操作装饰器方法的参数:app", port=8080, reload=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
FastAPI 请求体
嵌套模型
from fastapi import APIRouter from pydantic import BaseModel, Field, field_validator from datetime import date from typing import List, Union, Optional app03 = APIRouter() class Addr(BaseModel): province: str city: str class User(BaseModel): name: str age: int = Field(default=18, ge=18, le=60) friends: List[int] # 列表中的元素要求必须全是int类型 birthday: Optional[date] = None description: Union[str, None] = None addr: Addr @field_validator("name") def name_must_alpha(cls, value): assert str(value).isalpha(), "name must be alpha" return value class Data(BaseModel): data: List[User] @app03.post("/user") async def data(user: User): print(type(user), user) # <class 'apps.app03.User'> name='string' age=0 birthday=datetime.date(2024, 3, 15) friends=[0] print(user.name) # zhangsan print(user.age) # 18 print(user.dict()) # {'name': 'zhangsan', 'age': 18, 'birthday': datetime.date(2024, 3, 15), 'friends': [1, 2, 3]} print(user.json()) # {"name":"zhangsan","age":18,"birthday":"2024-03-15","friends":[1, 2, 3]} return user @app03.post("/data") async def data(data: Data): return data
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
在这种情况下，FastAPI将期望得到这样的请求体：
{
"name": "string",
"age": 18,
"friends": [
0
],
"birthday": "2024-03-17",
"description": "string",
"addr": {
"province": "string",
"city": "string"
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
Form表单
@app.post("/regin")
async def regin(username: str = Form(...), password: str = Form(...)):
print(f"username:{username}, password:{password}") # username:admin, password:123123
return {"username": username}
# 请求信息
# curl -X 'POST' \
# 'http://127.0.0.1:8080/app04/regin' \
# -H 'accept: application/json' \
# -H 'Content-Type: application/x-www-form-urlencoded' \
# -d 'username=admin&password=123123'
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
上传文件
# file:bytes = File():适合小文件上传 @app.post("/files/") async def create_file(file: bytes = File()): print("file:", file) return {"file size": len(file)} @app.post("/multiFiles/") async def create_files(files: List[bytes] = File()): return {"file sizes": [len(file) for file in files]} # file:UploadFile:适合大文件上传 @app.post("/uploadFile/") async def create_upload_file(file: UploadFile): with open(f"{file.filename}", 'wb') as f: for chunk in iter(lambda: file.file.read(1024), b''): f.write(chunk) return {"filename": file.filename} @app.post("/multiUploadFiles/") async def create_upload_files(files: List[UploadFile]): return {"filenames": [file.filename for file in files]}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
Request对象
有些情况下我们希望能直接访问请求(Request)对象。
例如我们在路径操作函数中想获取客户端的IP地址，需要在函数中声明请求(Request)类型的参数，FastAPI就会自动传递请求(Request)对象给这个参数，我们就可以获取到请求(Request)对象及其属性信息，例如标题、网址、cookie、会话等。
from fastapi import APIRouter, Request app06 = APIRouter() @app06.get("/items") async def items(request: Request): return { "请求URL": request.url, "请求IP": request.client.host, "请求端口": request.client.port, "请求协议": request.get("type"), "请求方式": request.get("method"), "请求路由": str(request.get("route")), "协议版本": request.get("http_version"), "请求宿主": request.headers.get("user-agent"), "cookies": request.cookies, }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
静态文件
在Web开发中，需要请求很多静态资源文件（不是由服务器生成的文件），如css、js和图片文件等；
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
# 主应用路由（FastAPI实例）
app = FastAPI()
# # mount表示将某个目录下一个完全独立的应用挂载过来，这个不会在API交互文档中展示
app.mount("/static", StaticFiles(directory="/../static"))
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
FastAPI 响应模型
下载文件
FileResponse是用于发送文件的简单响应类。当你使用FileResponse时，FastAPI 会一次性读取整个文件内容并将其作为 HTTP 响应的主体发送给客户端。这意味着无论文件大小如何，服务器都会尝试将整个文件加载到内存中，然后再发送给客户端。主要参数及其含义
- path：要发送的文件的路径
- filename:允许你自定义客户端看到的文件名，而不是服务器上的原始文件名。这对于重命名文件或避免编码问题很有用。
- media_type:当你想明确指定文件的MIME类型时，可以使用这个参数。如果不指定，FastAPI会根据文件的扩展名自动推断MIME类型。
- background:允许你在文件发送过程中执行异步操作，例如记录日志、发送通知等
- content_disposition_type：这个参数通常用于控制浏览器是否应该直接打开文件还是提示用户下载。设置为
"attachment"通常会导致浏览器提示用户下载文件。
from fastapi import FastAPI from fastapi.responses import FileResponse import os app = FastAPI() # 利用FileResponse将小型文件当作HTTP响应 @app.get("/download_file1/{filename}") def download_file1(filename: str): # 配置文件路径，拼接文件名称filename file_path = os.path.join("./", filename) return FileResponse(file_path, filename=filename) # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
StreamingResponse允许你按需发送文件内容。它接受一个生成器函数或异步生成器函数作为参数，该函数负责生成要发送的数据块。这意味着你可以控制每次发送给客户端的数据量，而不是一次性发送整个文件。使用
StreamingResponse的一个关键点是，你需要定义一个生成器函数来产生文件的数据块。这个生成器函数应该在每次迭代时返回文件的一个小部分（即一个块），然后 FastAPI 会将这些块逐个发送给客户端。主要参数及其含义
- content:这是一个生成器函数或异步生成器函数，用于产生要发送给客户端的数据块。
ContentStream是一个可以迭代的对象，每次迭代返回一个数据块。- headers:一个字典，包含要添加到HTTP响应中的额外头信息。
- media_type:当你想明确指定数据的MIME类型时，可以使用这个参数。如果不指定，FastAPI会根据
content的类型尝试推断MIME类型。- background:允许你在数据发送过程中执行异步操作，例如记录日志、发送通知等。
需要注意的是，
StreamingResponse类通常用于复杂的场景，如实时数据流或大型文件下载，其中一次性加载整个内容到内存中可能不现实或效率低下。通过流式传输，你可以按需生成和发送数据，从而更有效地处理这些场景。
from fastapi import FastAPI from fastapi.responses import StreamingResponse import os app = FastAPI() # 利用StreamingResponse按需发送文件内容 def file_chunk_iter(file_path, chunk_size=1024): with open(file_path, "rb") as file: while True: chunk = file.read(chunk_size) if not chunk: break yield chunk @app.get("/stream-file/") def stream_file(file_path: str): return StreamingResponse(file_chunk_iter(file_path), media_type="application/octet-stream") # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
response_model
前面写的这么多路由函数，最终return的都是自定义结构的字典；
FastAPI提供了response_model参数，声明return响应体的模型。
# 路由装饰器
@app.get("/items", response_model=Item)
# 路由函数
async def create_item(item: Item):
pass
- 1
- 2
- 3
- 4
- 5
上述示例中，response_model是路由装饰器的参数，并不是路由函数的参数哦；
FastAPI将使用response_model进行以下操作：
- 将输出数据转换为response_model中声明的数据类型
- 验证数据结构和类型
- 将输出数据限制为该model定义的
- 添加到OpenAPI中
- 在自动文档系统中使用
示例
from fastapi import APIRouter from pydantic import BaseModel, EmailStr from typing import Union app07 = APIRouter() class UserIn(BaseModel): username: str password: str email: EmailStr full_name: Union[str, None] class UserOut(BaseModel): username: str email: EmailStr @app07.post("/createUser1") async def items(user: UserIn): # 业务逻辑处理... # 存入数据库... return user @app07.post("/createUser2", response_model=UserOut) async def items(user: UserIn): # 业务逻辑处理... # 存入数据库... return user
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
response_model_exclude_unset
response_model_exclude_unset是一个布尔值，用于控制是否排除未设置（None或未定义）的模型字段。当设置为True时，如果模型中的某个字段在返回的数据中未设置（为None），则该字段将不会包含在响应中。
from fastapi import APIRouter from pydantic import BaseModel from typing import Union, List app08 = APIRouter() class Item(BaseModel): name: str description: Union[str, None] = None price: float tax: float = 10.5 tags: List[str] = [] items = { "foo": {"name": "Foo", "price": 50.2}, "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2}, "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []}, } @app08.get("/itemOut/{item_id}", response_model=Item, response_model_exclude_unset=True) async def read_item(item_id: str): return items.get(item_id.lower())
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
response_model_exclude
排除哪些字段
response_model_include
包含哪些字段
依赖项
什么是依赖注入？
在FastAPI中，依赖注入（Dependency Injection, DI）是一种设计模式；
它允许你声明性地指定一个函数或方法执行时需要的其他对象或数据。FastAPI运行时，这些依赖项会被自动解析并注入到函数中。
如何理解依赖注入？
理解FastAPI中的依赖注入，可以从以下几个方面入手：
- 基于类型提示的依赖声明：在FastAPI中，你可以通过在函数参数上使用类型提示来声明依赖。例如，如果你有一个需要数据库会话的函数，你可以通过在该参数上使用
Depends和相应的类型提示来声明这个依赖。- 自动解析依赖：当FastAPI应用程序接收到一个请求时，它会分析请求路径、查询参数、请求体等信息，并自动解析和注入所有声明的依赖。这意味着你不需要手动创建和传递这些依赖，FastAPI会为你处理。
- 提高代码重用性和可维护性：通过依赖注入，你可以将通用的逻辑（如身份验证、数据库连接等）封装成独立的函数或类，并在多个路由处理函数中重复使用。这有助于减少代码重复，提高代码的可维护性。
- 支持异步编程：FastAPI的依赖注入系统完全支持异步编程，这意味着你可以轻松地处理并发请求，提高应用程序的性能和响应速度。
- 增强测试性：依赖注入使得单元测试变得更容易。你可以为函数提供模拟的依赖项，而不是实际的资源，从而在不连接到数据库或外部服务的情况下测试你的逻辑。
依赖注入的优势？
- 模块化和重用性：通过依赖注入，可以将功能封装成独立的模块或依赖项，并在多个地方重复使用。这有助于减少代码重复，提高代码的可维护性和可重用性。
- 解耦：依赖注入有助于减少组件之间的耦合度。通过将依赖关系显式地声明在代码中，可以更容易地替换或修改组件，而不需要修改其他组件的代码。这有助于保持代码的清晰度和可维护性。
- 易于测试：依赖注入使得单元测试变得更加容易。可以通过注入模拟对象（mock objects）来替换真实的依赖项，从而在不依赖于外部资源或服务的情况下测试代码。这有助于确保代码的正确性和健壮性。
- 灵活性和可扩展性：依赖注入系统允许在运行时动态地更改或替换依赖项。这使得应用程序更加灵活，可以根据不同的需求或场景进行定制。同时，这也为应用程序的扩展提供了便利，你可以轻松地添加新的依赖项或修改现有依赖项的行为。
Depends类
Depends是 FastAPI 中的一个核心功能，它提供了一种声明依赖项的机制，使得开发者可以在路由操作中定义依赖关系，并确保这些依赖项在处理请求之前被正确地解析和验证。Depends的概念类似于其他框架中的中间件或钩子，但它提供了更加灵活和强大的功能。
概念和作用：
Depends允许你定义一个函数，该函数返回一个值（通常是某个对象或数据），这个值可以在后续的路由操作中被使用。更重要的是，Depends还允许你指定依赖项是否必须存在，以及如何处理依赖项的解析失败。在 FastAPI 中，你可以使用
Depends来实现以下功能：数据验证和转换：
- 可以定义一个
Depends函数，用于验证和转换请求中的数据，确保它们符合预期的格式和类型。权限和身份验证：
- 可以使用
Depends来实现权限控制和身份验证逻辑。- 例如，可以定义一个
Depends函数来检查用户的身份和权限，如果验证失败，则返回相应的错误响应。数据库会话管理：
- 在处理数据库相关的请求时，可能需要确保每个请求都使用相同的数据库会话。
- 通过
Depends，可以定义一个函数来创建和管理数据库会话，并在每个路由操作中注入这个会话。请求头、查询参数等的处理：
- 可以使用
Depends来处理请求头、查询参数等，提取需要的信息并在路由操作中使用。
使用技巧
使用
Depends时，你需要遵循以下步骤：定义 Depends 函数：
- 首先，需要定义一个普通的 Python 函数，该函数返回你想要在路由操作中使用的值。
- 这个函数可以接收任意数量和类型的参数，以便从请求中提取所需的信息。
添加 Depends 到路由操作：
- 在定义路由操作时，需要将
Depends函数作为参数添加到装饰器中。- FastAPI 会自动处理这个依赖项，确保在处理路由操作之前先解析和验证这个依赖项。
处理依赖项解析失败：
- 如果
Depends函数的解析失败（例如，数据验证失败），你可以返回一个HTTPException异常来指定错误响应。- FastAPI 会捕获这个异常，并将其转换为相应的 HTTP 错误响应。
使用多个 Depends：
- 可以在同一个路由操作中使用多个
Depends函数，以定义多个依赖项。- FastAPI 会按照你在装饰器中指定的顺序来解析这些依赖项。
视图函数使用Depends依赖项
from fastapi import FastAPI, Depends, HTTPException, status from pydantic import BaseModel app = FastAPI() # 定义一个 Pydantic 模型用于数据验证 class User(BaseModel): username: str password: str # 定义依赖: 依赖项可以是任何可调用的对象，通常是函数或类 # 定义一个 Depends 依赖函数用于验证用户身份 def get_current_user(user: User) -> str: if user.username != "admin" or user.password != "password": raise HTTPException(status_code=401, detail="Incorrect username or password") return user.username # 注入依赖: 在路径操作函数中，你可以使用Depends依赖项作为参数来注入依赖。 # FastAPI会自动解析这些依赖项，并在调用路径操作函数之前执行它们。 # 定义一个路由操作，使用 Depends 标注进行依赖用户验证函数 @app.post("/items/") async def create_item(current_user: str = Depends(get_current_user)): return {"user": current_user} # 运行应用: 当请求到达对应的路径时，FastAPI会解析并调用定义的依赖函数，然后将返回的结果传递给路径操作函数。 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
使用装饰器中的dependencies参数声明依赖项
dependencies:
- 概念: 一个依赖项列表，这些依赖项会在处理请求之前被调用。它们通常用于认证、授权、数据验证等。
- 作用: 允许开发者在路由处理函数之前执行一些前置逻辑。
在 FastAPI 中，直接在视图函数参数中声明依赖项和使用dependencies参数在装饰器中声明依赖项之间有一些区别：
视图函数参数中声明依赖项
- 当依赖项可以直接从请求中提取数据（如路径参数、查询参数、请求头等）时，这种方式非常有用。FastAPI 会自动处理这些依赖项，并将解析后的值作为参数传递给视图函数。这种方式简洁明了，适用于那些每个请求都需要执行且没有特殊依赖顺序的依赖项。
装饰器中的dependencies参数声明依赖项
- 使用dependencies参数可以让你更灵活地控制依赖项的执行顺序和条件。这对于那些需要根据其他依赖项的结果来决定是否执行某个依赖项，或者需要按照特定顺序执行多个依赖项的场景非常有用。此外，dependencies参数还支持异步依赖项，这是直接在参数中声明依赖项所不支持的。
from fastapi import FastAPI, Header, Depends, HTTPException from pydantic import BaseModel, Field app = FastAPI() def get_current_user_token(token: str = Header(None)): if token == "admin": return {"username": "admin", "password": "admin", "token": "admin"} else: raise HTTPException(status_code=401, detail="只准输入admin哦，这是我们的秘密！") def get_current_user_level(level: str = Header(None)): if level == "admin": return {"level": "admin"} elif level == "guest": return {"level": "guest"} else: raise HTTPException(status_code=401, detail="出门在外身份都是自己给的！") @app.get("/users/me1", dependencies=[Depends(get_current_user_token), Depends(get_current_user_level)]) async def read_user_me1(): # dependencies内两个依赖项执行成功后进入视图函数 return {"message": "很好，很高兴见到你！"} @app.get("/users/me2") async def read_user_me2(user_info: dict = Depends(get_current_user_token), user_level: dict = Depends(get_current_user_level)): return_info = {"message": "很好，很高兴见到你！"} return_info.update(user_info) return_info.update(user_level) return return_info # 运行FastAPI应用 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8080)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
后台任务
FastAPI 的后台任务功能基于 Python 的异步编程模型，巧妙地实现了在处理 HTTP 请求的同时，异步处理那些不需要实时响应但需要在 API 请求完成后执行的任务。
这些任务涵盖了广泛的场景，如发送电子邮件、处理图像、执行复杂的数据分析等。
熟练运用 FastAPI 的后台任务功能，能够轻松地将耗时操作从 API 的主流程中抽离出来，从而极大地提高了 API 的响应性能。
这意味着，即使后台任务需要消耗一定的时间和计算资源，也不会对 API 的实时性能和用户体验产生任何影响。
示例代码
import time # 1.首先导入 BackgroundTasks from fastapi import FastAPI, BackgroundTasks, Depends from typing import Optional app = FastAPI() # 3.创建一个任务函数，它可以是普通函数也可以是异步函数，并允许携带参数； async def bg_task(framework: str): with open("README.md", mode="a", encoding="utf-8") as f: for i in range(1, 10): time.sleep(1) f.write(f"### 标题{i}") f.write(f"> 标题{i} 的内容：我是后台任务，{framework}\n") # 标准后台任务示例 @app.post("/background_tasks") # 2.在视图函数中使用类型声明BackgroundTasks定义一个参数 async def run_bg_task(framework: str, background_tasks: BackgroundTasks): """ :param framework: 被调用的后台任务函数的参数 :param background_tasks: FastAPI.BackgroundTasks :return: """ # 4.添加到后台任务队列 background_tasks.add_task(bg_task, framework) return {"message": "任务已在后台执行"} def continue_write_readme(background_tasks: BackgroundTasks, q: Optional[str] = None): if q: background_tasks.add_task(bg_task, f"\n> 我是 Depends 依赖函数，我要写入：{q}") return q # 依赖后台任务示例 @app.post("/dependency/background_tasks") async def dependency_run_bg_task(q: str = Depends(continue_write_readme)): if q: return {"message": "README.md更新成功"} return {"message": "没有任务"} # 启动应用程序 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
异常错误
FastAPI 提供了一个方便且灵活的方式来处理异常错误；
当路由函数或依赖项抛出异常时，FastAPI 会捕获这些异常，并将其转换为适当的 HTTP 响应；
FastAPI 还允许你自定义错误处理，以返回自定义的错误响应；
内置异常处理类
**HTTPException：**基础的 HTTP 异常类，可以用它来抛出自定义的 HTTP 错误。
**HTTPException：**的两个常用属性：
- **status_code：**可以设置或读取 HTTP 状态码
- **detail：**用于提供关于错误的详细信息
自定义异常处理
可以通过继承HTTPException类创建自定义异常，并在路由或依赖中抛出它们
注册异常处理器
FastAPI 允许你注册自定义的错误处理器，这样你就可以控制如何处理特定的异常。错误处理器是一个函数，它接受异常实例和请求对象，并返回一个响应对象。
from fastapi import FastAPI, HTTPException, status from fastapi.responses import JSONResponse app = FastAPI() # 自定义异常类 class ItemNotFound(HTTPException): def __init__(self, item_id: str): super().__init__( status_code=status.HTTP_404_NOT_FOUND, detail=f"Item {item_id} not found", ) # 错误处理器 @app.exception_handler(ItemNotFound) async def item_not_found_handler(request, exc: ItemNotFound) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content={"message": exc.detail} ) # 路由函数使用内置异常 @app.get("/items/{item_id}") async def read_item(item_id: int): items = {1: "foo", 2: "bar"} if item_id not in items: raise HTTPException(status_code=404, detail="Item not found") return {"item": items[item_id]} # 路由函数使用自定义异常 @app.get("/custom-error/{item_id}") async def read_custom_error(item_id: int): if item_id != 1: raise ItemNotFound(item_id) return {"item": "custom error item"} # 启动应用程序 if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
安全性
安全认证是什么？
FastAPI框架中的安全性认证是一种机制，用于验证和授权访问应用程序的用户。
它的核心概念包括身份验证（Authentication）和授权（Authorization）；
身份验证是确认用户身份的过程，而授权则是确定经过身份验证的用户是否有权执行特定操作的过程；
在FastAPI中，安全性认证通常通过依赖项（Dependencies）和中间件（Middlewares）来实现。
依赖项用于在路由操作之前执行身份验证和授权逻辑，而中间件则用于在请求处理过程中执行身份验证。当用户发送请求时，FastAPI应用程序会首先经过中间件进行身份验证，然后将用户身份信息传递给路由操作进行进一步的授权验证。
如果用户通过了所有验证步骤，则路由操作将继续执行并返回相应的响应；否则，将返回一个错误响应。
工作原理是什么？
工作原理：
- 当客户端发送请求到FastAPI应用程序时，应用程序会首先经过中间件。中间件可以检查请求中的身份验证信息（如令牌、Cookie等），并根据这些信息验证用户的身份。
- 如果用户通过了身份验证，中间件会将用户身份信息传递给路由操作。在路由操作中，可以定义依赖项来进一步验证用户是否有权执行该操作。
- 依赖项函数会在路由操作之前执行，并可以访问请求中的用户身份信息。依赖项可以根据需要执行额外的验证逻辑，例如检查用户角色或权限。
- 如果用户通过了所有验证步骤，路由操作将继续执行，并返回相应的响应。否则，可以返回一个错误响应，指示用户未通过身份验证或授权。
如何实现？
在FastAPI框架中，OAuth2PasswordBearer和OAuth2PasswordRequestForm共同实现了基于密码的OAuth2认证流程；
OAuth2PasswordBearer负责验证后续请求中的令牌；
OAuth2PasswordRequestForm则处理登录请求，接收用户名和密码，并可能生成令牌。
这两个组件协同工作，为FastAPI应用程序提供了安全性认证的功能。1.OAuth2PasswordBearer：
- 作用：声明安全性方案，指定客户端获取令牌的URL。
- 用途：
- 在API路由操作中作为依赖项，自动检查请求的
Authorization头部。- 如果请求没有提供有效的Bearer令牌，则返回401状态码（未授权）。
- 不负责实际的登录或令牌获取逻辑，这些需要开发者自己实现。
2.OAuth2PasswordRequestForm：
- 作用：从客户端请求的表单体中接收用户名和密码。
- 用途：
- 在登录路由中作为依赖项，接收和处理来自客户端的用户名和密码。
- 提取的用户名和密码可用于验证用户身份，并生成访问令牌。
OAuth2实现
from typing import Optional from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from pydantic import BaseModel app = FastAPI() """ OAuth2 密码模式 和 FastAPI 的 OAuth2PasswordBearer OAuth2PasswordBearer 是接受URL作为参数的一个类，客户端会响该URL发送username和password参数，然后得到一个token值 OAuth2PasswordBearer 并不会创建响应的URL路径操作，只是指明客户端用来请求Token的URL地址 当请求到来的适合，FastAPI会检查请求的Authorization头信息， 如果没有找到Authorization头信息，或者头信息中的内容不是Bearer token 它会返回401状态码（UNAUTHORIZED） """ """ 基于 Password 和 Bearer token 的 OAuth2 认证""" # 模拟数据库用户数据 fake_users_db = { "zhangsan": { "username": "zhangsan", "full_name": "Zhang San", "email": "zhangsan@example.com", "hashed_password": "fake_hashed_zhangsan", "disabled": False, }, "lisi": { "username": "lisi", "password": "Li Si", "email": "lisi@example.com", "hashed_password": "fake_hashed_lisi", "disabled": True, } } def fake_hash_password(password: str): return "fake_hashed_" + password # 用户基础数据模型 class User(BaseModel): username: str email: Optional[str] = None full_name: Optional[str] = None disabled: Optional[bool] = None # 用户输入数据模型 class UserInDB(User): hashed_password: str @app.post("/token") async def login(form_data: OAuth2PasswordRequestForm = Depends()): # 模拟检查用户名是否存在数据库 user_dict = fake_users_db.get(form_data.username) if not user_dict: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"用户名或密码不正确(用户名不存在)" ) # 模拟查询用户对应加密密码 user = UserInDB(**user_dict) # 模拟加密用户当前输入的明文密码 hashed_password = fake_hash_password(form_data.password) # 校验密码 if not hashed_password == user.hashed_password: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"用户名或密码不正确(密码输出错误)" ) return {"access_token": user.username, "token_type": "bearer"} # 它会请求（本地）http://127.0.0.1:8000/chapter06/token oauth2_schema = OAuth2PasswordBearer(tokenUrl="/token") @app.get("/oauth2_password_bearer") async def oauth2_password_bearer(access_token: str = Depends(oauth2_schema)): return {"access_token": access_token} def get_user(db, username: str): if username in db: user_dict = db[username] return UserInDB(**user_dict) def fake_decode_token(token: str): user = get_user(fake_users_db, token) return user async def get_current_user(token: str = Depends(oauth2_schema)): user = fake_decode_token(token) if not user: raise HTTPException( status_code=status.HTTP_200_OK, detail="身份验证失败", headers={"WWW-Authenticate": "Bearer"} # OAuth2 的规范，如果认证失败，请求头中返回WWW- ) return user async def get_current_active_user(current_user: User = Depends(get_current_user)): if current_user.disabled: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user" ) return current_user @app.get("/users/me") async def read_users_me(current_user: User = Depends(get_current_active_user)): return current_user if __name__ == '__main__': import uvicorn # 启动命令：uvicorn hello_world:app --reload uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
JWT实现
基于FastAPI和JSON Web Tokens (JWT) 实现的安全认证系统，这个系统允许用户通过用户名和密码登录，并在成功登录后获得一个JWT令牌。这个令牌可以用于后续请求的认证，以访问受保护的路由。主要组件和流程如下：
用户模型:
- User: 基础用户模型，包含用户名、电子邮件、全名和禁用状态。
- UserInDB: 继承自User，增加了hashed_password字段，用于存储用户的加密密码。
密码加密:
- 使用passlib的CryptContext，配置为使用bcrypt算法，用于密码的加密和校验。
用户认证:
- verify_password: 校验明文密码和加密后的密码是否匹配。
- jwt_authenticate_user: 使用用户名和密码进行用户认证。首先通过用户名获取用户对象，然后校验密码。
JWT 令牌生成和验证:
- created_access_token: 生成一个JWT令牌，包含用户的身份信息（如用户名）和令牌的有效期。
- jwt_get_current_user: 从请求中提取JWT令牌，解码并验证令牌的有效性，最后获取令牌中的用户名，用于获取用户对象。
- jwt_get_current_active_user: 进一步验证用户是否被禁用，确保只有活跃用户可以访问受保护的路由。
- 路由:
- /jwt/token: 用户通过发送用户名和密码到这个路由来登录。如果认证成功，返回一个JWT令牌。
- /jwt/users/me: 一个受保护的路由，用户需要通过JWT令牌进行认证才能访问。返回当前登录用户的信息。
安全认证流程
- 用户发送用户名和密码到**/jwt/token**路由。
- 系统验证用户名和密码，如果认证成功，生成一个包含用户信息和有效期的JWT令牌。
- 用户在后续请求的Authorization头部中携带这个令牌。
- 系统解码并验证JWT令牌的有效性。如果令牌有效，提取令牌中的用户名，获取用户对象。
- 系统根据用户对象和路由的安全要求，决定是否允许用户访问受保护的资源。
关键技术点
- JWT: 用于在用户认证后生成访问令牌，令牌中包含用户身份信息和有效期。
- bcrypt: 用于用户密码的加密和校验，增强系统的安全性。
- FastAPI的依赖注入系统: 用于实现认证流程中的各种逻辑，如用户认证、令牌生成和验证等。
from datetime import datetime, timedelta from typing import Optional from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from pydantic import BaseModel from passlib.context import CryptContext from jose import JWTError, jwt app = FastAPI() """ 开发基于JSON Web Token 的安全认证""" # 模拟数据库数据 fake_users_db = { "zhangsan": { "username": "zhangsan", "full_name": "Zhang San", "email": "zhangsan@example.com", "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX30XePaWxn96p36WQoeG6Lruj3vjPGga31lW", "disabled": False, } } SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6fof4caa6cf63b88e8d3e7" ALGORITHM = "HS256" # 算法 ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 访问令牌过期时间（分钟） # token 输出模型 class Token(BaseModel): """返回给用户的Token""" access_token: str token_type: str # 用户基础数据模型 class User(BaseModel): username: str email: Optional[str] = None full_name: Optional[str] = None disabled: Optional[bool] = None # 用户输入数据模型 class UserInDB(User): hashed_password: str pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_schema = OAuth2PasswordBearer(tokenUrl="/jwt/token") def verify_password(plain_password: str, hashed_password: str) -> bool: """对密码进行校验""" # 模拟对用户的明文密码进行hash加密后校验 hashed_pwd = '$2b$12$EixZaYVK1fsbw1ZfbX30XePaWxn96p36WQoeG6Lruj3vjPGga31lW' if plain_password == "zhangsan" and hashed_password == hashed_pwd: return True return pwd_context.verify(plain_password, hashed_password) def jwt_get_user(db, username: str): """获取用户和密码""" if username in db: user_dict = db[username] return UserInDB(**user_dict) def jwt_authenticate_user(db, username: str, password: str): """验证用户和密码""" user = jwt_get_user(db=db, username=username) if not user: return False if not verify_password(plain_password=password, hashed_password=user.hashed_password): return False return user def created_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expires = datetime.utcnow() + expires_delta else: expires = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expires}) encode_jwt = jwt.encode(claims=to_encode, key=SECRET_KEY, algorithm=ALGORITHM) return encode_jwt @app.post("/jwt/token", response_model=Token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): user = jwt_authenticate_user(db=fake_users_db, username=form_data.username, password=form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password 账号或密码错误", headers={"WWW-Authenticate": "Bearer"} ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = created_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} async def jwt_get_current_user(token: str = Depends(oauth2_schema)): credentials_exception = HTTPException( status_code=status.HTTP_200_OK, detail="ICould not validate credentials 身份验证失效", headers={"WWW-Authenticate": "Bearer"} # OAuth2 的规范，如果认证失败，请求头中返回WWW- ) try: payload = jwt.decode(token=token, key=SECRET_KEY, algorithms=[ALGORITHM]) username = payload.get("sub") if username is None: raise credentials_exception except JWTError: raise credentials_exception user = jwt_get_user(db=fake_users_db, username=username) if user is None: raise credentials_exception return user async def jwt_get_current_active_user(current_user: User = Depends(jwt_get_current_user)): if current_user.disabled: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="无效用户" ) return current_user @app.get("/jwt/users/me") async def jwt_read_users_me(current_user: User = Depends(jwt_get_current_active_user)): return current_user if __name__ == '__main__': import uvicorn # 启动命令：uvicorn hello_world:app --reload uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
中间件
“中间件”是一个函数，它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前介入；
- 中间件的作用：中间件可以检查请求的内容，如头部信息、查询参数、路径参数等，并据此执行相应的逻辑。 同样，它也可以在响应被发送回客户端之前修改或增强响应。
- 中间件的流程：中间件首先接收请求，执行一些操作，然后将请求传递给应用程序的其他部分；当应用程序生成响应后，中间件再次介入，可能执行一些额外的操作，然后返回响应。
- 中间件链：在许多Web框架中，中间件可以形成一个链式结构。一个请求会依次经过多个中间件，每个中间件都会按照上述流程处理请求和响应。这允许开发者在应用程序的不同层次上应用不同的逻辑和策略。
- 中间件的应用场景：中间件可以应用于许多场景，如身份验证、日志记录、异常处理、缓存、CORS策略等。通过中间件，开发者可以以一种模块化和可重用的方式实现这些功能，而不需要在每个路由或控制器中重复相同的代码。
- 中间件和路由的关系：在FastAPI或Starlette这样的框架中，中间件和路由是相辅相成的。路由定义了应用程序的端点（即特定的URL路径），而中间件则负责在这些端点之前和之后执行额外的逻辑。这使得开发者可以灵活地组合不同的中间件和路由，以满足复杂的应用程序需求。
在视图函数的顶部使用**@app.middleware(“http”)**创建中间件
中间件参数说明：
**request：**当前HTTP请求对象；
**call_next：**一个可调用的对象，它代表了中间件链中的下一个中间件或最终的路由操作；
如果该中间件后面还有中间件，那么call_next就是下一个中间件；
如果没有，那么call_next就是对应的视图函数；
示例代码
import time import uvicorn from fastapi import FastAPI, Request # 主应用路由（FastAPI实例） app = FastAPI() @app.middleware("http") async def m2(request: Request, call_next): # 请求代码卡 print("m2 request") response = await call_next(request) # 响应代码卡 print("m2 response") return response @app.middleware("http") async def m1(request: Request, call_next): # 请求代码卡 print("m1 request") # 限制IP # if request.client.host in ["127.0.0.1", ]: # 黑名单 # return Response(content="黑名单") # 限制路由 # if request.url.path in ["/user"]: # return Response(content="黑名单") # 记录请求时间 start_time = time.time() response = await call_next(request) # 响应代码块 print("m1 response") # 记录响应时间 end_time = time.time() # 计算耗时 take_time = end_time - start_time response.headers["X-start_time"] = str(start_time) response.headers["X-end_time"] = str(end_time) response.headers["X-take_time"] = str(take_time) return response @app.get("/user") def get_user(): print("get_user函数执行了...") return { "user": "current user" } @app.get("/item/{item_id}") def get_item(item_id: int): print("get_item函数执行了...") return { "item": f"current item, {item_id}" } if __name__ == '__main__': uvicorn.run(f"main:app", port=8080, reload=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
中间件一旦形成链，需要明白中间件的执行顺序，以及中间件的介入点；
跨域/资源共享
随着前后端分离的盛行，后端程序员和前端程序员的职责划分愈发清晰。
后端专注于构建和提供稳定的API接口，确保返回格式统一的JSON数据，而前端则致力于呈现丰富的用户界面和流畅的交互体验。
这种明确的分工不仅简化了数据接入流程，还提高了开发效率。
然而，在网络安全日益受到重视的背景下，浏览器为保护网络安全实施了同源策略，要求前端请求与页面同源，包括协议、域名和端口。
违反此策略的请求将被阻止，增强了安全性。
跨域
CORS（跨域资源共享）是一种机制，允许在浏览器中运行的前端代码，如 JavaScript，与位于不同源的后端进行通信。这种场景在现代 web 开发中极为常见，尤其是在前后端分离架构盛行的今天。
同源
源是由协议（如http、https）、域名（如myapp.com、localhost、localhost.tiangolo.com）以及端口号（如80、443、8080）共同构成的组合。
这三个元素共同决定了网页或应用程序的来源。
基于这个定义，以下示例中的每一个都是独立的源：
- http://localhost
- https://localhost
- http://localhost:8080
尽管它们都指向同一台机器（localhost）
但由于使用了不同的通信协议或端口号，因此被视为不同的源。这种机制是浏览器安全策略的一部分，用于防止恶意脚本对不同来源的资源进行未经授权的访问，从而保护了用户数据的安全。
场景
假设你的浏览器中有一个前端运行在http://localhost:8080；
并且它的JavaScript正在尝试与运行在http://localhost的后端通信（因为我们没有指定端口，浏览器会采用默认的端口80）；然后，浏览器会向后端发送一个HTTP OPTIONS请求，如果后端发送适当的 headers来授权来自这个不同源（http://localhost:8080）的通信，浏览器将允许前端的JavaScript向后端发送请求。
为此，后端必须有一个「允许的源」列表；
在这种情况下，它必须包含http://localhost:8080，前端才能正常工作。
模拟跨域
要在FastAPI中轻松实现跨域资源共享（CORS），可以利用内置的CORSMiddleware中间件；
- 这个中间件为开发者提供了灵活的方式来定义哪些来源能够访问你的API；
- 并精确控制这些来源可以使用哪些HTTP方法和头部信息。
通过合理地配置这个中间件，可以确保API的安全性和数据的完整性，同时为用户提供更加友好的访问体验。
main.py代码
import uvicorn from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware # 主应用路由（FastAPI实例） app = FastAPI() # 简单请求的跨域解决 # @app.middleware("http") # async def MyCORSMiddleware(request: Request, call_next): # response = await call_next(request) # response.headers["Access-Control-Allow-Origin"] = "*" # return response # 使用FastAPI已经封装完善的跨域解决 origins = [ "http://localhost:63342" ] app.add_middleware( CORSMiddleware, # 允许跨域的源列表，例如 ["http://localhost:8080"] # ["*"] 表示允许任何源 allow_origins=origins, # 跨域请求是否支持 cookie，默认是 False # 如果为 True，allow_origins 必须为具体的源，不可以是 ["*"] allow_credentials=True, # 允许跨域请求的 HTTP 方法列表，默认是 ["GET"] allow_methods=["GET"], # 允许跨域请求的 HTTP 请求头列表，默认是 [] # 可以使用 ["*"] 表示允许所有的请求头 # 当然下面几个请求头总是被允许的 # Accept、Accept-Language、Content-Language、Content-Type allow_headers=["*"], # 可以被浏览器访问的响应头, 默认是 []，一般很少指定 # expose_headers=["*"] # 设定浏览器缓存 CORS 响应的最长时间，单位是秒 # 默认为 600，一般也很少指定 # max_age=1000 ) @app.get("/user") def get_user(): print("user -> test") print("get_user函数执行了...") return { "user": "current user" } if __name__ == '__main__': uvicorn.run(f"main:app", port=8080, reload=True)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
index.html代码
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Title</title> <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script> </head> <body> <p>Click</p> <script> $("p").click(function () { $.ajax({ url: "http://127.0.0.1:8080/user", success: function (res) { console.log(res) console.log(res.user) $("p").html(res.messags) }, }) }) </script> </body> </html>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/638238
推荐阅读
相关标签
