FastAPI
FastAPI 是一个用于构建 API 的现代、快速(高性能)的 Web 框架,使用 Python 并基于标准的 Python 类型提示。具有以下特性:
- 快速:可与 NodeJS 和 Go 并肩的极高性能(归功于 Starlette 和 Pydantic)
- 高效编码:提高功能开发速度约 200% 至 300%
- 更少 bug:减少约 40% 的人为(开发者)导致错误
- 智能:极佳的编辑器支持,处处皆可自动补全,减少调试时间
- 简单:设计的易于使用和学习,阅读文档的时间更短
- 自动文档:自动生成交互式 API 文档
本指南介绍如何在 CloudBase HTTP 云函数上部署 FastAPI 应用程序。
前置条件
在开始之前,请确保您已经:
- 安装了 Python 3.10 或更高版本
- 拥有腾讯云账号并开通了 CloudBase 服务
- 了解基本的 Python 和 FastAPI 开发知识
第一步:创建 FastAPI 项目
💡 提示:如果您已经有一个 FastAPI 项目,可以跳过此步骤。
创建项目目录
mkdir fastapi-cloudbase
cd fastapi-cloudbase
创建应用文件
创建 app.py 文件,这是应用的入口文件:
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
app = FastAPI(
title="FastAPI CloudBase Demo",
description="A FastAPI application running on CloudBase HTTP Functions",
version="1.0.0"
)
# 数据模型
class User(BaseModel):
id: int
name: str
email: str
age: Optional[int] = None
class UserCreate(BaseModel):
name: str
email: str
age: Optional[int] = None
# 模拟数据库
users_db = [
User(id=1, name="张三", email="zhangsan@example.com", age=25),
User(id=2, name="李四", email="lisi@example.com", age=30),
User(id=3, name="王五", email="wangwu@example.com", age=28)
]
@app.get("/")
async def root():
"""根路径处理函数"""
return {
"message": "Hello from FastAPI on CloudBase!",
"framework": "FastAPI",
"docs": "/docs",
"redoc": "/redoc"
}
@app.get("/health")
async def health_check():
"""健康检查接口"""
return {
"status": "healthy",
"framework": "FastAPI",
"version": "1.0.0"
}
@app.get("/api/users", response_model=List[User])
async def get_users(
page: int = Query(1, ge=1, description="页码"),
limit: int = Query(10, ge=1, le=100, description="每页数量")
):
"""获取用户列表(支持分页)"""
start_index = (page - 1) * limit
end_index = start_index + limit
paginated_users = users_db[start_index:end_index]
return paginated_users
@app.get("/api/users/{user_id}", response_model=User)
async def get_user(user_id: int):
"""根据 ID 获取用户"""
user = next((user for user in users_db if user.id == user_id), None)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.post("/api/users", response_model=User, status_code=201)
async def create_user(user: UserCreate):
"""创建新用户"""
# 检查邮箱是否已存在
if any(u.email == user.email for u in users_db):
raise HTTPException(status_code=400, detail="Email already registered")
# 生成新 ID
new_id = max(u.id for u in users_db) + 1 if users_db else 1
# 创建新用户
new_user = User(id=new_id, **user.dict())
users_db.append(new_user)
return new_user
@app.put("/api/users/{user_id}", response_model=User)
async def update_user(user_id: int, user_update: UserCreate):
"""更新用户信息"""
user_index = next((i for i, u in enumerate(users_db) if u.id == user_id), None)
if user_index is None:
raise HTTPException(status_code=404, detail="User not found")
# 检查邮箱是否被其他用户使用
if any(u.email == user_update.email and u.id != user_id for u in users_db):
raise HTTPException(status_code=400, detail="Email already registered")
# 更新用户
updated_user = User(id=user_id, **user_update.dict())
users_db[user_index] = updated_user
return updated_user
@app.delete("/api/users/{user_id}")
async def delete_user(user_id: int):
"""删除用户"""
user_index = next((i for i, u in enumerate(users_db) if u.id == user_id), None)
if user_index is None:
raise HTTPException(status_code=404, detail="User not found")
deleted_user = users_db.pop(user_index)
return {"message": f"User {deleted_user.name} deleted successfully"}
# 自定义异常处理
@app.exception_handler(404)
async def not_found_handler(request, exc):
return {"error": "Not Found", "message": "The requested resource was not found"}
@app.exception_handler(500)
async def internal_error_handler(request, exc):
return {"error": "Internal Server Error", "message": "Something went wrong"}
if __name__ == "__main__":
# CloudBase HTTP 云函数要求监听 9000 端口
uvicorn.run(app, host="0.0.0.0", port=9000)
创建依赖文件
创建 requirements.txt 文件:
fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
💡 说明:
fastapi:FastAPI 框架uvicorn:ASGI 服务器,用于运行 FastAPI 应用pydantic:数据验证和序列化库
第二步:本地测试
⚠️ 重要提示:CloudBase HTTP 云函数要求应用监听 9000 端口。
设置虚拟环境
# 创建虚拟环境
python -m venv venv
# 激活虚拟环境
# Linux/macOS
source venv/bin/activate
# Windows
# venv\Scripts\activate
安装依赖
pip install -r requirements.txt
启动应用
# 方式一:直接运行
python app.py
# 方式二:使用 uvicorn 命令
uvicorn app:app --host 0.0.0.0 --port 9000 --reload
测试 API
应用启动后,您可以通过以下方式测试:
访问文档:
- Swagger UI:
http://localhost:9000/docs - ReDoc:
http://localhost:9000/redoc
测试 API 接口:
# 测试根路径
curl http://localhost:9000/
# 测试健康检查
curl http://localhost:9000/health
# 测试用户列表
curl http://localhost:9000/api/users
# 测试分页
curl "http://localhost:9000/api/users?page=1&limit=2"
# 测试获取单个用户
curl http://localhost:9000/api/users/1
# 测试创建用户
curl -X POST http://localhost:9000/api/users \
-H "Content-Type: application/json" \
-d '{"name":"新用户","email":"newuser@example.com","age":25}'
# 测试更新用户
curl -X PUT http://localhost:9000/api/users/1 \
-H "Content-Type: application/json" \
-d '{"name":"更新用户","email":"updated@example.com","age":26}'
第三步:创建启动脚本
创建 scf_bootstrap 文件(无扩展名):
#!/bin/bash
export PYTHONPATH="./venv/lib/python3.10/site-packages:$PYTHONPATH"
/var/lang/python310/bin/python3.10 -m uvicorn app:app --host 0.0.0.0 --port 9000
为启动脚本添加执行权限:
chmod +x scf_bootstrap
💡 说明:
scf_bootstrap是 CloudBase 云函数的启动脚本- 使用 uvicorn 启动 FastAPI 应用
- 确保应用监听 9000 端口
第四步:准备部署文件
确保您的项目目录结构如下:
fastapi-cloudbase/
├── venv/ # 虚拟环境(可选)
├── app.py # 应用主文件
├── requirements.txt # 依赖列表
└── scf_bootstrap # 启动脚本
第五步:部署到 CloudBase HTTP 云函数
通过控制台部署
- 登录 CloudBase 控制台
- 选择您的环境,进入「云函数」页面
- 点击「新建云函数」
- 选择「HTTP 云函数」
- 填写函数名称(如:
fastapi-app) - 选择运行时:Python 3.10
- 提交方法选择:本地上传文件夹
- 函数代码选择项目根目录进行上传
- 自动安装依赖:开启此选项
- 点击「创建」按钮等待部署完成
通过 CLI 部署(敬请期待)
打包部署
如果需要手动打包:
# 创建部署包(排除虚拟环境)
zip -r fastapi-app.zip . -x "venv/*" ".git/*" "__pycache__/*" "*.pyc"
第六步:访问您的应用
部署成功后,您可以参考通过 HTTP 访问云函数设置自定义域名访问HTTP 云函数。
您可以访问以下接口:
- 根路径:
/- 欢迎信息 - API 文档:
/docs- Swagger UI 文档 - 替代文档:
/redoc- ReDoc 文档 - 健康检查:
/health- 应用状态 - 用户 API:
/api/users- RESTful 用户接口
常见问题
Q: 为什么必须使用 9000 端口?
A: CloudBase HTTP 云函数要求应用监听 9000 端口,这是平台的标准配置。
Q: FastAPI 的自动文档在云函数中可以正常使用吗?
A: 是的,FastAPI 的 Swagger UI 和 ReDoc 文档在云函数中可以正常访问和使用。
Q: 如何处理静态文件?
A: FastAPI 可以通过 StaticFiles 中间件处理静态文件,或者使用 CDN 服务。
Q: 如何查看应用日志?
A: 在 CloudBase 控制台的云函数页面,点击函数名称进入详情页查看运行日志。
Q: 支持哪些 Python 版本?
A: CloudBase 支持 Python 3.6、3.7、3.9、3.10 等版本,建议使用 Python 3.10。
最佳实践
1. 环境变量管理
import os
from fastapi import FastAPI
# 使用环境变量
DEBUG = os.getenv("DEBUG", "False").lower() == "true"
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")
app = FastAPI(debug=DEBUG)
2. 数据库集成
# 使用 SQLAlchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
3. 中间件配置
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
# CORS 配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Gzip 压缩
app.add_middleware(GZipMiddleware, minimum_size=1000)
4. 日志配置
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
5. 依赖注入
from fastapi import Depends
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/api/users")
async def get_users(db: Session = Depends(get_db)):
return db.query(User).all()
进阶功能
1. 身份验证
pip install python-jose[cryptography] passlib[bcrypt] python-multipart
2. 后台任务
from fastapi import BackgroundTasks
@app.post("/send-email/")
async def send_email(background_tasks: BackgroundTasks):
background_tasks.add_task(send_email_task, "user@example.com")
return {"message": "Email sent in background"}
3. WebSocket 支持
from fastapi import WebSocket
@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: {data}")
下一步
- 了解更多 HTTP 云函数配置选项
- 学习如何 连接 CloudBase 数据库