FastAPI - 现代 Web 框架

安装

pip install fastapi uvicorn

快速开始

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return {"message": "Hello!"}

# 启动: uvicorn main:app --reload

路由参数

@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"item_id": item_id, "name": "item"}

# 查询参数
@app.get("/search")
def search(q: str, limit: int = 10):
    return {"q": q, "limit": limit}

请求体 & 响应模型

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    is_available: bool = True

@app.post("/items/")
def create_item(item: Item):
    return item

@app.get("/items/{item_id}", response_model=Item)
def get_item(item_id: int):
    return Item(name="Test", price=10.0)

依赖注入

from fastapi import Depends

def get_db():
    db = connect_db()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/")
def get_users(db = Depends(get_db)):
    return db.query(User).all()

查询参数

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

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

# 可选查询参数
@app.get("/search/")
def search(q: Optional[str] = None):
    return {"query": q}

请求体

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: Optional[float] = None

@app.post("/items/")
def create_item(item: Item):
    return {"item": item, "total_price": item.price + (item.tax or 0)}

响应模型

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str

class UserOut(BaseModel):
    username: str
    email: str

@app.post("/users/", response_model=UserOut)
def create_user(user: UserIn):
    # 创建用户逻辑
    return {"username": user.username, "email": f"{user.username}@example.com"}

表单数据

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

文件上传

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents)
    }

Cookie 和 Header

from fastapi import FastAPI, Cookie, Header

app = FastAPI()

@app.get("/items/")
def read_items(
    ads_id: str = Cookie(None),
    user_agent: str = Header(None)
):
    return {"ads_id": ads_id, "User-Agent": user_agent}

响应状态码

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(name: str):
    return {"name": name}

@app.get("/items/{item_id}", status_code=status.HTTP_200_OK)
def read_item(item_id: int):
    return {"item_id": item_id}

异常处理

from fastapi import FastAPI, HTTPException

app = FastAPI()

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

@app.get("/items/{item_id}")
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]}