【Python全栈开发】第8讲 | Web 全栈之巅:FastAPI 高性能后端开发

环境声明

  • Python版本:Python 3.12+ (建议使用 3.10 以上版本)
  • 开发工具:PyCharm 或 VS Code
  • 操作系统:Windows / macOS / Linux (通用)

1. 为什么是 FastAPI?

如果你还在学习传统的 Django 或者 Flask,那这一讲你得认真看看了。

在现代全栈开发里,FastAPI 已经是很多大厂和初创公司的首选。为什么?

  1. 速度快:它的运行速度可以和 NodeJS 或 Go 媲美,这在 Python 界是突破性的。
  2. 类型驱动:它利用 Python 的类型提示(Type Hints),能自动帮你生成接口文档、做数据校验。
  3. 异步原生:写起来代码极少,而且原生支持 async/await 异步。

今天,咱们就用 FastAPI 亲手搭建一个生产级的 API 服务。


2. 环境搭建:三行命令搞定

首先,你需要安装两个东西:FastAPI 框架本身,和用来运行它的 Web 服务器 uvicorn

pip install fastapi uvicorn 

3. 底层原理:ASGI 协议详解

在深入 FastAPI 之前,我们需要理解它背后的核心协议 —— ASGI(Asynchronous Server Gateway Interface)。

3.1 什么是 ASGI?

ASGI 是 Python 异步 Web 服务器和应用程序之间的标准接口。你可以把它理解为一座桥梁:

协议特性代表框架
WSGI同步、单线程Flask、Django 早期
ASGI异步、支持 WebSocketFastAPI、Django Channels

3.2 ASGI 的工作原理

ASGI 应用本质上是一个可调用对象(Callable),接收三个参数:

asyncdefapplication(scope, receive, send):# scope: 包含请求信息(HTTP 方法、路径、Headers 等)# receive: 异步函数,用于接收请求体# send: 异步函数,用于发送响应await send({'type':'http.response.start','status':200,'headers':[(b'content-type',b'text/plain')],})await send({'type':'http.response.body','body':b'Hello, ASGI!',})

3.3 FastAPI 与 ASGI 的关系

FastAPI 是一个 ASGI 应用框架,它内部使用 Starlette 来处理 ASGI 协议。当你运行 uvicorn main:app 时:

  1. Uvicorn 作为 ASGI 服务器启动
  2. 接收到的 HTTP 请求被解析成 ASGI 事件
  3. FastAPI(通过 Starlette)处理这些事件
  4. 响应通过 ASGI 协议返回给客户端

一句话总结:ASGI 让 Python 能够高效处理并发请求,FastAPI 则在此基础上提供了优雅的开发体验。


4. Hello World:五行代码跑起一个服务

# main.pyfrom fastapi import FastAPI app = FastAPI()@app.get("/")defread_root():return{"message":"你好,全栈之路!"}# 启动命令:uvicorn main:app --reload

解释--reload 是开发神器。你改完代码保存,服务器会自动重启,不需要手动关掉再开。


5. Pydantic:你的"数据质检员"

写 Web 服务最头疼的就是验证用户传过来的数据。万一该传数字的地方传了字符串,程序就崩了。
FastAPI 搭配 Pydantic,一行代码搞定验证:

from pydantic import BaseModel # 定义一个"商品"的数据结构classItem(BaseModel): name:str price:float is_offer:bool|[email protected]("/items/")defcreate_item(item: Item):# 如果用户传的 price 不是数字,FastAPI 会自动返回 422 错误return{"item_name": item.name,"total_price": item.price *1.2}

6. Pydantic 数据验证原理深度解析

6.1 验证流程

当 FastAPI 接收到请求时,Pydantic 会执行以下验证步骤:

  1. 类型检查:确保每个字段符合声明的类型
  2. 约束验证:检查 min_length、max_length、ge(大于等于)、le(小于等于)等
  3. 默认值填充:为可选字段填充默认值
  4. 自定义验证器:执行 @validator@field_validator 装饰的方法

6.2 自定义验证器示例

from pydantic import BaseModel, field_validator classUser(BaseModel): username:str age:int email:str@field_validator('username')@classmethoddefvalidate_username(cls, v):iflen(v)<3:raise ValueError('用户名至少3个字符')ifnot v.isalnum():raise ValueError('用户名只能包含字母和数字')return v @field_validator('age')@classmethoddefvalidate_age(cls, v):if v <0or v >150:raise ValueError('年龄必须在0-150之间')return v # 测试验证from fastapi import FastAPI app = FastAPI()@app.post("/users/")defcreate_user(user: User):return{"message":"用户创建成功","user": user}

