面向0基础的FastAPI教程:0.前置知识
FastAPI 的门槛不在框架本身,而在 “Web 开发的底层认知” 和 “Python 的核心能力”。这篇文章会把我当初学 FastAPI 前踩过的坑、必须搞懂的前置知识,用 “人话 + 例子 + 关联 FastAPI”的方式讲清楚,帮你彻底告别 “跟着敲代码但不懂原理” 的困境
FastAPI 的门槛不在框架本身,而在 “Web 开发的底层认知” 和 “Python 的核心能力”。这篇文章会把我当初学 FastAPI 前踩过的坑、必须搞懂的前置知识,用 “人话 + 例子 + 关联 FastAPI”的方式讲清楚,帮你彻底告别 “跟着敲代码但不懂原理” 的困境。
0. 先搞懂:FastAPI 到底是做什么的?
在讲前置知识前,必须先给你一个 **“Web 开发的全局地图”**—— 否则后面的内容你会越看越懵。
Web 开发的基本流程(用淘宝举例子)
当你打开淘宝 App,点击 “我的订单”:
- 客户端(淘宝 App):给淘宝的服务器发一个HTTP 请求(相当于说:“我要查订单,用户 ID 是 123”);
- 服务器(用 FastAPI 写的):接收请求,处理逻辑(比如从数据库查用户 123 的订单);
- 服务器返回响应:把订单数据用JSON格式发给客户端(相当于说:“你的订单是这些,订单号 12345”);
- 客户端渲染:把 JSON 数据变成你看到的 “订单列表”。
FastAPI 的角色
FastAPI 是 **“帮你写服务器的框架”**—— 它帮你解决了 “如何接收请求”“如何处理参数”“如何返回响应” 这些底层问题,让你只需要专注于 “查订单”“创建用户” 这些业务逻辑。
简单来说:
- 没有 FastAPI,你得自己写几百行代码处理 HTTP 请求;
- 有了 FastAPI,你只需要写
@app.get("/orders")就能定义一个 “查订单” 的接口 —— 剩下的交给框架。
1. 前置知识 1:HTTP 的基本概念(FastAPI 的 “沟通语言”)
HTTP 是客户端和服务器之间的 “沟通协议”—— 就像你和外国朋友交流要用英语,客户端和服务器交流要用 HTTP。
1.1 什么是 HTTP 请求?什么是 HTTP 响应?
- HTTP 请求:客户端发给服务器的 “消息”(比如 “查订单”);
- HTTP 响应:服务器回给客户端的 “消息”(比如 “你的订单列表”)。
1.2 HTTP 的 4 个核心方法(FastAPI 的接口定义全靠它)
HTTP 方法是 “请求的类型”—— 不同的方法对应不同的操作,FastAPI 用@app.get() @app.post()这样的装饰器对应这些方法:
| 方法 | 用途 | FastAPI 示例 | 生活例子 |
|---|---|---|---|
| GET | 获取数据(读) | @app.get("/users/{id}") |
你问淘宝:“给我看看用户 123 的信息” |
| POST | 提交数据(写) | @app.post("/users") |
你给淘宝发:“帮我创建一个新用户” |
| PUT | 更新数据(改) | @app.put("/users/{id}") |
你给淘宝发:“把用户 123 的名字改成张三” |
| DELETE | 删除数据(删) | @app.delete("/users/{id}") |
你给淘宝发:“把用户 123 删掉” |
1.3 HTTP 状态码(服务器的 “情绪表达”)
状态码是服务器返回的 “数字代码”,告诉你 “请求处理得怎么样了”——FastAPI 会自动或手动返回这些状态码:
| 状态码 | 含义 | FastAPI 场景 |
|---|---|---|
| 200 | 成功 | 查订单成功,返回订单数据 |
| 404 | 资源未找到 | 查用户 123,但用户不存在 |
| 422 | 参数校验失败 | 你传了个字符串当用户 ID(应该传 int) |
| 500 | 服务器内部错误 | 代码里写了1/0(除以 0) |
我踩过的坑:当初学 FastAPI 时,写了个@app.get("/users/{id}")接口,传了个字符串"abc"当 id,结果返回 422 错误 —— 后来才知道,FastAPI 会根据你写的id: int自动校验参数类型,字符串不符合 int 类型就会报 422。
2. 前置知识 2:Python 的函数高级用法(FastAPI 的 “接口骨架”)
FastAPI 的接口本质是 Python 函数—— 比如:
@app.get("/users/{id}")
def get_user(id: int): # 这个函数就是接口的逻辑
return {"user_id": id}
所以,你必须先搞懂 Python 函数的高级参数用法和类型提示—— 这是 FastAPI 接口的 “骨架”。
2.1 函数的参数类型(FastAPI 的 “参数校验器”)
Python 的函数参数有 4 种类型,FastAPI 会用这些参数来自动解析请求中的数据:
(1)路径参数(Path Parameter)
- 定义:写在 URL 里的参数(比如
/users/{id}里的id); - FastAPI 关联:FastAPI 会自动把 URL 里的
id提取出来,传给函数的id参数; - 例子:
当你访问@app.get("/users/{id}") # {id}是路径参数 def get_user(id: int): # id的类型是int(类型提示) return {"user_id": id}/users/123,函数的id会收到123(int 类型);如果访问/users/abc,FastAPI 会返回 422 错误(参数类型不对)。
(2)查询参数(Query Parameter)
- 定义:URL 里
?后面的参数(比如/users?page=1&size=10里的page和size); - FastAPI 关联:函数中未写在 URL 里的参数,默认是查询参数;
- 例子:
当你访问@app.get("/users") # URL里没有参数 def get_users(page: int = 1, size: int = 10): # page和size是查询参数,默认值1和10 return {"page": page, "size": size}/users?page=2&size=20,函数的page是2,size是20;如果不写参数,就用默认值1和10。
(3)类型提示(Type Hint)——FastAPI 的 “魔法来源”
- 定义:给函数参数或返回值加类型说明(比如
id: int,-> dict); - FastAPI 关联:FastAPI 靠类型提示做 3 件事:
- 自动校验参数(比如
id: int会校验参数是不是数字); - 自动生成文档(FastAPI 的
/docs页面会显示参数类型); - 自动转换数据(比如把 URL 里的字符串
"123"转成 int 类型的123)。
- 自动校验参数(比如
我踩过的坑:当初我写函数时,没加类型提示(比如def get_user(id):),结果 FastAPI 无法校验参数 —— 用户传个字符串"abc"当 id,函数里用id去查数据库,直接报 500 错误(数据库查不到字符串 id)。后来加了id: int,FastAPI 直接在 “入口” 就把无效参数挡住了,再也没报过这个错。
2.2 函数的默认参数与关键字参数
- 默认参数:函数定义时给参数设默认值(比如
page: int = 1); - 关键字参数:调用函数时用
参数名=值的方式传参(比如get_users(page=2, size=20))。
FastAPI 的应用:查询参数的默认值就是用默认参数实现的 —— 比如page: int = 1表示 “如果用户没传 page 参数,就用 1”。
3. 前置知识 3:类与对象(FastAPI 的 “组件积木”)
FastAPI 的很多核心组件都是类(比如FastAPI实例、Request对象、Response对象)—— 你需要用类来定义 “可复用的组件”(比如数据库连接、权限校验)。
3.1 类与对象的基本概念
- 类(Class):是 “模板”(比如 “人类”),定义了对象的属性(比如 “名字”“年龄”)和方法(比如 “说话”);
- 对象(Object):是类的 “实例”(比如 “张三” 是 “人类” 的实例)。
3.2 类的构造方法(__init__)
构造方法是创建对象时自动调用的方法—— 用来初始化对象的属性。比如:
class User:
def __init__(self, name: str, age: int): # 构造方法
self.name = name # 实例属性(每个对象都有自己的name)
self.age = age # 实例属性
# 创建对象(实例化)
user = User(name="张三", age=18) # 调用__init__方法,传入name和age
print(user.name) # 输出:张三
3.3 FastAPI 中的类应用
FastAPI 的FastAPI类是整个框架的 “核心实例”—— 你需要创建它来定义接口:
from fastapi import FastAPI
app = FastAPI() # 创建FastAPI的实例(对象)
@app.get("/") # 给app实例添加接口
def read_root():
return {"message": "Hello World"}
我踩过的坑:当初我写 FastAPI 项目时,想把数据库连接做成 “可复用的组件”—— 于是写了个Database类:
class Database:
def __init__(self, db_url: str):
self.db_url = db_url
self.connection = self.connect() # 初始化时连接数据库
def connect(self):
# 连接数据库的逻辑(比如用SQLAlchemy)
pass
# 创建数据库实例
db = Database(db_url="sqlite:///test.db")
# 在接口中使用
@app.get("/users/{id}")
def get_user(id: int):
user = db.connection.query(User).get(id) # 用db实例的connection查数据
return user
这样一来,所有接口都能复用这个db实例 —— 不用每次查数据库都重新连接,大大提升了性能。如果没学过类,我可能会在每个接口里写一遍数据库连接的代码,又冗余又容易错。
3.4 继承与多态(FastAPI 的 “扩展技巧”)
- 继承:子类可以继承父类的属性和方法(比如 “学生类” 继承 “人类” 的 “名字”“年龄”);
- 多态:不同的子类可以用相同的方法名实现不同的逻辑(比如 “学生说话” 和 “老师说话” 的内容不同)。
FastAPI 的应用:你可以用继承来定义 “通用的接口逻辑”—— 比如所有需要权限校验的接口,都继承一个BaseController类:
class BaseController:
def check_permission(self, user_id: int):
# 权限校验逻辑(比如查用户是不是管理员)
if user_id != 1: # 假设1是管理员ID
raise HTTPException(status_code=403, detail="没有权限")
# 子类继承BaseController
class UserController(BaseController):
@app.get("/users/{id}")
def get_user(self, id: int, current_user_id: int):
self.check_permission(current_user_id) # 调用父类的权限校验方法
return {"user_id": id}
4. 前置知识 4:JSON 与数据序列化(FastAPI 的 “数据格式”)
FastAPI 默认用JSON(JavaScript Object Notation)作为 “数据交换格式”—— 因为 JSON 是 “跨语言的”(不管客户端是 Python、Java 还是前端 JS,都能看懂 JSON)。
4.1 JSON 的基本格式
JSON 的结构和 Python 的字典、列表几乎一样:
- 键值对:用
{"key": "value"}表示(比如{"name": "张三", "age": 18}); - 数组:用
[value1, value2]表示(比如["苹果", "香蕉", "橙子"]); - 数据类型:支持字符串(
"张三")、数字(18)、布尔值(true/false)、null(空值)。
4.2 Python 与 JSON 的转换
FastAPI 会自动帮你做 JSON 的转换,但你得知道底层逻辑:
- Python 转 JSON:用
json.dumps()(dumps=dump string); - JSON 转 Python:用
json.loads()(loads=load string)。
例子:
import json
# Python字典转JSON字符串
python_dict = {"name": "张三", "age": 18}
json_str = json.dumps(python_dict) # 结果:'{"name": "\u5f20\u4e09", "age": 18}'(注意中文会被转义)
# JSON字符串转Python字典
json_str = '{"name": "张三", "age": 18}'
python_dict = json.loads(json_str) # 结果:{"name": "张三", "age": 18}
4.3 FastAPI 的自动 JSON 转换
当你在 FastAPI 接口中返回一个字典或列表,FastAPI 会自动把它转成 JSON 字符串 —— 比如:
@app.get("/users/{id}")
def get_user(id: int):
return {"name": "张三", "age": 18} # FastAPI自动转成JSON
我踩过的坑:当初我写接口时,返回了一个 Python 的datetime对象(比如datetime.now()),结果 FastAPI 报错 —— 因为datetime对象不能直接转 JSON。后来我查了文档,发现要把datetime转成字符串(比如str(datetime.now())),或者用pydantic的BaseModel(后面会讲)自动处理。
5. 前置知识 5:模块与包(FastAPI 的 “代码组织”)
当你的 FastAPI 项目变大时,你需要把代码分成多个模块和包—— 否则所有代码都写在一个main.py里,会乱得像 “垃圾堆”。
5.1 模块(Module)
- 模块:是一个
.py文件(比如user.py),里面可以定义函数、类、变量; - 导入模块:用
import语句(比如import user)。
FastAPI 的应用:你可以把 “用户相关的接口” 放到user.py模块里:
# user.py(模块)
from fastapi import APIRouter
router = APIRouter() # 用APIRouter来定义子路由
@router.get("/users/{id}")
def get_user(id: int):
return {"user_id": id}
@router.post("/users")
def create_user(name: str, age: int):
return {"name": name, "age": age}
然后在main.py里导入这个模块:
# main.py(主文件)
from fastapi import FastAPI
from user import router as user_router # 导入user模块的router
app = FastAPI()
app.include_router(user_router) # 把user的路由加到主app里
这样一来,main.py就会变得很干净 —— 所有用户相关的接口都在user.py里,逻辑清晰。
5.2 包(Package):FastAPI 的 “代码组织术”
之前讲了模块(单个.py文件),但当项目变大时,模块会太多—— 比如用户、订单、商品各有一个模块,这时候需要用包来组织这些模块。
包是包含__init__.py文件的文件夹——__init__.py可以是空文件,用来告诉 Python “这个文件夹是一个包”。
标准 FastAPI 项目结构示例(必看!)
我写过的所有 FastAPI 项目,都是用这个结构 —— 逻辑清晰,后期维护成本低:
fastapi_project/ # 项目根目录
├── main.py # 主文件(启动FastAPI,整合所有路由)
├── api/ # 接口模块包(放所有接口逻辑)
│ ├── __init__.py # 包标识文件(必须有)
│ ├── user.py # 用户接口(查用户、创建用户)
│ ├── order.py # 订单接口(创建订单、查订单)
│ └── product.py # 商品接口(查商品列表)
├── models/ # 数据模型包(数据库模型+Pydantic模型)
│ ├── __init__.py
│ ├── db_model.py # SQLAlchemy数据库表映射(ORM模型)
│ └── schema.py # Pydantic请求体/响应体模型
├── crud/ # CRUD逻辑包(增删改查业务逻辑)
│ ├── __init__.py
│ ├── user_crud.py # 用户CRUD(get_user、create_user)
│ └── order_crud.py# 订单CRUD
├── core/ # 核心配置包(全局设置、安全逻辑)
│ ├── __init__.py
│ ├── config.py # 环境变量(数据库URL、JWT秘钥)
│ └── security.py # 密码哈希、JWT令牌生成
├── db/ # 数据库连接包(引擎、Session管理)
│ ├── __init__.py
│ └── database.py # 创建SQLAlchemy引擎、SessionLocal
└── requirements.txt # 依赖清单(fastapi、uvicorn、sqlalchemy等)
每个文件夹的作用(小白必懂)
-
api/:接口模块包用APIRouter定义子路由(代替直接在main.py写所有接口),避免主文件臃肿。示例(api/user.py):from fastapi import APIRouter, Depends from models.schema import UserCreate, UserResponse from crud.user_crud import create_user, get_user from db.database import get_db from sqlalchemy.orm import Session # 前缀"/users":所有接口URL都带/users(比如/users/1) # 标签"用户管理":FastAPI文档会按标签分类 router = APIRouter(prefix="/users", tags=["用户管理"]) # 创建用户接口(请求体用UserCreate,响应体用UserResponse) @router.post("/", response_model=UserResponse) def create_user_api(user: UserCreate, db: Session = Depends(get_db)): return create_user(db, user) # 调用crud层逻辑 # 查询用户接口(路径参数id,响应体用UserResponse) @router.get("/{id}", response_model=UserResponse) def get_user_api(id: int, db: Session = Depends(get_db)): return get_user(db, id) # 调用crud层逻辑 -
models/:数据模型包-
db_model.py:用 ORM(如 SQLAlchemy)定义数据库表映射—— 把数据库表变成 Python 类,避免写原生 SQL。示例(models/db_model.py):from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() # 所有数据库模型的基类(自动管理表结构) class User(Base): __tablename__ = "users" # 数据库表名 id = Column(Integer, primary_key=True, index=True) # 主键(自增) name = Column(String(50), nullable=False) # 姓名(非空) email = Column(String(100), unique=True, index=True) # 邮箱(唯一) password_hash = Column(String(255), nullable=False) # 密码哈希(不存明文) -
schema.py:用 Pydantic 定义请求体 / 响应体模型—— 规定客户端 “传什么”、服务器 “返回什么”,避免数据格式混乱。示例(models/schema.py):from pydantic import BaseModel, EmailStr # EmailStr:邮箱格式校验 # 创建用户的请求体模型(客户端需要传这些字段) class UserCreate(BaseModel): name: str # 姓名(字符串) email: EmailStr # 邮箱(自动校验格式,如xxx@xxx.com) password: str # 密码(字符串) # 返回用户的响应体模型(服务器只返回这些字段,隐藏敏感信息) class UserResponse(BaseModel): id: int # 用户ID name: str # 姓名 email: EmailStr # 邮箱 class Config: orm_mode = True # 允许从ORM对象(如SQLAlchemy的User)转换为Pydantic模型
-
-
crud/:CRUD 逻辑包CRUD 是 ** 增(Create)、删(Delete)、改(Update)、查(Read)** 的缩写 ——分离业务逻辑与接口逻辑,让代码更易维护。示例(crud/user_crud.py):from sqlalchemy.orm import Session from models.db_model import User from models.schema import UserCreate from core.security import hash_password # 密码哈希函数(来自core/security.py) # 查询用户(根据ID) def get_user(db: Session, user_id: int): return db.query(User).filter(User.id == user_id).first() # 创建用户(哈希密码,避免存明文) def create_user(db: Session, user: UserCreate): password_hash = hash_password(user.password) # 哈希密码 db_user = User( name=user.name, email=user.email, password_hash=password_hash ) db.add(db_user) # 添加到数据库会话 db.commit() # 提交到数据库(真正写入) db.refresh(db_user) # 刷新对象,获取自增的ID return db_user -
core/:核心配置包-
config.py:读取环境变量(如数据库 URL、JWT 秘钥),避免把敏感信息写死在代码里。示例(core/config.py):from pydantic_settings import BaseSettings # Pydantic的设置类(代替旧版BaseSettings) class Settings(BaseSettings): DATABASE_URL: str = "sqlite:///./test.db" # SQLite数据库URL JWT_SECRET_KEY: str = "your-secret-key-keep-it-safe" # JWT秘钥(要保密) JWT_EXPIRE_MINUTES: int = 30 # JWT过期时间(分钟) class Config: env_file = ".env" # 从.env文件读取环境变量(比如DATABASE_URL=postgresql://user:pass@localhost/db) settings = Settings() # 全局设置实例,其他模块可导入 -
security.py:处理安全逻辑(密码哈希、JWT 令牌生成),避免敏感操作散落在代码中。示例(core/security.py):from passlib.context import CryptContext # 密码哈希库 from jose import JWTError, jwt # JWT库 from datetime import datetime, timedelta from core.config import settings # 密码哈希上下文(用bcrypt算法) pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # 哈希密码(存数据库) def hash_password(password: str) -> str: return pwd_context.hash(password) # 验证密码(对比明文与哈希值) def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) # 生成JWT令牌(登录时返回给客户端) def create_access_token(data: dict): to_encode = data.copy() # 过期时间:当前时间+30分钟 expire = datetime.utcnow() + timedelta(minutes=settings.JWT_EXPIRE_MINUTES) to_encode.update({"exp": expire}) # 添加过期时间 # 编码生成令牌(用秘钥和HS256算法) return jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm="HS256")
-
-
db/:数据库连接包负责创建数据库引擎(Python 与数据库的连接)和Session(操作数据库的 “把手”)。示例(db/database.py):from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from core.config import settings from models.db_model import Base # 数据库模型基类 # 创建数据库引擎(连接数据库) # SQLite需要加connect_args={"check_same_thread": False}(避免线程问题) engine = create_engine( settings.DATABASE_URL, connect_args={"check_same_thread": False} ) # Session工厂:每次请求生成一个新Session(避免线程安全问题) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 初始化数据库(启动时创建所有表) def init_db(): Base.metadata.create_all(bind=engine) # 根据Base模型创建表 # 获取数据库Session(依赖注入用) def get_db(): db = SessionLocal() try: yield db # 生成Session,供接口使用 finally: db.close() # 请求结束后关闭Session(必须!否则连接池会满) -
main.py:主文件 ——FastAPI 的 “启动入口”整合所有模块,创建 FastAPI 实例,启动服务。示例(main.py):from fastapi import FastAPI from api import user, order, product # 导入api模块 from db.database import init_db # 数据库初始化函数 from core.config import settings # 创建FastAPI实例(标题、版本会显示在文档中) app = FastAPI(title="FastAPI 用户管理系统", version="1.0.0") # 启动事件:服务启动时自动创建数据库表 @app.on_event("startup") def on_startup(): init_db() # 整合api模块的路由(把用户、订单、商品接口加进来) app.include_router(user.router) # 用户接口(前缀/users) app.include_router(order.router) # 订单接口(前缀/orders) app.include_router(product.router)# 商品接口(前缀/products) # 测试接口(访问http://localhost:8000/会返回这个结果) @app.get("/") def read_root(): return {"message": "FastAPI 服务已启动!"}
我的踩坑经历:刚开始写 FastAPI 时,我把所有代码堆在main.py里 —— 用户、订单、数据库逻辑全混在一起,500 行代码找 BUG 要翻半小时。后来用包组织代码,逻辑瞬间清晰—— 现在改用户接口只需要动api/user.py和crud/user_crud.py,不用碰其他文件。
6. 前置知识 6:Pydantic——FastAPI 的 “数据模型引擎”
如果说 FastAPI 是 “汽车”,Pydantic 就是 “发动机”——FastAPI 的所有数据校验、序列化 / 反序列化都靠它。
6.1 什么是 Pydantic?
Pydantic 是 Python 的数据验证库,帮你解决 3 个核心问题:
- 校验数据格式(比如邮箱是不是
xxx@xxx.com,年龄是不是整数); - 转换数据类型(比如把 JSON 字符串转成 Python 的 int,把 ORM 对象转成 JSON);
- 定义数据结构(用类代替字典,数据格式更清晰)。
6.2 Pydantic 的核心:BaseModel
所有 Pydantic 模型都要继承BaseModel—— 它是 “数据模型的模板”。
(1)字段校验:Pydantic 的 “数据安检机”
Pydantic 支持多种校验规则,比如:
- 类型校验(
name: str、age: int); - 格式校验(
email: EmailStr(邮箱)、phone: constr(regex=r"^\d{11}$")(11 位手机号)); - 约束校验(
age: conint(ge=18)(年龄≥18)、password: constr(min_length=8)(密码≥8 位))。
示例:严格的用户注册模型
from pydantic import BaseModel, EmailStr, constr, conint
class UserCreate(BaseModel):
name: constr(min_length=2, max_length=50) # 姓名2-50字
email: EmailStr # 邮箱格式校验
age: conint(ge=18, le=100) # 年龄18-100岁
# 密码:至少8位,含大小写字母+数字
password: constr(min_length=8, regex=r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).+$")
(2)FastAPI 中的 Pydantic 应用:请求体与响应体
- 请求体:用 Pydantic 模型定义客户端要传的数据(比如
UserCreate),FastAPI 会自动解析 JSON 请求体; - 响应体:用 Pydantic 模型定义服务器要返回的数据(比如
UserResponse),避免返回敏感信息(如password_hash)。
示例:创建用户接口
from fastapi import FastAPI
from models.schema import UserCreate, UserResponse
from crud.user_crud import create_user
from db.database import get_db
from sqlalchemy.orm import Session
app = FastAPI()
@app.post("/users", response_model=UserResponse)
def create_user_api(user: UserCreate, db: Session = Depends(get_db)):
return create_user(db, user)
当客户端发送以下 JSON 请求:
{
"name": "张三",
"email": "zhangsan@example.com",
"age": 20,
"password": "Zhangsan123"
}
FastAPI 会自动做 3 件事:
- 把 JSON 转成
UserCreate实例(反序列化); - 校验
email格式、age范围、password复杂度; - 调用
create_user逻辑,返回UserResponse模型(只含id、name、email)。
如果客户端传的email是zhangsan.example.com(少@),FastAPI 会返回422 错误(参数校验失败),并提示:
{
"detail": [
{
"loc": ["body", "email"],
"msg": "value is not a valid email address",
"type": "value_error.email"
}
]
}
我的踩坑经历:刚开始用 Pydantic 时,我忘了给UserResponse加orm_mode=True—— 结果返回的 JSON 是空的!后来查文档才知道,orm_mode=True允许 Pydantic 从ORM 对象(如 SQLAlchemy 的User)中读取数据。加了之后,立刻返回正确的id、name、email。
7. 前置知识 7:异步编程(async/await)——FastAPI 的 “高并发秘诀”
FastAPI 的一大优势是支持异步—— 能处理高并发请求(比如同时 1000 个用户访问),而同步框架(如 Flask)只能逐个处理。
7.1 什么是异步编程?
异步编程是 **“非阻塞” 的编程方式 **—— 当遇到IO 密集型任务(比如数据库查询、网络请求、文件读写)时,程序不会 “等” 任务完成,而是去做其他事,等任务完成后再回来处理结果。
生活例子:同步 vs 异步
- 同步(阻塞):点外卖后,你一直盯着手机等骑手,期间什么都不做 —— 直到骑手打电话,你才去取外卖;
- 异步(非阻塞):点外卖后,你去做饭、打扫卫生(做其他事),骑手到了会给你打电话(触发 “回调”),你再去取外卖。
专业解释:IO 密集型 vs CPU 密集型
异步编程只对 IO 密集型任务有效,对 CPU 密集型任务(比如复杂计算、图像处理)无效 —— 因为 CPU 密集型任务需要占用 CPU 一直运算,无法 “挂起” 去做其他事。
| 任务类型 | 例子 | 异步是否有效 |
|---|---|---|
| IO 密集型 | 数据库查询、网络请求、文件读写 | ✅ 有效 |
| CPU 密集型 | 计算 π 的 100 万位、图像识别 | ❌ 无效 |
7.2 异步编程的核心语法:async/await
Python 的异步编程依赖两个关键字:
async def:定义异步函数(无法用普通的def定义);await:等待异步操作完成(只能在异步函数里用)。
示例:异步函数的基本用法
import asyncio # 异步IO库(Python内置)
# 异步函数(模拟IO密集型任务:比如数据库查询)
async def fetch_data(delay: int):
print(f"开始获取数据,延迟{delay}秒...")
await asyncio.sleep(delay) # 模拟IO等待(非阻塞)
print("数据获取完成!")
return {"data": "异步数据"}
# 主异步函数(必须用async def定义)
async def main():
# 同时启动两个异步任务(并发执行)
task1 = asyncio.create_task(fetch_data(2)) # 任务1:延迟2秒
task2 = asyncio.create_task(fetch_data(3)) # 任务2:延迟3秒
# 等待两个任务完成
result1 = await task1
result2 = await task2
print(f"结果1:{result1}")
print(f"结果2:{result2}")
# 启动异步程序(Python 3.7+可用)
asyncio.run(main())
运行结果(关键看顺序):
开始获取数据,延迟2秒...
开始获取数据,延迟3秒...
数据获取完成!(任务1,2秒后)
数据获取完成!(任务2,3秒后)
结果1:{"data": "异步数据"}
结果2:{"data": "异步数据"}
解释:
asyncio.create_task():把异步函数包装成 “任务”,提交给事件循环(Event Loop)管理;- 事件循环是异步编程的 “调度器”:负责安排任务的执行顺序,比如先执行任务 1,然后挂起(等待 IO),再执行任务 2,等任务 1 完成后再处理结果;
- 两个任务并发执行:总时间是 3 秒(任务 2 的延迟),而不是 2+3=5 秒(同步的情况)。
7.3 FastAPI 中的异步接口
FastAPI 的接口可以用 **async def定义 —— 只要接口中有IO 密集型任务 **(比如异步数据库查询、异步 HTTP 请求),就应该用异步接口提升并发。
示例 1:异步接口(调用异步数据库操作)
假设你用异步 SQLAlchemy(支持异步的数据库库),异步查用户的接口:
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession # 异步Session
from models.db_model import User
from db.database import get_async_db # 异步数据库连接(返回AsyncSession)
app = FastAPI()
# 异步接口(async def)
@app.get("/users/{id}")
async def get_user(id: int, db: AsyncSession = Depends(get_async_db)):
# 异步查询数据库(await等待结果)
user = await db.get(User, id) # get是异步方法,必须用await
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
示例 2:异步接口(调用异步 HTTP 请求)
用aiohttp(异步 HTTP 库)调用外部接口:
import aiohttp
from fastapi import FastAPI
app = FastAPI()
# 异步接口
@app.get("/external-data")
async def get_external_data():
# 异步HTTP请求(aiohttp.ClientSession是异步的)
async with aiohttp.ClientSession() as session:
async with session.get("https://api.github.com/users/octocat") as response:
data = await response.json() # 等待JSON数据
return {"github_data": data}
7.4 异步的 “踩坑点”(必看!)
异步编程虽好,但踩坑率极高—— 尤其是小白,容易写出 “假异步” 代码。
坑 1:在异步函数里调用同步函数
如果在async def接口里调用同步函数(比如同步的数据库查询、同步的文件读写),会阻塞整个事件循环—— 导致所有异步任务都卡住,异步失效。
错误示例:
from fastapi import FastAPI
from sqlalchemy.orm import Session # 同步Session
from db.database import get_sync_db # 同步数据库连接
app = FastAPI()
# 错误:async def接口里调用同步函数(get_sync_db返回同步Session)
@app.get("/users/{id}")
async def get_user(id: int, db: Session = Depends(get_sync_db)):
user = db.query(User).get(id) # 同步查询(阻塞事件循环)
return user
解决方法:改用异步的库(比如异步 SQLAlchemy、asyncpg、aiohttp),确保接口里的所有 IO 操作都是异步的。
坑 2:异步函数里没有 await
如果异步函数里没有await,函数会立即执行完毕,不会等待异步操作。
错误示例:
async def fetch_data(delay: int):
print("开始获取数据...")
asyncio.sleep(delay) # 没有用await,函数直接返回
print("数据获取完成!") # 不会执行
解决方法:所有异步操作都要加await——await asyncio.sleep(delay)。
坑 3:CPU 密集型任务用异步
如果接口是CPU 密集型任务(比如计算 π 的 100 万位),用async def会更慢—— 因为异步需要切换任务的开销,而 CPU 密集型任务不需要等待。
解决方法:CPU 密集型任务用多线程或多进程(比如concurrent.futures.ThreadPoolExecutor),不要用异步。
7.5 什么时候用异步?
- 推荐用异步:接口中有 IO 密集型任务(比如数据库查询、网络请求、文件读写),需要高并发;
- 不推荐用异步:接口是 CPU 密集型任务(比如复杂计算),或没有 IO 操作(比如纯返回字符串)。
我的踩坑经历:刚开始用 FastAPI 时,我写了个异步接口,里面调用同步的 SQLAlchemy 查询 —— 结果并发量比同步接口还低!后来才知道,同步函数会阻塞事件循环,导致所有异步任务都等着,这时候要改用异步的数据库库(比如sqlalchemy.ext.asyncio),接口才真正异步起来。
8. 前置知识 8:ASGI 与 UVicorn——FastAPI 的 “服务器引擎”
FastAPI 是ASGI 框架(异步服务器网关接口),需要用 ASGI 服务器(比如 Uvicorn、Hypercorn)才能运行 —— 就像 Flask(WSGI 框架)需要用 Gunicorn、Waitress 服务器一样。
8.1 什么是 ASGI?
ASGI 是Async Server Gateway Interface的缩写 —— 是 Python 异步 Web 框架的 “通信协议”,定义了服务器与框架之间的接口。
对比 WSGI(同步):
- WSGI:只能处理同步请求,每个请求对应一个线程;
- ASGI:能处理同步 + 异步请求,支持 WebSocket、长连接(比如聊天应用)。
8.2 为什么用 Uvicorn?
Uvicorn 是高性能的 ASGI 服务器,专门为 FastAPI 设计 —— 速度快、稳定、支持异步,是 FastAPI 的 “官方推荐服务器”。
如何运行 FastAPI?
用 Uvicorn 运行main.py中的app实例:
uvicorn main:app --reload --port 8000
main:app:main.py中的app实例(FastAPI 对象);--reload:开发模式(代码修改后自动重启);--port 8000:端口号(访问http://localhost:8000)。
8.3 生产环境的部署
开发时用--reload没问题,但生产环境不能用—— 因为reload会消耗更多资源,且不稳定。生产环境的命令:
uvicorn main:app --host 0.0.0.0 --port 80 --workers 4
--host 0.0.0.0:允许所有 IP 访问(服务器对外提供服务);--port 80:默认 HTTP 端口(不用输端口号就能访问);--workers 4:启动 4 个工作进程(提升并发量,根据 CPU 核心数调整)。
9. 前置知识 9:依赖注入(Dependency Injection)——FastAPI 的 “模块化工具”
依赖注入是 FastAPI 的核心特性之一—— 帮你把 “重复的逻辑”(比如数据库连接、权限校验)抽成 “依赖项”,在接口中复用。
9.1 什么是依赖注入?
依赖注入是 **“控制反转”(IoC)** 的一种实现方式 —— 简单说:
- 你需要某个 “资源”(比如数据库 Session、用户信息),不用自己创建,让框架给你 “注入”;
- 这个 “资源” 就是依赖项(Dependency)。
生活例子:餐厅吃饭
- 你去餐厅吃饭,需要 “餐具”—— 不用自己带,餐厅会给你 “注入”(摆到桌上);
- 餐具就是 “依赖项”,餐厅就是 “依赖注入框架”。
9.2 FastAPI 中的依赖注入
FastAPI 用Depends()函数实现依赖注入 —— 把依赖项作为接口的参数,FastAPI 会自动创建依赖项的实例,传给接口。
示例 1:数据库连接的依赖项
之前的get_async_db就是一个依赖项 —— 负责创建异步数据库 Session:
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from core.config import settings
# 创建异步数据库引擎(连接数据库)
engine = create_async_engine(settings.DATABASE_URL)
# 创建异步Session工厂
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# 依赖项:获取异步数据库Session
async def get_async_db():
async with AsyncSessionLocal() as session:
yield session # 生成Session,供接口使用
await session.commit() # 请求结束后提交(如果有修改)
然后在接口中用Depends(get_async_db)注入 Session:
@app.get("/users/{id}")
async def get_user(
id: int,
db: AsyncSession = Depends(get_async_db) # 注入异步Session
):
user = await db.get(User, id)
return user
示例 2:权限校验的依赖项
抽一个 “获取当前用户” 的依赖项,所有需要登录的接口都能复用:
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from core.config import settings
from crud.user_crud import get_user
from db.database import get_async_db
# OAuth2密码模式:客户端需要传Authorization头,格式是"Bearer token"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")
# 依赖项:获取当前用户(解析JWT令牌)
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_async_db)
):
credentials_exception = HTTPException(
status_code=401,
detail="令牌无效或已过期",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# 解析JWT令牌
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
user_id: int = payload.get("sub") # sub是令牌中的用户ID
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
# 查询当前用户
user = await get_user(db, user_id)
if user is None:
raise credentials_exception
return user
然后在需要登录的接口中注入get_current_user:
@app.get("/my-orders")
async def get_my_orders(
current_user: User = Depends(get_current_user), # 注入当前用户
db: AsyncSession = Depends(get_async_db)
):
# 查询当前用户的订单
orders = await db.query(Order).filter(Order.user_id == current_user.id).all()
return orders
9.3 依赖注入的好处
- 代码复用:依赖项只写一次,所有接口都能复用(比如
get_async_db); - 逻辑分离:接口只关注业务逻辑,依赖项关注 “资源创建”(比如数据库连接);
- 易测试:测试时可以替换依赖项(比如用测试数据库代替真实数据库)。
我的踩坑经历:刚开始用 FastAPI 时,我在每个接口里都写一遍 “获取当前用户” 的逻辑 —— 代码重复了 10 次!后来用依赖注入,把 “获取当前用户” 抽成get_current_user,所有需要登录的接口只要加current_user: User = Depends(get_current_user)就行,代码量减少了 80%。
10. 前置知识 10:OpenAPI 与 Swagger——FastAPI 的 “自动文档”
FastAPI 会自动生成接口文档—— 基于 OpenAPI 标准,支持 Swagger UI 和 Redoc 两种文档风格。
10.1 什么是 OpenAPI?
OpenAPI 是API 描述标准—— 定义了如何用 JSON/YAML 描述 API 的接口、参数、响应等信息。FastAPI 会根据你的代码(类型提示、依赖项、路径参数)自动生成 OpenAPI 文档。
10.2 如何查看自动文档?
运行 FastAPI 服务后,访问以下地址:
- Swagger UI:
http://localhost:8000/docs(交互式文档,能直接测试接口); - Redoc:
http://localhost:8000/redoc(更简洁的文档,适合阅读)。
Swagger UI 的示例
打开http://localhost:8000/docs,你会看到:
- 所有接口的列表(按标签分类,比如 “用户管理”);
- 每个接口的参数(路径参数、查询参数、请求体);
- 每个参数的类型(比如
id: int); - “Try it out” 按钮:能直接发送请求,测试接口(不用 Postman)。
10.3 自动文档的好处
- 节省时间:不用手动写文档,代码变文档也变;
- 降低沟通成本:前端和后端都能看同一个文档,避免 “接口参数不一致”;
- 方便测试:直接在文档里测试接口,不用 Postman。
总结:FastAPI 的 “全链路逻辑” 与小白学习建议
到这里,你已经掌握了 FastAPI 的所有前置知识 —— 接下来,我把这些知识串成一个完整的接口处理流程,帮你彻底打通 “从 HTTP 请求到响应” 的逻辑:
一、FastAPI 的 “全链路逻辑”(用 “查用户” 接口举例)
假设客户端发送一个GET 请求:http://localhost:8000/users/123(查询 ID 为 123 的用户),FastAPI 会做以下 8 件事:
- 匹配接口:根据 URL
/users/{id}匹配到用@app.get("/users/{id}")装饰的接口函数; - 解析路径参数:从 URL 中提取
id=123,用函数的id: int类型提示校验(确保是整数); - 注入依赖项:调用
get_async_db依赖项,生成异步数据库 Session,传给接口的db参数; - 执行接口逻辑:调用
db.get(User, 123)异步查询数据库,获取用户对象; - 处理业务逻辑:如果用户不存在,抛出
404错误;如果存在,继续; - 序列化响应:用
UserResponsePydantic 模型,把用户对象转成 JSON(隐藏password_hash等敏感字段); - 返回响应:返回
200状态码和 JSON 数据({"id": 123, "name": "张三", "email": "zhangsan@example.com"}); - 清理资源:依赖项
get_async_db中的yield后代码执行,关闭数据库 Session。
这个流程里,所有底层细节都被 FastAPI 封装了—— 你只需要写接口函数、数据模型、依赖项,剩下的交给框架。
二、给 0 基础小白的学习建议(必看!)
FastAPI 是最适合小白入门的 Python Web 框架(没有之一),但学习时要注意 **“循序渐进”**,不要一开始就写复杂项目。以下是我总结的 “小白学习路线”:
1. 第一步:搞定 Python 基础(2-3 天)
FastAPI 的核心是 Python—— 如果 Python 基础不牢,学框架会像 “空中楼阁”。必学内容:
- 函数(
def、参数、返回值、类型提示); - 类与对象(
class、__init__、继承); - 模块与包(
import、__init__.py); - 异常处理(
try/except、raise)。
2. 第二步:快速入门 FastAPI(1-2 天)
先写最基础的接口,建立 “框架能跑通” 的信心:
- 安装 FastAPI 和 Uvicorn(
pip install fastapi uvicorn); - 写一个
Hello World接口:from fastapi import FastAPI app = FastAPI() @app.get("/") def read_root(): return {"message": "Hello FastAPI!"} - 运行服务(
uvicorn main:app --reload),访问http://localhost:8000看结果; - 打开
http://localhost:8000/docs,用 Swagger UI 测试接口(点 “Try it out”→“Execute”)。
3. 第三步:学习核心功能(3-5 天)
逐个攻克 FastAPI 的核心功能,每学一个功能就写一个小例子:
- 路径参数:写
/users/{id}接口,返回{"user_id": id}; - 查询参数:写
/items/?skip=0&limit=10接口,返回{"skip": skip, "limit": limit}; - 请求体:用 Pydantic 的
BaseModel定义UserCreate模型,写/users接口创建用户; - 依赖注入:写一个
get_current_user依赖项,模拟 “登录校验”; - 自动文档:用
tags给接口分类(比如tags=["用户管理"]),看 Swagger UI 的效果。
4. 第四步:做一个小项目(5-7 天)
用最小可行性项目(MVP)巩固所有知识 —— 比如 “用户管理系统”,涵盖以下功能:
- 接口:创建用户(
POST /users)、查询用户(GET /users/{id})、删除用户(DELETE /users/{id}); - 数据库:用异步 SQLAlchemy 连接 SQLite(不用装数据库,文件存储);
- 数据模型:用 Pydantic 定义
UserCreate(请求体)和UserResponse(响应体); - 权限校验:用 JWT 生成登录令牌,需要登录的接口用
get_current_user依赖项; - 自动文档:用 Swagger UI 测试所有接口,确保能正常工作。
5. 第五步:进阶学习(按需)
如果想深入,可以学以下内容:
- WebSocket:用 FastAPI 写实时聊天接口;
- 中间件:写一个记录请求日志的中间件;
- 测试:用
pytest测试接口; - 部署:用 Docker+Nginx 部署 FastAPI 到服务器。
三、小白常见问题解答
最后,解答几个小白最常问的问题,帮你避坑:
Q1:学 FastAPI 需要先学 Django 吗?
不需要!FastAPI 和 Django 是完全不同的框架——Django 是 “大而全” 的同步框架,适合做复杂的 CMS、电商系统;FastAPI 是 “轻量级” 的异步框架,适合做 API、微服务。小白直接学 FastAPI 更简单。
Q2:需要学前端知识吗?
不需要!FastAPI 是后端框架,负责处理数据逻辑,前端可以用 Vue、React 或直接调用接口(比如用 Postman)。
Q3:遇到问题怎么办?
- 先看FastAPI 官方文档(最权威,中文翻译很全:https://fastapi.tiangolo.com/zh/);
- 再查CSDN / 博客园的文章(搜 “FastAPI 422 错误”“FastAPI JWT” 等关键词);
- 最后问ChatGPT或技术社区(比如知乎、V2EX)—— 记得把问题描述清楚(比如报错信息、代码片段)。
写在最后
FastAPI 是我用过最 “贴心” 的框架—— 它把所有复杂的底层细节都封装了,却给了你足够的灵活性;它的自动文档、类型提示、依赖注入,都是为了 “让开发者少写代码,多做有意义的事”。
作为小白,你不需要一开始就掌握所有细节 ——先跑通一个接口,再写一个小项目,慢慢迭代。等你写了 10 个接口、1 个项目,再回头看这些前置知识,会发现 “原来所有逻辑都是通的”。
最后,送你一句我学编程时的座右铭:“代码是写出来的,不是看出来的”—— 放下顾虑,打开 VS Code,开始写第一个 FastAPI 接口吧!
祝你学习顺利!—— 一个踩过无数坑的 FastAPI 开发者
更多推荐
所有评论(0)