FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,使用 Python 并基于标准的 Python 类型提示。
特性
- 快速:可与 NodeJS 和 Go 并肩的极高性能(归功于 Starlette 和 Pydantic快速编码:提高功能开发速度约 200% 至 300更少bug:减少约 40% 的人为(开发者)导致错误智能:强大的编辑器支持简单:设计的易于使用和学习,阅读文档的时间更短简短:使代码重复最小化健壮:生产可用级别的代码。还有自动生成的交互式文档标准化:基于(并完全兼容)API 开放标准
FastAPI 的核心
Uvicorn
官方文档:www.uvicorn.org/
- Uvicorn 是一个基于 uvloop 和 httptools 实现的高性能 ASGI 服务器,可以实现异步处理 HTTP 请求。FastAPI 使用 Uvicorn 作为其默认的 Web 服务器,是因为 Uvicorn 是一个非常快速、可靠且易于使用的 ASGI 服务器,可以在处理大量并发连接时保持稳定和高效。Uvicorn 还支持 WebSocket 和 HTTP/2 等新特性,符合 FastAPI 提倡的现代 Web 开发理念。因此,使用 Uvicorn 作为 FastAPI 的 Web 服务器是一个很好的选择
Starlette
官方文档:www.starlette.io/
Starlette 是一个轻量级的 ASGI(Asynchronous Server Gateway Interface)框架,专为构建异步 Web 应用设计。它是 FastAPI 的底层框架,提供了路由、请求处理、中间件、WebSocket 支持等核心功能。Starlette 的设计目标是简单、高效,同时保持足够的灵活性,适用于各种 Web 开发场景。拥有以下特性:
- 异步支持:基于 Python 的 asyncio,支持异步 I/O 操作,适合高并发场景。路由系统:提供直观的路径操作装饰器,用于定义 API 端点。请求与响应处理:内置对 HTTP 请求和响应的支持,支持 JSON、表单数据等格式。中间件:允许开发者在请求处理前后添加自定义逻辑。WebSocket 支持:支持实时双向通信,适用于聊天应用等场景。
FastAPI 直接继承了 Starlette 的所有功能,例如路由、请求处理和中间件。FastAPI 的 @app.get()、@app.post() 等装饰器实际上是对 Starlette 路由系统的封装。换句话说,FastAPI 在 Starlette 的基础上增加了类型检查、自动文档生成等高级特性
Pydantic
Pydantic 是一个基于 Python 类型提示(type hints)的数据验证和解析库,它能自动校验数据是否符合预定义的类型和约束,广泛用于 API 开发、配置管理、数据序列化等场景。其核心优势是结合了 Python 类型系统和运行时数据验证,既提升了代码可读性,又确保了数据合法性
FastAPI 与 Uvicorn、 Starlette、Pydantic的关系
FastAPI、Starlette、Uvicorn 和 Pydantic 是 Python 异步 Web 开发生态中紧密协作的四个核心组件,它们分工明确又相互配合,共同构成了高效的 API 开发体系。它们的关系可以概括为:
- Uvicorn:底层运行引擎(服务器)Starlette:基础 Web 框架(提供路由、请求处理等核心功能)Pydantic:数据验证工具(负责数据校验和类型处理)FastAPI:上层 API 框架(整合前三者,专注 API 开发)
具体关系解析
FastAPI 与 Uvicorn
- Uvicorn 是一个基于
asyncio
的 ASGI 服务器(异步服务器网关接口),负责处理底层网络通信FastAPI 和 Starlette 都是 ASGI 应用框架,必须依赖 Uvicorn(或其他 ASGI 服务器,如 Hypercorn)才能运行关系类比:Uvicorn 像 "服务器主机",FastAPI/Starlette 像运行在主机上的 "应用程序"FastAPI 与 Starlette
- FastAPI 直接继承自 Starlette,是 Starlette 的增强版Starlette 提供基础 Web 功能(路由、中间件、WebSocket、异步支持等)FastAPI 在 Starlette 基础上增加了 API 专用功能(自动文档、依赖注入等)关系类比:Starlette 是 "基础模板",FastAPI 是基于模板定制的 "API 专用版本"
FastAPI 与 Pydantic
- Pydantic 是独立的数据验证库,基于类型提示实现数据校验FastAPI 深度集成 Pydantic,将其作为数据验证的核心工具所有请求参数、响应模型都通过 Pydantic 模型定义和验证关系类比:Pydantic 是 "数据安检仪",确保进入 FastAPI 的数据符合规范
基础示例
from fastapi import FastAPI # 创建 FastAPI 应用实例 app = FastAPI() # 定义根路径的 GET 请求接口@app.get("/") # 处理根路径请求的函数 # 返回一个字典,内容为{"Hello": "World"} def read_root(): return {"Hello": "World"} # 定义带有路径参数的 GET 请求接口 @app.get("/items/{item_id}") # 处理带参数请求的函数,item_id 为路径参数,q 为可选查询参数 def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q} # 定义带有查询参数的 GET 请求接口@app.get("/items/")# 处理带查询参数请求的函数,skip 和 limit 为查询参数def read_items(skip: int = 0, limit: int = 10): return {"skip": skip, "limit": limit} # 仅在直接运行本文件时启动 uvicorn 服务器 if __name__ == "__main__": # 导入 uvicorn 服务器 import uvicorn # 启动应用,host 为 127.0.0.1,端口为 8000,自动重载 uvicorn.run(app, host="127.0.0.1", port=8000, reload=True)
运行以上 main.py 文件,启动服务访问: http://127.0.0.1:8000/docs ,可看到由Swagger UI生成的交互式API文档
请求体
# 导入 FastAPI 和 Pydantic 的 BaseModelfrom fastapi import FastAPIfrom pydantic import BaseModel# 定义请求体的数据模型class Item(BaseModel): # 名称,字符串类型 name: str # 描述,字符串类型,可选 description: str = None # 价格,浮点数类型 price: float # 税,浮点数类型,可选 tax: float = None# 创建 FastAPI 应用实例app = FastAPI()# 定义 POST 请求接口,接收 Item 类型的请求体@app.post("/items/")def create_item(item: Item): return itemif __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
响应模型
# 导入 FastAPI 和 Pydantic 的 BaseModelfrom fastapi import FastAPIfrom pydantic import BaseModel# 定义响应模型class Item(BaseModel): name: str description: str = None price: float tax: float = Noneapp = FastAPI()# 指定 response_model,返回的数据会自动校验和序列化@app.post("/items/", response_model=Item)def create_item(item: Item): return itemif __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
表单
# 导入 FastAPI 和 Formfrom fastapi import FastAPI, Formapp = FastAPI()# 定义 POST 接口,接收表单数据@app.post("/login/")def login(username: str = Form(...), password: str = Form(...)): return {"username": username}if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
文件上传
# 导入 FastAPI、UploadFile 和 Filefrom fastapi import FastAPI, UploadFile, Fileapp = FastAPI()# 上传原始字节文件@app.post("/files/")def create_file(file: bytes = File(...)): return {"file_size": len(file)}# 上传文件对象@app.post("/uploadfile/")def create_upload_file(file: UploadFile = File(...)): return {"filename": file.filename}if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
异常处理
# 导入 FastAPI 和 HTTPExceptionfrom fastapi import FastAPI, HTTPExceptionapp = FastAPI()# 假设有一个 items 字典存储数据items = {"foo": "The Foo item", "bar": "The Bar item"}# 定义带有异常处理的接口@app.get("/items/{item_id}")def read_item(item_id: str): # 如果 item_id 不在 items 中,抛出 404 异常 if item_id not in items: raise HTTPException(status_code=404, detail="Item not found") # 返回对应的 item return {"item": items[item_id]}if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
中间件
# 导入 FastAPI 和 Requestfrom fastapi import FastAPI, Requestimport timeapp = FastAPI()# 定义中间件,统计请求处理时间@app.middleware("http")async def add_process_time_header(request: Request, call_next): # 记录开始时间 start_time = time.time() # 处理请求 response = await call_next(request) # 计算处理时间 process_time = time.time() - start_time # 添加自定义响应头 response.headers["X-Process-Time"] = str(process_time) return response# 示例接口@app.get("/")async def root(): return {"message": "Hello World"}if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
依赖注入
# 导入 FastAPI 和 Dependsfrom fastapi import FastAPI, Dependsapp = FastAPI()# 定义依赖函数,返回查询参数def common_parameters(q: str = None, skip: int = 0, limit: int = 100): return {"q": q, "skip": skip, "limit": limit}# 使用 Depends 注入依赖@app.get("/items/")def read_items(commons: dict = Depends(common_parameters)): return commonsif __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
安全性
# 导入 FastAPI 和 OAuth2PasswordBearerfrom fastapi import FastAPI, Dependsfrom fastapi.security import OAuth2PasswordBearerapp = FastAPI()# 创建 OAuth2PasswordBearer 实例,指定 tokenUrloauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")# 依赖注入 token@app.get("/items/")def read_items(token: str = Depends(oauth2_scheme)): return {"token": token}if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
数据库链接
from fastapi import FastAPI, APIRouter, HTTPException, Dependsfrom pydantic import BaseModelfrom typing import Listfrom sqlalchemy import create_engine, Column, String, Integerfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker, Session# 初始化FastAPI应用app = FastAPI(title="用户管理API", description="一个简单的用户表CRUD示例")application = APIRouter()# 数据库配置 - MySQL# 请替换为你的MySQL数据库信息SQLALCHEMY_DATABASE_URL = ( "mysql+pymysql://username:password@localhost/fastapi?charset=utf8mb4")engine = create_engine(SQLALCHEMY_DATABASE_URL)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 数据库模型基类Base = declarative_base()# 用户模型class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) user_name = Column(String(100), index=True) # 为MySQL指定VARCHAR长度 user_age = Column(Integer)# 创建数据库表Base.metadata.create_all(bind=engine)# Pydantic模型 - 用于数据验证和序列化class UserBase(BaseModel): user_name: str user_age: intclass UserCreate(UserBase): passclass UserUpdate(UserBase): passclass UserInDB(UserBase): id: int class Config: orm_mode = True# 依赖项 - 获取数据库会话(修正的部分)def get_db() -> Session: db = SessionLocal() try: yield db finally: db.close()# CRUD操作实现# 创建用户 - 使用Depends(get_db)而非next(get_db)@application.post("/users/", response_model=UserInDB, summary="创建新用户")def create_user(user: UserCreate, db: Session = Depends(get_db)): # 检查用户名是否已存在 db_user = db.query(User).filter(User.user_name == user.user_name).first() if db_user: raise HTTPException(status_code=400, detail="用户名已存在") # 创建新用户 db_user = User(user_name=user.user_name, user_age=user.user_age) db.add(db_user) db.commit() db.refresh(db_user) return db_user# 获取所有用户@application.get("/users/", response_model=List[UserInDB], summary="获取所有用户")def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): users = db.query(User).offset(skip).limit(limit).all() return users# 根据ID获取用户@application.get("/users/{user_id}", response_model=UserInDB, summary="根据ID获取用户")def read_user(user_id: int, db: Session = Depends(get_db)): db_user = db.query(User).filter(User.id == user_id).first() if db_user is None: raise HTTPException(status_code=404, detail="用户不存在") return db_user# 更新用户@application.put("/users/{user_id}", response_model=UserInDB, summary="更新用户信息")def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)): db_user = db.query(User).filter(User.id == user_id).first() if db_user is None: raise HTTPException(status_code=404, detail="用户不存在") # 检查新用户名是否已被其他用户使用 if user.user_name != db_user.user_name: existing_user = db.query(User).filter(User.user_name == user.user_name).first() if existing_user: raise HTTPException(status_code=400, detail="用户名已存在") # 更新用户信息 db_user.user_name = user.user_name db_user.user_age = user.user_age db.commit() db.refresh(db_user) return db_user# 删除用户@application.delete("/users/{user_id}", summary="删除用户")def delete_user(user_id: int, db: Session = Depends(get_db)): db_user = db.query(User).filter(User.id == user_id).first() if db_user is None: raise HTTPException(status_code=404, detail="用户不存在") db.delete(db_user) db.commit() return {"message": f"用户 {user_id} 已成功删除"}# 添加路由app.include_router(application, prefix="/demo", tags=["用户操作"])if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)
运行上述main.py文件,访问:http://127.0.0.1:8000/docs ,可以看到如下交互API页面,就可以对fastapi数据库的user表进行增删改查