跳到主要内容

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 云函数

通过控制台部署

  1. 登录 CloudBase 控制台
  2. 选择您的环境,进入「云函数」页面
  3. 点击「新建云函数」
  4. 选择「HTTP 云函数」
  5. 填写函数名称(如:fastapi-app
  6. 选择运行时:Python 3.10
  7. 提交方法选择:本地上传文件夹
  8. 函数代码选择项目根目录进行上传
  9. 自动安装依赖:开启此选项
  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}")

下一步