6.3 嵌套模型验证

from typing import List from pydantic import BaseModel classAddress(BaseModel): city:str street:str zipcode:strclassUserWithAddress(BaseModel): name:str addresses: List[Address]# 嵌套模型列表@app.post("/users-with-address/")defcreate_user_with_address(user: UserWithAddress):return user 

一句话总结:Pydantic 在数据进入你的业务逻辑之前就完成所有验证,让你专注于业务而不是防御性编程。


7. 路径参数 vs 查询参数

  • 路径参数/items/42(指定 ID)。
  • 查询参数/items/?skip=0&limit=10(翻页、搜索)。
@app.get("/users/{user_id}")defread_user(user_id:int, q:str|None=None):# user_id 会自动转成整数,q 是可选的搜索关键词return{"user_id": user_id,"query": q}

8. 依赖注入容器原理

8.1 什么是依赖注入?

依赖注入(Dependency Injection, DI)是一种设计模式,它将对象的创建和管理从使用它的代码中分离出来。

比喻:就像你去餐厅吃饭,不需要自己种菜、做饭,只需要点菜(声明依赖),服务员会把菜端上来(注入依赖)。

8.2 FastAPI 依赖注入的核心机制

FastAPI 的依赖注入系统基于 Python 的函数和类型注解:

from fastapi import Depends, FastAPI app = FastAPI()# 定义一个依赖函数defcommon_parameters(q:str|None=None, skip:int=0, limit:int=100):return{"q": q,"skip": skip,"limit": limit}# 使用依赖@app.get("/items/")defread_items(commons:dict= Depends(common_parameters)):return commons @app.get("/users/")defread_users(commons:dict= Depends(common_parameters)):return commons 

8.3 依赖的执行流程

  1. 解析依赖树:FastAPI 分析路由函数的参数,识别所有 Depends 标记的依赖
  2. 递归解析:如果依赖 A 依赖 B,先解析 B,再解析 A
  3. 缓存机制:同一个请求中,相同的依赖只执行一次(可配置)
  4. 注入执行:将解析结果作为参数传给路由函数

8.4 类作为依赖(依赖容器)

from fastapi import Depends, FastAPI app = FastAPI()classDatabaseConnection:def__init__(self): self.connection ="模拟数据库连接"defquery(self, sql:str):returnf"执行: {sql}"defclose(self): self.connection =Nonedefget_db(): db = DatabaseConnection()try:yield db # 生成器方式,支持清理操作finally: db.close()@app.get("/data/")defget_data(db: DatabaseConnection = Depends(get_db)): result = db.query("SELECT * FROM users")return{"result": result}

8.5 子依赖与依赖链

from fastapi import Depends, FastAPI, Header, HTTPException app = FastAPI()# 基础依赖:验证 Tokendefverify_token(x_token:str= Header(...)):if x_token !="secret-token":raise HTTPException(status_code=400, detail="Token 无效")return x_token # 子依赖:获取当前用户(依赖 verify_token)defget_current_user(token:str= Depends(verify_token)):# 根据 token 查询用户信息return{"username":"zhangsan","id":1}# 使用依赖链@app.get("/profile/")defread_profile(user:dict= Depends(get_current_user)):return user 

一句话总结:依赖注入让代码解耦、可测试、可复用,是 FastAPI 的灵魂特性。


9. 中间件执行流程

9.1 什么是中间件?

中间件是在请求到达路由处理函数之前和响应返回客户端之前执行的代码。它可以:

  • 记录请求日志
  • 添加响应头
  • 处理跨域(CORS)
  • 认证和授权
  • 压缩响应

9.2 中间件执行顺序

请求 -> 中间件1 -> 中间件2 -> 路由处理 -> 中间件2 -> 中间件1 -> 响应 

形象比喻:中间件就像洋葱的层层包裹,请求从外向内穿透,响应从内向外返回。

9.3 自定义中间件

from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import time app = FastAPI()@app.middleware("http")asyncdefadd_process_time_header(request: Request, call_next):# 请求前处理 start_time = time.time()print(f"[请求] {request.method}{request.url.path}")# 继续处理请求 response =await call_next(request)# 响应后处理 process_time = time.time()- start_time response.headers["X-Process-Time"]=str(process_time)print(f"[响应] 耗时: {process_time:.4f}s")return response # 错误处理中间件@app.middleware("http")asyncdefcatch_exceptions(request: Request, call_next):try:returnawait call_next(request)except Exception as e:return JSONResponse( status_code=500, content={"error":str(e)})@app.get("/")defread_root():return{"message":"Hello"}

