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
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()
@app.get("/items") def get_items(): return {"method": "GET", "action": "获取所有物品"}
@app.post("/items") def create_item(): return {"method": "POST", "action": "创建物品"}
@app.put("/items/{item_id}") def update_item(item_id: int): return {"method": "PUT", "action": "更新物品", "item_id": item_id}
@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): return {"skip": skip, "limit": limit}
|
5.3 请求体
使用Pydantic模型定义请求体:
1 2 3 4 5 6 7 8 9 10 11 12
| from pydantic import BaseModel
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 直接返回响应
使用JSONResponse、HTMLResponse等直接返回响应:
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 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()
app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], 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
| 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:
构建和运行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. 参考资料