FastAPI 是一个现代、快速(高性能)的 Web 框架,用于构建 API,基于 Python 3.6+ 类型提示设计。以下是 FastAPI 的快速入门指南:

1. 安装 FastAPI

首先安装 FastAPI 和一个 ASGI 服务器(如 Uvicorn):

pip install fastapi uvicorn

2. 创建第一个 FastAPI 应用

创建一个 main.py 文件:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

3. 运行应用

使用 Uvicorn 运行应用:

uvicorn main:app --reload
  • main: 你的模块名(main.py 不带 .py
  • app: 你创建的 FastAPI 实例
  • --reload: 开发时使用,代码修改后自动重载

访问 http://127.0.0.1:8000/ 查看结果。

4. 自动生成的 API 文档

FastAPI 自动为你的 API 生成交互式文档:

  • Swagger UI: http://127.0.0.1:8000/docs
  • ReDoc: http://127.0.0.1:8000/redoc

5. 请求参数

路径参数

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

查询参数

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

请求体

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

6. 响应模型

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    full_name: str | None = None

@app.get("/users/me", response_model=User)
async def read_user_me():
    return User(username="johndoe", full_name="John Doe")

@app.get("/", response_class=HTMLResponse)
async def read_root():
    return "<html><body><h1>Hello World</h1></body></html>"

7. 状态码

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item():
    return {"message": "Item created"}

8. 异常处理

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

9. 依赖注入

from fastapi import FastAPI, Depends

app = FastAPI()

def verify_token(x_token: str | None = None):
    if not x_token or x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
    return x_token

@app.get("/items/")
async def read_items(token: str = Depends(verify_token)):
    return [{"item": "Foo"}, {"item": "Bar"}]

10. 完整示例

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import Annotated

app = FastAPI()

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

class User(BaseModel):
    username: str
    full_name: str | None = None

# 模拟数据库
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

# 依赖项
def common_parameters(
    q: str | None = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}

async def verify_token(x_token: str | None = None):
    if not x_token or x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
    return x_token

async def verify_key(x_key: str | None = None):
    if not x_key or x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

# 路由
@app.get("/items/")
async def read_items(
    commons: dict = Depends(common_parameters),
    token: str = Depends(verify_token),
    key: str = Depends(verify_key)
):
    return {
        "commons": commons,
        "token": token,
        "key": key,
        "data": fake_items_db
    }

@app.post("/items/", response_model=Item, status_code=201)
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    fake_items_db.append(item_dict)
    return item_dict

@app.get("/users/me")
async def read_user_me():
    return User(username="johndoe", full_name="John Doe")

11. 测试 API

你可以使用以下工具测试你的 FastAPI 应用:

  • 浏览器直接访问
  • Swagger UI (/docs)
  • ReDoc (/redoc)
  • 命令行工具如 curlhttpie
  • Postman 或 Insomnia 等 API 测试工具

12. 生产环境部署

对于生产环境,你可以使用:

  • Uvicorn + Gunicorn (Unix 系统)
  • Uvicorn + Waitress (Windows 系统)
  • Docker 容器化部署

示例 Gunicorn 命令:

gunicorn -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000 main:app

FastAPI 提供了许多高级功能,包括 WebSocket 支持、GraphQL 支持、后台任务、中间件等。随着你对 FastAPI 的熟悉,可以逐步探索这些功能。

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