9.4 内置中间件

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware app = FastAPI()# CORS 中间件 app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000","https://example.com"], allow_credentials=True, allow_methods=["GET","POST","PUT","DELETE"], allow_headers=["*"],)# GZip 压缩中间件 app.add_middleware(GZipMiddleware, minimum_size=1000)

9.5 中间件 vs 依赖注入

特性中间件依赖注入
执行时机所有请求特定路由
粒度全局局部
使用场景日志、CORS、压缩数据库连接、认证
返回值修改 request/response注入到路由函数

一句话总结:中间件处理横切关注点,依赖注入提供模块化服务,两者配合让代码更优雅。


10. Web 安全基础

Web 安全是每个后端开发者必须掌握的知识。以下是三大常见攻击及防护方法。

10.1 CSRF(跨站请求伪造)

攻击原理:诱导用户在已登录的网站上执行非预期的操作。

防护措施

from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse app = FastAPI()# CSRF Token 验证示例asyncdefverify_csrf_token(request: Request):# 从 Header 获取 CSRF Token csrf_token = request.headers.get("X-CSRF-Token")# 从 Session/Cookie 获取存储的 Token stored_token = request.cookies.get("csrf_token")ifnot csrf_token or csrf_token != stored_token:raise HTTPException(status_code=403, detail="CSRF Token 无效")[email protected]("/transfer/")asyncdeftransfer_money(request: Request):await verify_csrf_token(request)return{"message":"转账成功"}

最佳实践

  • 使用 SameSite Cookie 属性
  • 验证 Referer/Origin Header
  • 关键操作使用二次确认

10.2 XSS(跨站脚本攻击)

攻击原理:注入恶意脚本到网页中,窃取用户 Cookie 或执行恶意操作。

防护措施

from fastapi import FastAPI from pydantic import BaseModel, field_validator import html app = FastAPI()classComment(BaseModel): content:str@field_validator('content')@classmethoddefsanitize_content(cls, v):# HTML 转义,防止脚本注入return html.escape(v)@app.post("/comments/")defcreate_comment(comment: Comment):# 此时 content 已经被转义,<script> 变成 <script>return{"content": comment.content}

最佳实践

  • 对所有用户输入进行转义
  • 使用 Content Security Policy (CSP)
  • 设置 HttpOnly Cookie

10.3 SQL 注入防护

攻击原理:通过构造特殊输入,篡改 SQL 查询语句。

防护措施

from fastapi import FastAPI, HTTPException import sqlite3 app = FastAPI()# 错误示例(不要这样做)@app.get("/users/unsafe/{user_id}")defget_user_unsafe(user_id:str): conn = sqlite3.connect("test.db") cursor = conn.cursor()# 危险!直接拼接 SQL cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")return cursor.fetchone()# 正确做法:使用参数化查询@app.get("/users/safe/{user_id}")defget_user_safe(user_id:int): conn = sqlite3.connect("test.db") cursor = conn.cursor()# 安全:使用占位符,自动转义 cursor.execute("SELECT * FROM users WHERE id = ?",(user_id,))return cursor.fetchone()

最佳实践

  • 永远使用参数化查询/预编译语句
  • ORM(如 SQLAlchemy)自动处理转义
  • 最小权限原则:数据库用户只授予必要权限

10.4 安全头部配置

from fastapi import FastAPI from fastapi.middleware.trustedhost import TrustedHostMiddleware from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware app = FastAPI()# 安全头部中间件@app.middleware("http")asyncdefadd_security_headers(request, call_next): response =await call_next(request)# 防止 XSS response.headers["X-Content-Type-Options"]="nosniff" response.headers["X-Frame-Options"]="DENY" response.headers["X-XSS-Protection"]="1; mode=block"# 强制 HTTPS response.headers["Strict-Transport-Security"]="max-age=31536000; includeSubDomains"return response # 限制允许的 Host app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com","*.example.com"])

一句话总结:安全不是功能,而是底线。永远不信任用户输入,永远使用参数化查询,永远做好输出转义。


11. FastAPI 0.100+ 新特性

FastAPI 0.100 版本是一个重大里程碑,带来了许多重要更新。

11.1 Pydantic V2 支持

FastAPI 0.100+ 默认使用 Pydantic V2,性能提升 5-50 倍:

