0%

FastAPI基础应用

FastAPI基础应用

1. FastAPI简介

FastAPI是一个现代化、高性能的Web框架,用于构建API,基于Python 3.6+的类型提示和异步支持。它具有以下特点:

  • 快速:基于Starlette和Pydantic,性能可与NodeJS和Go媲美
  • 自动API文档:自动生成交互式API文档(Swagger UI和ReDoc)
  • 类型安全:利用Python类型提示进行数据验证和自动生成文档
  • 异步支持:原生支持异步操作,提高并发处理能力
  • 简单易用:API设计简洁,学习曲线平缓
  • 社区活跃:持续更新和完善,拥有丰富的生态系统

2. 环境搭建与依赖安装

2.1 安装Python

FastAPI需要Python 3.6+,建议使用Python 3.8或更高版本。可以从Python官网下载并安装。

2.2 安装FastAPI和依赖

使用pip安装FastAPI及其依赖:

1
pip install fastapi uvicorn
  • fastapi:FastAPI框架本身
  • uvicorn:ASGI服务器,用于运行FastAPI应用

3. 第一个FastAPI应用

3.1 创建基础应用

创建一个名为main.py的文件,编写以下代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from fastapi import FastAPI

# 创建FastAPI实例
app = FastAPI()

# 定义根路径路由
@app.get("/")
def read_root():
return {"message": "Hello, FastAPI!"}

# 定义带路径参数的路由
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}

3.2 运行应用

在终端中执行以下命令运行应用:

1
uvicorn main:app --reload
  • main:Python文件名(不含.py扩展名)
  • app:FastAPI实例名称
  • --reload:启用自动重载,代码修改后自动重启服务器

运行成功后,在浏览器中访问:

4. 路由与HTTP方法

FastAPI支持所有常见的HTTP方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from fastapi import FastAPI

app = FastAPI()

# GET请求
@app.get("/items")
def get_items():
return {"method": "GET", "action": "获取所有物品"}

# POST请求
@app.post("/items")
def create_item():
return {"method": "POST", "action": "创建物品"}

# PUT请求
@app.put("/items/{item_id}")
def update_item(item_id: int):
return {"method": "PUT", "action": "更新物品", "item_id": item_id}

# DELETE请求
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
return {"method": "DELETE", "action": "删除物品", "item_id": item_id}

5. 请求处理

5.1 路径参数

路径参数是URL路径的一部分,用于标识特定资源:

1
2
3
4
5
6
7
@app.get("/users/{user_id}/items/{item_id}")
def get_user_item(user_id: int, item_id: int, q: str = None):
return {
"user_id": user_id,
"item_id": item_id,
"q": q
}

5.2 查询参数

查询参数是URL中?后面的键值对,用于过滤、排序等操作:

1
2
3
4
@app.get("/items/")
def get_items(skip: int = 0, limit: int = 10):
# skip和limit是查询参数,有默认值
return {"skip": skip, "limit": limit}

5.3 请求体

使用Pydantic模型定义请求体:

1
2
3
4
5
6
7
8
9
10
11
12
from pydantic import BaseModel

# 定义Pydantic模型
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None

@app.post("/items/")
def create_item(item: Item):
return item

5.4 表单数据

安装依赖:

1
pip install python-multipart

使用Form类处理表单数据:

1
2
3
4
5
from fastapi import Form

@app.post("/login/")
def login(username: str = Form(...), password: str = Form(...)):
return {"username": username, "password": "******"}

5.5 文件上传

使用UploadFile类处理文件上传:

1
2
3
4
5
6
7
8
from fastapi import UploadFile, File

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
return {
"filename": file.filename,
"content_type": file.content_type
}

6. 响应处理

6.1 返回JSON响应

FastAPI默认将Python字典转换为JSON响应:

1
2
3
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"item_id": item_id, "name": "Item Name"}

6.2 自定义响应状态码

使用status参数指定响应状态码:

1
2
3
4
5
from fastapi import status

@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
return item

6.3 自定义响应头

使用Response类自定义响应头:

1
2
3
4
5
6
from fastapi import Response

@app.get("/items/{item_id}")
def read_item(item_id: int, response: Response):
response.headers["X-Item-ID"] = str(item_id)
return {"item_id": item_id}

6.4 直接返回响应

使用JSONResponseHTMLResponse等直接返回响应:

1
2
3
4
5
6
7
8
9
from fastapi.responses import JSONResponse, HTMLResponse

@app.get("/json/")
def get_json():
return JSONResponse(content={"message": "Hello"}, status_code=200)

@app.get("/html/")
def get_html():
return HTMLResponse(content="<h1>Hello, FastAPI!</h1>")

7. Pydantic模型

Pydantic是FastAPI的核心依赖之一,用于数据验证和设置管理。

7.1 基础模型定义

