概述

FastAPI,GitHub,一个现代的、轻量级、快速(高性能)的Web框架,专为构建基于Python的API服务而设计。提供异步编程(基于async/await)、强类型自动数据验证(基于Pydantic)、依赖注入、类型安全等特性,使得开发高性能、可扩展的API变得简单高效。非常适合处理高并发的大模型(如GPT、Qwen、DeepSeek等)推理任务。

特点:

  • 构建REST API:适合开发需要高性能和类型安全的API服务;
  • 微服务架构:作为微服务的一部分,提供高效的接口;
  • 自动生成文档:Swagger & ReDoc;
  • 机器学习推理服务:结合大模型构建推理API;
  • 实时数据流:通过WebSocket或SSE实现实时数据推送。

对比

对比项 FastAPI Flask
性能 异步支持,超快(比Flask快3~5倍) 同步阻塞,性能相对较低
类型检查 Pydantic强类型验证,减少错误 手动解析请求数据,易出错
自动文档 内置Swagger和ReDoc 需手动集成Swagger
异步支持 天生支持async/await,高并发任务不阻塞 需要手动使用线程池或协程
学习成本 比Flask略高,但上手快 语法简单,适合初学者
适用场景 高性能API、微服务、异步应用 小型Web项目、简单API

入门

环境安装:pip install fastapi uvicorn SQLAlchemy aiomysql redis gunicorn -i https://pypi.tuna.tsinghua.edu.cn/simple
解读:

  • fastapi:核心库;
  • uvicorn:ASGI服务器,用于运行应用;
  • SQLAlchemy:ORM库,用于数据库交互;
  • aiomysql:支持异步操作的MySQL客户端;
  • redis:Redis客户端;
  • gunicorn:一个高效的WSGI HTTP服务器,用于生产环境中运行应用。

典型的FastAPI项目结构,通过组织不同的功能模块和代码,使得项目更易于维护和扩展:

├── api
│   └── users.py # 用户相关API
├── forms
│   └── user_form.py
├── utils
│   ├── depends.py # 依赖注入工具
│   └── response.py # HTTP响应工具类
├── main.py
└── requirements.txt

解读:

  • api/:存放路由文件,每个文件负责不同的业务模块;
  • forms/:存放Pydantic数据模型,用于数据验证和序列化;
  • utils/:存放通用的工具类和函数,如响应工具和依赖注入工具;
  • main.py:FastAPI应用的入口,通常在此处启动应用并包含路由;
  • requirements.txt:存放所有项目依赖包的列表。

main.py中,通过FastAPI创建应用实例,并将不同的API路由引入到应用中。

from fastapi import FastAPI, Request
from api import users

app = FastAPI()
app.include_router(users.users_router) # 引用路由

users.py中,定义与用户相关的API路由。FastAPI提供非常灵活的方式来定义路径参数、查询参数、请求体参数,并支持详细的验证:

  • 使用APIRouter()来定义路由;
  • 定义多个具体的API路由,包括路径参数、查询参数、请求体参数和依赖注入。
from typing import Optional
from fastapi import APIRouter, Depends, Query, Body, Path
from forms.user_form import UserReq
from utils.depends import default_user
from utils.response import HttpResponse


users_router = APIRouter(
	prefix="/api/v1", # 路径前缀
	tags=["users"], # 文档分组
	responses={404: {"description": "Not found"}}, # 自定义某个状态码的响应内容
)


# 路径参数及验证
@users_router.get("/users_v1/{user_id}")
async def get_users_v1(
	user_id: int = Path(
		..., 
		ge=1,
		title="The ID of the user to get"
	)
):
	return HttpResponse.success(data={"user_id": user_id})


# 查询参数及验证
@users_router.get("/users_v2/")
async def get_users_v2(
	p: str = Query(
		..., # ... 代表必需填
		min_length=1,
		max_length=20,
		regex="^@", # 以@开头
		title="Query string",
		description="Query string to search",
		alias="params", # url查询参数的别名
		deprecated=True,
	),
	q: Optional[str] = Query(None) # 不是必填,默认为None
):
	result = p + (q if q else '')
	return HttpResponse.success(data={"result": result})