from pydantic import BaseModel, Field # Pydantic V2 新特性classItem(BaseModel): name:str= Field(min_length=1, max_length=100) price:float= Field(gt=0, description="必须大于0")# V2 新特性:严格模式 model_config ={"strict":True,# 禁止类型强制转换"extra":"forbid"# 禁止额外字段}

11.2 新的 Annotated 语法

from typing import Annotated from fastapi import FastAPI, Query, Path app = FastAPI()# 0.100+ 推荐的新语法@app.get("/items/{item_id}")defread_item( item_id: Annotated[int, Path(title="项目ID", ge=1)], q: Annotated[str|None, Query(min_length=3)]=None, limit: Annotated[int, Query(le=100)]=10):return{"item_id": item_id,"q": q,"limit": limit}

11.3 更快的启动速度

Pydantic V2 使用 Rust 编写的核心验证逻辑,带来:

  • 更快的应用启动时间
  • 更低的内存占用
  • 更高的请求处理吞吐量

11.4 改进的异常处理

from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError app = FastAPI()# 自定义验证错误处理@app.exception_handler(RequestValidationError)asyncdefvalidation_exception_handler(request: Request, exc: RequestValidationError): errors =[]for error in exc.errors(): errors.append({"field":" -> ".join(str(x)for x in error["loc"]),"message": error["msg"],"type": error["type"]})return JSONResponse( status_code=422, content={"detail":"数据验证失败","errors": errors})

11.5 迁移指南

从旧版本迁移到 0.100+:

# 1. 升级依赖 pip install--upgrade fastapi pydantic # 2. 检查 Pydantic V2 兼容性 pip install pydantic-v2-migration python -m pydantic_v2_migration 

一句话总结:FastAPI 0.100+ 是一次质的飞跃,Pydantic V2 带来的性能提升让 FastAPI 在生产环境中更加游刃有余。


12. 自动化文档:FastAPI 的"杀手锏"

如果你以前写 API,肯定被写 Swagger 文档折磨过。
在 FastAPI 里,你什么都不用做。服务跑起来后,直接访问:

  • http://127.0.0.1:8000/docs

你会看到一个超级精美的交互式文档。你可以在上面直接点"Try it out"测试你的接口,再也不用开 Postman 了!


13. 综合实战:构建一个待办事项 (To-Do) API

咱们来个带增删改查(CRUD)功能的实战案例。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI()# 模拟数据库 todo_db =[]classTodo(BaseModel):id:int title:str completed:[email protected]("/todos/", status_code=201)defadd_todo(todo: Todo): todo_db.append(todo)return{"message":"添加成功","data": todo}@app.get("/todos/")deflist_todos():return todo_db @app.get("/todos/{todo_id}")defget_todo(todo_id:int):for t in todo_db:if t.id== todo_id:return t raise HTTPException(status_code=404, detail="任务找不到了...")@app.delete("/todos/{todo_id}")defdelete_todo(todo_id:int):global todo_db todo_db =[t for t in todo_db if t.id!= todo_id]return{"message":"删除成功"}

14. 避坑小贴士(经验总结)

  1. async def 别乱用:如果你的函数里没有 await 任何东西(比如没去读文件、没去查数据库),直接写 def 往往更快。FastAPI 会在独立的线程池里跑它,不会阻塞。
  2. 状态码很重要:成功了返回 200/201,找不到了返回 404,权限不够返回 401/403。别不管啥情况都返回 200,前端同学会想打人的。
  3. 依赖注入 (Dependency Injection):当你以后要处理用户登录(Token 验证)时,去搜一下 FastAPI 的 Depends。它是这款框架的灵魂。

15. 实战演练:巩固你的内功

题目 1:查询参数与校验

需求
编写一个 API GET /search/,接收参数 keyword(必填,长度至少 2)和 limit(选填,默认 10,最大 50)。
使用 Query 进行参数校验。

点击查看参考答案

from fastapi import FastAPI, Query app = FastAPI()@app.get("/search/")defsearch( keyword:str= Query(..., min_length=2, description="搜索关键词"), limit:int= Query(10, le=50, description="返回结果数量")):return{"keyword": keyword,"limit": limit,"results":["模拟数据1","模拟数据2"]}

题目 2:依赖注入与 Token 验证

需求
编写一个依赖函数 verify_token,检查请求头 x-token 是否为 “fake-super-secret-token”。
如果是,允许访问;否则抛出 400 错误。
将此依赖应用到 GET /protected/ 路由上。

点击查看参考答案