1
2
3
4
5
6
7
8
9
from pydantic import BaseModel, Field

class User(BaseModel):
username: str
email: str
full_name: str = None
disabled: bool = False
# 使用Field添加额外验证和描述
age: int = Field(..., ge=0, le=120, description="用户年龄")

7.2 嵌套模型

1
2
3
4
5
6
class Address(BaseModel):
city: str
country: str

class UserWithAddress(User):
address: Address

7.3 模型继承

1
2
3
4
5
6
7
8
9
10
11
12
class BaseItem(BaseModel):
name: str
description: str = None

class ItemIn(BaseItem):
price: float
tax: float = None

class ItemOut(BaseItem):
id: int
price: float
tax: float = None

8. 中间件

中间件是在请求和响应之间执行的代码,可以用于日志记录、身份验证等。

8.1 创建自定义中间件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 添加CORS中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 允许所有来源,生产环境应指定具体域名
allow_credentials=True,
allow_methods=["*"], # 允许所有HTTP方法
allow_headers=["*"], # 允许所有请求头
)

# 自定义中间件
@app.middleware("http")
async def add_process_time_header(request, call_next):
import time
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

9. 异常处理

9.1 抛出HTTP异常

1
2
3
4
5
6
7
8
9
10
11
from fastapi import HTTPException

@app.get("/items/{item_id}")
def read_item(item_id: int):
if item_id not in items:
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "Item not found"}
)
return {"item_id": item_id}

9.2 自定义异常处理器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from fastapi import Request
from fastapi.responses import JSONResponse

# 自定义异常类
class CustomException(Exception):
def __init__(self, name: str):
self.name = name

# 注册异常处理器
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(
status_code=418,
content={"message": f"I am a teapot, {exc.name}"}
)

@app.get("/teapot")
def get_teapot():
raise CustomException(name="FastAPI")

10. API文档

FastAPI自动生成两种API文档:

10.1 Swagger UI

访问 http://127.0.0.1:8000/docs,可以看到交互式API文档,支持直接测试API。

10.2 ReDoc

访问 http://127.0.0.1:8000/redoc,可以看到更简洁的API文档。

10.3 自定义文档

可以自定义文档的标题和描述:

1
2
3
4
5
app = FastAPI(
title="My FastAPI Application",
description="This is a simple FastAPI application",
version="0.1.0",
)

11. 部署FastAPI应用

11.1 使用Uvicorn部署

生产环境部署命令:

1
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
  • --workers:指定工作进程数,建议为CPU核心数

11.2 使用Gunicorn + Uvicorn

安装依赖:

1
pip install gunicorn

运行命令:

1
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

11.3 容器化部署

创建Dockerfile

1
2
3
4
5
6
7
8
9
10
FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

创建requirements.txt

1
2
fastapi
uvicorn

构建和运行Docker容器:

1
2
docker build -t fastapi-app .
docker run -d -p 8000:8000 fastapi-app

12. 常见问题与解决方案

12.1 跨域问题

使用CORS中间件解决:

1
2
3
4
5
6
7
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

12.2 类型错误

确保所有函数参数都有正确的类型提示,Pydantic模型字段也有正确的类型。

12.3 依赖注入错误

确保依赖项的返回类型与预期一致,使用Depends类正确注入依赖。

13. 进阶主题

13.1 依赖注入

使用Depends类实现依赖注入:

1
2
3
4
5
6
7
8
9
10
11
12
from fastapi import Depends

def get_db():
db = "Database Connection"
try:
yield db
finally:
db = "Closed"

@app.get("/items/")
def get_items(db: str = Depends(get_db)):
return {"db": db}

13.2 安全认证

使用OAuth2PasswordBearer实现OAuth2认证:

1
2
3
4
5
6
7
8
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
def read_users_me(token: str = Depends(oauth2_scheme)):
return {"token": token}

13.3 测试FastAPI应用

使用TestClient测试FastAPI应用:

1
2
3
4
5
6
7
8
9
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, FastAPI!"}

14. 总结

FastAPI是一个现代化、高性能的Web框架,具有以下优势:

  • 快速的开发速度
  • 自动生成API文档
  • 类型安全
  • 异步支持
  • 简单易用
  • 丰富的生态系统

通过本文的介绍,你已经了解了FastAPI的基础应用,包括:

  • 环境搭建与依赖安装
  • 第一个FastAPI应用
  • 路由与HTTP方法
  • 请求处理(路径参数、查询参数、请求体、表单数据、文件上传)
  • 响应处理
  • Pydantic模型
  • 中间件
  • 异常处理
  • API文档
  • 部署方式
  • 常见问题与解决方案

FastAPI是构建现代API的理想选择,无论是小型项目还是大型应用,都能提供高效、可靠的支持。

15. 参考资料

欢迎关注我的其它发布渠道