# 请求体参数示例
@users_router.post(
	"/users_v3/",
	tags=["users"],
	summary="Get an user", # 简短描述
	description="Get an user detail",  # 详细描述
	response_model=UserReq, # 响应数据的模型类型
	response_model_exclude_unset=True, # 忽略默认参数
	response_model_include={"name", "mobile"}, # 包含(忽略其他的)
	response_description="The user item", # 对响应内容的描述
	deprecated=True # 已废弃
)
async def get_users_v3(
    user: UserReq = Body(...)  
):
	return HttpResponse.success(data={"user": user})


# 依赖注入示例
@users_router.get("/users_v4/")
async def get_users_v4(user: UserReq = Depends(default_user)):
    return HttpResponse.success(data={"user": user})

user_form.py中,通过Pydantic创建UserReq类,用于对请求体数据进行验证和序列化:

  • 继承自BaseModel,所有字段都会自动验证;
  • 使用Field(..., gt=0)进行校验;
  • Optional标记字段可选。
from typing import Optional
from pydantic import BaseModel, Field


class UserReq(BaseModel):
	name: str
	mobile: str
	age: int = Field(..., gt=0, description="The age must be greater than zero")
	introduction: Optional[str] = Field(
	    None, 
	    title="The introduction of the user", 
	    max_length=300, 
	    example="A very nice man"
	)

depends.py中,定义一个默认的用户对象,使用FastAPI的依赖注入系统返回一个 UserReq 对象。default_user创建默认的用户对象,返回给调用者。它被用作依赖注入,可以在路由中通过Depends()获取。

from forms.user_form import UserReq


async def default_user() -> UserReq:
	user = UserReq(
		name='default',
		mobile='123',
		age=18,
		introduction='default'
	)
	return user

response.py中定义HttpResponse工具类,简化响应生成过程:

  • 提供多种常见HTTP状态类型的响应方法:成功、未授权、参数错误和服务器错误;
  • response()根据给定参数构建标准的响应格式。
from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse, Response
from typing import Any, Dict, Optional


class HttpResponse:
	ok = 200
	unautherror = 401
	paramserror = 400
	servererror = 500
	
	@classmethod
	def response(cls, code: int, msg: str, data: Optional[Any], state: Optional[Any]) -> Response:
		result = {
			"code": code,
			"msg": msg,
			"data": data,
			"state": state
		}
		return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder(result))
	
	
	@classmethod
	def success(
		cls,
		msg: str = '操作成功',
		data: Optional[Dict] = None,
		state: Optional[Any] = None,
	) -> Response:
		return cls.response(cls.ok, msg, data, state)
	
	
	@classmethod
	def unauth_error(
		cls,
		msg: str = '登录信息已过期',
		data: Optional[Dict] = None,
		state: Optional[Any] = None,
	) -> Response:
		return cls.response(cls.unautherror, msg, data, state)
	
	
	@classmethod
	def params_error(
		cls,
		msg: str = '客户端参数错误',
		data: Optional[Dict] = None,
		state: Optional[Any] = None,
	) -> Response:
		return cls.response(cls.paramserror, msg, data, state)
	
	
	@classmethod
	def server_error(
		cls,
		msg: str = '服务器内部错误',
		data: Optional[Dict] = None,
		state: Optional[Any] = None,
	) -> Response:
		return cls.response(cls.servererror, msg, data, state)

API测试

启动服务后,访问http://127.0.0.1:5000/docs来查看自动生成的在线Swagger文档,访问http://127.0.0.1:5000/redoc查看ReDoc文档。

使用curl或浏览器文档来测试API接口:

  • 支持直接在文档中测试API,并可复制为curl命令;
  • 使用curl命令发送请求,测试路径参数验证、查询参数验证和请求体参数。
# 启动服务
uvicorn main:app --reload --port 5000
# 路径参数验证
curl -X GET 127.0.0.1:5000/api/v1/users_v1/0
# 查询参数及验证
curl -X GET 127.0.0.1:5000/api/v1/users_v2/?params=@hello
# 请求体参数示例
curl -H "Content-Type: application/json" -X POST -d '{"name":"myname", "mobile":"123", "age": 18}' 127.0.0.1:5000/api/v1/users_v3/
# 依赖注入示例
curl -X GET 127.0.0.1:5000/api/v1/users_v4/

后台任务

FastAPI提供BackgroundTasks,适用于不影响主请求的任务,如发送邮件、日志记录:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(msg: str):
	with open("log.txt", "a") as f:
		f.write(msg + "\n")

@app.get("/task")
def start_task(background_tasks: BackgroundTasks):
	background_tasks.add_task(write_log, "异步任务执行!")
	return {"message": "任务已提交"}