from fastapi import FastAPI, Header, HTTPException, Depends app = FastAPI()defverify_token(x_token:str= Header(...)):if x_token !="fake-super-secret-token":raise HTTPException(status_code=400, detail="Token 无效")return x_token @app.get("/protected/")defprotected_route(token:str= Depends(verify_token)):return{"message":"验证通过","token": token}

题目 3:文件上传

需求
编写一个 API POST /upload/,接收一个文件上传。
返回文件的文件名和文件大小。
提示:需要安装 python-multipart

点击查看参考答案

from fastapi import FastAPI, UploadFile, File app = FastAPI()@app.post("/upload/")asyncdefupload_file(file: UploadFile = File(...)):# file.filename 是文件名# await file.read() 读取内容 content =awaitfile.read()return{"filename":file.filename,"content_type":file.content_type,"size":len(content)}

16. 系列索引


写在最后
这一讲学完,你已经能够把你的 Python 逻辑变成全世界都能访问的服务了。
Web 开发的水很深,安全、高并发、缓存、部署都是大课。咱们先把 API 跑起来,以后再慢慢优化。
别只是看,去电脑上运行那个 To-Do API,打开 /docs 玩一下。
觉得有收获的话,点赞、收藏!咱们下一讲聊聊怎么让你的代码达到生产级别——工程化实战!

Read more

【前端】前端面试题

【前端】前端面试题

前端面试题 闭包 1. 定义: 闭包(Closure) 是指一个函数能够访问并记住其外部作用域中的变量,即使外部函数已经执行完毕。闭包由两部分组成: * 一个函数(通常是内部函数)。 * 该函数被创建时所在的作用域(即外部函数的变量环境) functionouter(){let count =0;// 外部函数的变量functioninner(){ count++;// 内部函数访问外部变量 console.log(count);}return inner;}const counter =outer();counter();// 输出 1counter();// 输出 2 2. 闭包的核心原理 * 作用域链:函数在定义时,会记住自己的词法环境(即外部作用域)。当内部函数访问变量时,会沿着作用域链向上查找。 * 变量持久化:闭包使得外部函数的变量不会被垃圾回收,因为内部函数仍持有对它们的引用 3. 闭包的常见用途 3.1 私有变量封装 通过闭包隐藏内部变量,

By Ne0inhk

web前端JS—基本语法

一、引入方式 1、内部脚本:将代码定义在HTML页面里面 * 将JS定义在<script></script>之间 * 可以在html里面的任意位置放置任意数量的<script></script> * 一般放置在<body>元素的底部,改善显示速度 <script> console.log('页面加载时执行'); function localFunction() { return '内部函数'; } </script> 2、外部脚本:额外定义一个.js文件,引入到HTML里面 * 只能包含js文件,不包含&

By Ne0inhk
Spring Web MVC 入门秘籍:从概念到实践的快速通道(上)

Spring Web MVC 入门秘籍:从概念到实践的快速通道(上)

个人主页:♡喜欢做梦 欢迎  👍点赞  ➕关注  ❤️收藏  💬评论 目录 一、什么是Spring Web MVC? 1.定义 2.关于MVC 2.1什么是MVC? 二、学习Sring MVC 1.项目准备 2.@RestController注解的介绍 3.@RequestMapping注解 1.@RequestMapping只用在类上(相当于“只给文件夹起名,不给文件起名”) 2.@RequestMapping只用在方法上(相当于“只给文件起名,不给文件夹起名”) 3.@RequestMapping用在类和方法上(相当于给文件夹和文件都取名) 三、Postman 1.什么是Postman? 2.关于API 3.传参介绍 1.普通传参:

By Ne0inhk
【前端实战】Axios 错误处理的设计与进阶封装,实现网络层面的数据与状态解耦

【前端实战】Axios 错误处理的设计与进阶封装,实现网络层面的数据与状态解耦

目录 【前端实战】Axios 错误处理的设计与进阶封装,实现网络层面的数据与状态解耦 一、为什么网络错误处理一定要下沉到 Axios 层 二、Axios 拦截器 interceptors 1、拦截器的基础应用 2、错误分级和策略映射的设计 3、错误对象标准化 三、结语         作者:watermelo37         ZEEKLOG优质创作者、华为云云享专家、阿里云专家博主、腾讯云“创作之星”特邀作者、火山KOL、支付宝合作作者,全平台博客昵称watermelo37。         一个假装是giser的coder,做不只专注于业务逻辑的前端工程师,Java、Docker、Python、LLM均有涉猎。 --------------------------------------------------------------------- 温柔地对待温柔的人,包容的三观就是最大的温柔。 --------------------------------------------------------------------- 【前

By Ne0inhk