1. 安装 FastAPI 和 Uvicorn
首先，使用 pip 安装 FastAPI 和 Uvicorn（一个 ASGI 服务器）：
pip install fastapi uvicorn
2. 创建 FastAPI 应用
创建一个名为 main.py 的文件，并编写以下代码：
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello, FastAPI!"}
3. 运行应用
使用 Uvicorn 运行应用：
uvicorn main:app --reload
访问 http://127.0.0.1:8000 查看结果。
4. 路径参数和查询参数
FastAPI 支持路径参数和查询参数，以下是一个示例：
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
5. 请求体
使用 Pydantic 模型定义请求体：
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
app = FastAPI()
@app.post("/items/")
async def create_item(item: Item):
return item
6. 响应模型
定义响应模型以确保返回数据的结构：
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
app = FastAPI()
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
return item
7. 错误处理
FastAPI 提供了内置的错误处理机制：
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in [1, 2, 3]:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
8. 依赖注入
使用依赖注入来管理代码的依赖关系：
from fastapi import FastAPI, Depends
app = FastAPI()
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
9. 中间件
添加中间件以处理请求和响应：
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=["*"],
)
10. 后台任务
在响应后执行后台任务：
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_notification(email: str, message: str):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
11. 安全性
使用 OAuth2 和 JWT 进行身份验证：
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def create_access_token(data: dict):
to_encode = data.copy()
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# 这里应该验证用户名和密码
access_token = create_access_token(data={"sub": form_data.username})
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise HTTPException(status_code=401, detail="Invalid authentication credentials")
except JWTError:
raise HTTPException(status_code=401, detail="Invalid authentication credentials")
return {"username": username}
12. 文件上传
FastAPI 支持文件上传功能。以下是一个上传文件的示例：
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
return {"filename": file.filename}
13. 文件下载
以下是一个文件下载的示例：
from fastapi import FastAPI
from fastapi.responses import FileResponse
app = FastAPI()
@app.get("/downloadfile/")
async def download_file():
file_path = "path/to/your/file.txt"
return FileResponse(file_path, filename="downloaded_file.txt")
14. 图片上传
图片上传与文件上传类似，但可以添加验证以确保上传的是图片：
from fastapi import FastAPI, File, UploadFile
import imghdr
app = FastAPI()
@app.post("/uploadimage/")
async def create_upload_image(file: UploadFile = File(...)):
# 检查文件是否为图片
image_type = imghdr.what(file.file)
if not image_type:
return {"error": "Uploaded file is not an image"}
return {"filename": file.filename, "image_type": image_type}
15. 图片下载
图片下载与文件下载类似，但可以设置媒体类型：
from fastapi import FastAPI
from fastapi.responses import FileResponse
app = FastAPI()
@app.get("/downloadimage/")
async def download_image():
image_path = "path/to/your/image.jpg"
return FileResponse(image_path, media_type="image/jpeg", filename="downloaded_image.jpg")
16. WebSocket
FastAPI 支持 WebSocket 通信：
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
17. 测试应用
使用以下命令运行应用：
uvicorn main:app --reload
然后可以使用工具如 curl 或 Postman 测试上传和下载功能。
18. 部署
使用 Uvicorn 部署到生产环境：
uvicorn main:app --host 0.0.0.0 --port 80
或者使用 Gunicorn 作为进程管理器：
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app
19. 性能优化
使用异步编程提高性能：
from fastapi import FastAPI
import asyncio
app = FastAPI()
async def async_operation():
await asyncio.sleep(1)
return "Done"
@app.get("/async")
async def read_async():
result = await async_operation()
return {"result": result}
20. 总结
FastAPI 提供了简单而强大的方式来处理文件上传和下载，包括图片。通过使用 UploadFile 和 FileResponse，可以轻松实现这些功能。同时，FastAPI 还支持路径参数、查询参数、请求体、响应模型、错误处理、依赖注入、中间件、后台任务、安全性、WebSocket 等高级功能，使其成为一个功能齐全的现代 Web 框架。