其他

应用

使用FastAPI的框架或平台非常多,包括:

推理服务

深度学习模型从训练阶段进入生产环境的关键桥梁,其核心价值在于将训练好的模型转化为实际可用的应用能力。

  • 训练阶段:模型在离线环境中,依赖大量计算资源进行参数优化,关注模型性能指标(如准确率、损失值)。
  • 推理阶段:模型需在生产环境中实时响应用户请求,关注低延迟、高吞吐量、资源利用率,且需适应动态变化的输入数据。

选择FastAPI构建推理服务的核心原因:

  1. 极致性能:异步 + 高并发
    • 异步非阻塞 IO:FastAPI基于Starlette和Pydantic,原生支持异步编程,能高效处理大量并发请求,避免线程阻塞。比如:推理服务需同时响应多个用户请求(如智能客服),异步特性可显著降低延迟;
    • 性能对比:在基准测试中,FastAPI 的性能接近(甚至超过)Node.js 和 Go,远超传统同步框架,如Flask、Django。
  2. 开发效率:类型安全 + 自动文档
    • 类型安全:FastAPI强制使用Python类型注解(Type Hints),减少因参数错误导致的运行时崩溃;
    • 自动生成交互式文档:包括Swagger和ReDoc文档,便于前后端联调和测试。
  3. 生态兼容:深度学习框架无缝集成
    • 与PyTorch/TensorFlow无缝对接:可直接调用PyTorch、TensorFlow等深度学习模型,适合推理场景。比如:使用Transformers加载模型,通过FastAPI提供推理接口;
    • 支持GPU加速:可与CUDA无缝协作,充分利用GPU提升推理速度。

将本地部署模型封装为REST API,提供推理服务:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import pipeline
# 初始化FastAPI应用
app = FastAPI(title="大模型推理服务", description="基于FastAPI和HuggingFace Transformers")
# 使用transformers加载预训练模型(例如文本生成模型),并将其封装为可复用的类或函数
model = pipeline("text-generation", model="DeepSeek-R1")
# 定义请求体模型
class TextInput(BaseModel):
    prompt: str
    max_length: int = 50
    num_return_sequences: int = 1
# 定义推理接口
@app.post("/predict")
async def predict(input: TextInput):
    try:
        # 调用模型进行推理
        result = model(input.prompt, max_length=input.max_length, num_return_sequences=input.num_return_sequences)
        return {"output": result[0]['generated_text']}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

FastAPI通过封装预训练大模型为可复用类,结合异步接口处理用户请求,利用 uvicorn启动高性能 ASGI 服务,实现快速构建大模型推理服务:uvicorn my_app:app --host 0.0.0.0 --port 8000 --reload--reload启用自动重载(开发环境使用)。

uvicorn和gunicorn

上面简单介绍过:

  • uvicorn:ASGI服务器;
  • gunicorn:WSGI服务器。

Uvicorn,一个轻量级的 ASGI (Asynchronous Server Gateway Interface) 服务器。特点:

  • 专为Python异步框架设计(如FastAPI,Starlette,Quart等);
  • 基于uvloop和httptools构建,性能优异;
  • 开发环境快速启动;
  • 轻量级API服务;
  • 原生支持HTTP/1.1和WebSockets;
  • 支持HTTP/2(需要额外配置)

Gunicorn,一个成熟的 WSGI (Web Server Gateway Interface) 服务器,特点:

  • 传统的Web服务器;
  • 支持多种worker类型(同步和异步);
  • 提供进程管理、负载均衡等功能;
  • 稳定性高,适合生产环境
特性 Uvicorn Gunicorn
协议 ASGI WSGI,可通过worker支持ASGI
性能 更高,尤其异步场景 良好,取决于worker类型
并发模型 原生异步 多进程+线程,可配异步worker
适用框架 FastAPI,Starlette等异步框架 Django,Flask等传统框架
配置 较简单 较复杂(更多可调参数)
生产环境成熟度 较新但快速增长 非常成熟稳定
HTTP2 支持 不支持

Uvicorn + Gunicorn组合

  • 生产环境的最佳实践
  • Gunicorn作为进程管理器,Uvicorn作为worker
Logo

开源鸿蒙跨平台开发社区汇聚开发者与厂商,共建“一次开发,多端部署”的开源生态,致力于降低跨端开发门槛,推动万物智联创新。

更多推荐