本教程介绍了 FastAPI 的基本使用，包括接口定义、数据模型、数据校验、依赖注入、中间件及接口文档。通过示例代码，展示了如何快速构建高性能 API 服务，适用于入门学习和实际开发。
GitHub：https://github.com/fastapi/fastapi
Document：https://fastapi.tiangolo.com/
conda create -n fastapi-env python=3.10 pip install uvicorn fastapi
1. 基本使用
这一部分讲解 FastAPI 的基本 API 定义方式，包括无参数、路径参数和请求参数的使用。
from fastapi import FastAPI
import uvicorn
# 初始化 api 应用
app = FastAPI()
# 1. 无参数
@app.post('/demo01')
async def demo01():
return {'result': 100}
# 2. 路径参数
@app.post('/demo02/{param}')
async def demo02(param):
return {'param': param}
# 3. 请求参数
@app.post('/demo03')
async def demo03(param):
return {'param': param}
if __name__ == '__main__':
# 使用 uvicorn 启动 api 应用
uvicorn.run(app, host='127.0.0.1', port=8000)
我们可以使用 curl 工具来请求定义的 3 个服务 url，命令如下：
# 无请求参数 curl -X POST http://127.0.0.1:8000/demo01 # 路径参数 curl -X POST http://127.0.0.1:8000/demo02/666 # 请求参数 curl -X POST http://127.0.0.1:8000/demo03?param=300
2. 数据模型
这一部分介绍 Pydantic 数据模型的定义，并通过 response_model 指定返回数据格式。
import uvicorn
from pydantic import BaseModel
from fastapi import FastAPI
app = FastAPI()
class Request(BaseModel):
name : str = 'undefined'
age : int = 0
class Status(BaseModel):
is_status1 : bool = False
is_status2 : bool = False
class Response(BaseModel):
grade: int
score: int
status : Status = {True, True}
# response_model 设置响应数据类型
@app.post('/demo', response_model=Response)
# request 设置传递进来的数据类型
async def demo(request : Request):
return {'grade': 1001, 'score': 85, 'status': Status(is_status1=True, is_status2=False)}
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=8000)
curl -X POST http://127.0.0.1:8000/demo -H "Content-Type: application/json" -d "{\"name\": \"smith\", \"age\": 25}"
3. 数据校验
这一部分展示如何使用 Pydantic 的 Field 进行数据校验，包括字符串格式、数值范围及小数精度等。
import uvicorn
from pydantic import BaseModel
from pydantic import Field
from fastapi import FastAPI
from fastapi import Path
from fastapi import Query
from decimal import Decimal
app = FastAPI()
# 1. 请求和响应数据校验
class Data(BaseModel):
# 字段默认指是undefined，最小长度2，最大长度10，必须满足前面是字母后面是数字的格式
name : str = Field('undefined', min_length=2, max_length=10, pattern='^[a-zA-Z]+[0-9]+$')
# 字段默认值是0，必须大于等于10，小于等于20，同时必须是2的倍数
age : int = Field(0, ge=10, le=20, multiple_of=2)
# 字段默认值是0.0，整个数字最大位数是5，小数部分位数是2，整数部分位数为3
# 注意：Decimal 和 float 都是表示小数的类型，不同的是前者使用十进制方式进行存储，精度更好
salary : Decimal = Field(0.0, max_digits=5, decimal_places=2)
@app.post('/demo01', response_model=Data)
async def demo01(data : Data):
return data
# 2. 路径和请求参数校验
# path_param 为路径参数
# query_param 为请求参数
@app.post('/demo02/{path_param}')
# path_param 最小长度2，最大长度5，注意: Path 不可以设置默认值
# query_param 默认值 0，必须大于等于0并且小于20
async def demo02(path_param : str = Path(min_length=2, max_length=5),
query_param : int = Query(0, ge=0, lt=20)):
return {'path_param': path_param, 'query_param': query_param}
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=8000)
4. 依赖注入
这一部分讲解 FastAPI 依赖注入机制，包括全局依赖、局部依赖及多层依赖的使用。
依赖注入（Dependency Injection，简称 DI）是一种设计模式，它允许我们将对象的依赖关系从对象本身的创建和管理中分离出来。在 FastAPI 中，依赖注入用于将一些通用的逻辑（如身份验证、数据库连接等）封装到独立的函数中，然后在路径操作函数中使用这些依赖项。
from fastapi import FastAPI
from fastapi import Depends
import uvicorn
# 1. 全局依赖注入，每次请求都会执行依赖的可调用对象
def global_dependency1():
print('全局依赖注入1')
def global_dependency2():
print('全局依赖注入2')
app = FastAPI(dependencies=[Depends(global_dependency1), Depends(global_dependency2)])
# 2. 局部依赖注入，只有执行该请求时才会执行依赖可调用对象
# 2.1 函数作为依赖项
async def get_config():
return {'c1': 10, 'c2': 20}
@app.post('/demo01')
async def demo01(param : dict = Depends(get_config)):
return {'param': param}
# 2.2 类作为依赖项
class Config:
def __init__(self):
self.c1 = 10
self.c2 = 20
@app.post('/demo02')
async def demo02(param : Config = Depends()):
param.c1 = 100
param.c2 = 200
return {'param': param}
# 2.3 多层依赖项
async def get_dependency01():
return {'c1': 1000, 'c2': 2000}
async def get_dependency02(param : dict = Depends(get_dependency01)):
return param
@app.post('/demo03')
async def demo03(param : dict = Depends(get_dependency02)):
return {'param': param}
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=8000)
5. 中间件
这一部分介绍如何使用中间件拦截 HTTP 请求和响应，实现请求处理逻辑。当应用中存在多个中间件时，它们会按照注册的顺序依次执行。请求会先经过第一个中间件，然后依次经过后续的中间件，最后到达路径操作函数；响应则会按照相反的顺序经过各个中间件。
from fastapi import FastAPI
from fastapi import Request
from fastapi import Response
import uvicorn
app = FastAPI()
# 第一个中间件
@app.middleware('http')
async def middleware1(request: Request, call_next):
print('middleware1:', request, call_next)
response : Response = await call_next(request)
print('middleware1:', response)
return response
# 第二个中间件
@app.middleware('http')
async def middleware2(request: Request, call_next):
print('middleware2:', request, call_next)
response : Response = await call_next(request)
print('middleware2:', response)
return response
@app.post('/demo')
async def demo():
return {'result': 'ok'}
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=8000)
6. 接口文档
这一部分讲解如何配置 FastAPI 的 API 文档，包括全局信息、请求和响应参数说明及接口描述。启动 uvicorn 服务之后，可以：
- 通过 http://127.0.0.1:8000/docs 访问交互式文档
- 通过 http://127.0.0.1:8000/redoc 访问另外一种形式的文档
from fastapi import FastAPI
from fastapi import Query
from fastapi import Path
from pydantic import BaseModel
from pydantic import Field
import uvicorn
# 1. 定义全局文档信息
license_info={'name': 'Apache 2.0', 'url': 'https://www.apache.org/licenses/LICENSE-2.0.html'}
contact = {'name': 'Edward Meng', 'url': 'https://mengbaoliang.cn', 'email': 'chinapp@foxmail.com'}
terms_of_service = 'http://example.com/terms/'
app = FastAPI(title='接口文档',
description='该接口提供了丰富的 LLM 访问接口.',
version='1.0.0',
redoc_url='/docs1',
docs_url='/docs2',
terms_of_service=terms_of_service,
license_info=license_info,
contact=contact)
# 2. 定义请求和响应参数文档信息
class MyRequest(BaseModel):
name : str = Field(title='姓名', description='输入者的姓名')
age : int = Field(title='年龄', description='输入者的年龄')
class MyResponse(BaseModel):
grade: int = Field(title='等级', description='返回等级')
score: int = Field(title='分数', description='返回分数')
# 3. 定义具体 api 文档信息
@app.post('/demo',
summary='示例接口',
description='我是对示例接口的描述',
response_description='我是对接口返回值的描述',
response_model=MyResponse)
async def demo(request : MyRequest):
return {'grade': 1001, 'score': 85}
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=8000)
7. 应用案例
本案例使用 FastAPI 框架构建了一个简单的 员工管理系统，主要提供 添加、查询、修改、删除 员工信息的基本功能。数据存储使用 SQLite，并借助 SQLModel 进行 ORM 操作。整个系统采用 RESTful API 设计，前端或其他客户端可通过 HTTP 请求与后端交互。此外，案例中包含了一个 requests 客户端，用于演示 API 的调用方式。
本案例会用到 sqlite 数据库，请安装下面的模块：
pip install sqlmodel
7.1 接口开发
系统共设计了四个核心接口：
- 添加员工信息 (
POST /add)：接受员工信息，添加到数据库中并返回新增的员工数据。 - 查询员工信息 (
GET /lst)：返回所有员工的详细信息。 - 修改员工信息 (
PUT /upd)：根据id进行员工信息更新，支持部分字段修改。 - 删除员工信息 (
DELETE /del?emp_id={id})：根据id删除对应的员工记录，并返回删除状态。
接口通过 依赖注入（Depends） 方式获取数据库会话，保证数据操作的一致性，同时返回的数据均为 JSON 格式，便于客户端解析。
import uvicorn
from fastapi import FastAPI, Depends
from sqlmodel import SQLModel, Field, create_engine, Session, select
from pydantic import BaseModel
import os
from typing import List
# 定义表结构
class Employee(SQLModel, table=True):
id : int | None = Field(default=None, primary_key=True)
name : str = Field(default='undefined')
age : int = Field(default=0)
salary : float = Field(default=0.0)
# 创建数据库
db_name = 'employee.db'
if os.path.exists(db_name):
os.remove(db_name)
engine = create_engine(f'sqlite:///{db_name}')
SQLModel.metadata.create_all(engine)
# 创建 fastapi 应用
app = FastAPI()
# 创建依赖注入函数
def get_db_session():
with Session(engine) as session:
yield session
# 定义服务接口
# 1. 添加员工信息
@app.post('/add')
def add_emp(employee : Employee, session : Session = Depends(get_db_session)) -> Employee:
# 数据入库
session.add(employee)
# 提交事务
session.commit()
# 更新数据
session.refresh(employee)
return employee
# 2. 展示员工信息
@app.get('/lst')
def lst_emp(session : Session = Depends(get_db_session)) -> List[Employee]:
# 使用 select 构建一个 select 查询，并交给 session 去执行，通过 all 筛选出所有的数据，并返回
# 注意: 执行结果是一个 List[Dict] 类型的数据
employees = session.exec(select(Employee)).all()
return employees
# 3. 删除员工信息
class DeleteResponse(BaseModel):
status : str
employee : Employee | None
@app.delete('/del')
def del_emp(emp_id : int, session : Session = Depends(get_db_session)) -> DeleteResponse:
# 根据主键查询要删除的数据
employee = session.get(Employee, emp_id)
if not employee:
return DeleteResponse(status='failure', employee=None)
# 直接从数据库中删除指定数据
session.delete(employee)
session.commit()
return DeleteResponse(status='success', employee=employee)
# 4. 修改员工信息
@app.put('/upd')
def upd_emp(employee : Employee, session : Session = Depends(get_db_session)) -> Employee:
# 查询要修改的信息是否存在
employee_db : Employee = session.get(Employee, employee.id)
# 将输入的 employee 对象转换为 dict 类型，并且将未设置字段删除
employee_dict : dict = employee.model_dump(exclude_unset=True)
# 更新 employee_db 对象
employee_db.sqlmodel_update(employee_dict)
# 将新的数据更新到数据库中
session.add(employee_db)
session.commit()
session.refresh(employee_db)
return employee_db
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=8000)
7.2 接口使用
案例中提供了 demo() 函数，使用 requests 库调用各个 API，完成以下操作：
- 添加员工信息：调用 /add 接口，批量添加三名员工，并打印返回的员工信息。
- 查询员工信息：调用
/lst 接口，获取所有员工的详细信息并打印。 - 修改员工信息：调用 /upd 接口，修改某个员工的
name、age和salary，并查看更新后的数据。 - 删除员工信息：调用 /del 接口，删除某个员工，并返回删除状态和被删除的员工数据
pip install requests
import requests
def demo():
# 1. 添加员工信息
employees = [{'id': None, 'name': 'obama', 'age': 70, 'salary': 5000.00},
{'id': None, 'name': 'smith', 'age': 40, 'salary': 3000.00},
{'id': None, 'name': 'polly', 'age': 90, 'salary': 7000.00}]
for employee in employees:
response = requests.post('http://127.0.0.1:8000/add', json=employee)
print('添加的信息:', response.json())
print('-' * 65)
# 2. 列出员工信息
response = requests.get('http://127.0.0.1:8000/lst')
for employee in response.json():
print('查询的数据:', employee)
# 3. 修改员工信息
update_employee = {'id': 2, 'name': 'new_name', 'age': 55, 'salary': 1000.00}
response = requests.put('http://127.0.0.1:8000/upd', json=update_employee)
print('修改的数据:', response.json())
# 4. 删除员工信息
params = {'emp_id': 1}
response = requests.delete('http://127.0.0.1:8000/del', params=params)
print('删除的数据:', response.json())
if __name__ == '__main__':
demo()
冀公网安备13050302001966号
讲的真好