使用FastAPI構(gòu)建高性能RESTful API的完整指南

更新時間：2025年11月04日 09:24:23 作者：天天進步2015

FastAPI是一個現(xiàn)代、高性能的Python Web框架,專為構(gòu)建API而設(shè)計,本指南將帶你從基礎(chǔ)到高級,全面掌握FastAPI的使用,感興趣的小伙伴跟隨小編一起學(xué)習(xí)一下吧

引言

FastAPI是一個現(xiàn)代、高性能的Python Web框架,專為構(gòu)建API而設(shè)計。它基于標準的Python類型提示,提供了自動的API文檔生成、數(shù)據(jù)驗證和序列化等強大功能。本指南將帶你從基礎(chǔ)到高級,全面掌握FastAPI的使用。

為什么選擇FastAPI

FastAPI相比其他框架有以下顯著優(yōu)勢:

高性能: 性能可與NodeJS和Go媲美,是最快的Python框架之一
開發(fā)效率高: 比傳統(tǒng)框架減少約40%的開發(fā)時間
自動文檔: 自動生成交互式API文檔(Swagger UI和ReDoc)
類型安全: 基于Python類型提示,提供編輯器自動補全和類型檢查
標準化: 基于OpenAPI和JSON Schema標準
異步支持: 原生支持async/await語法

環(huán)境搭建

安裝依賴

# 創(chuàng)建虛擬環(huán)境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安裝FastAPI和ASGI服務(wù)器
pip install fastapi
pip install "uvicorn[standard]"

第一個API應(yīng)用

創(chuàng)建main.py文件:

from fastapi import FastAPI

app = FastAPI()

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

運行應(yīng)用:

uvicorn main:app --reload

訪問http://localhost:8000即可看到返回結(jié)果,訪問http://localhost:8000/docs可以看到自動生成的交互式文檔。

核心概念

路徑參數(shù)

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

FastAPI會自動進行類型轉(zhuǎn)換和驗證,如果item_id不是整數(shù),會返回清晰的錯誤信息。

查詢參數(shù)

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

查詢參數(shù)在URL中以?skip=0&limit=10的形式傳遞。

請求體

使用Pydantic模型定義請求體:

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):
    return item

Pydantic會自動驗證請求數(shù)據(jù),并提供清晰的錯誤提示。

數(shù)據(jù)驗證

使用Field進行高級驗證

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0, description="價格必須大于0")
    quantity: int = Field(default=1, ge=1, le=1000)
    tags: list[str] = Field(default_factory=list)

使用Query、Path和Body

from fastapi import Query, Path, Body

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="商品ID", ge=1),
    q: str | None = Query(None, min_length=3, max_length=50),
    size: int = Query(default=10, le=100)
):
    return {"item_id": item_id, "q": q, "size": size}

響應(yīng)模型

定義響應(yīng)結(jié)構(gòu)

class ItemResponse(BaseModel):
    id: int
    name: str
    price: float
    
    class Config:
        from_attributes = True

@app.post("/items/", response_model=ItemResponse)
async def create_item(item: Item):
    # 假設(shè)保存到數(shù)據(jù)庫后返回
    return {"id": 1, **item.dict()}

多種響應(yīng)狀態(tài)碼

from fastapi import status

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

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None

錯誤處理

拋出HTTP異常

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=404,
            detail="商品未找到",
            headers={"X-Error": "Item not found"}
        )
    return items_db[item_id]

自定義異常處理器

from fastapi.responses import JSONResponse
from fastapi import Request

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"處理 {exc.name} 時出錯"}
    )

依賴注入

基礎(chǔ)依賴

from fastapi import Depends

async def common_parameters(q: str | None = 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

類作為依賴

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 100):
        self.skip = skip
        self.limit = limit

@app.get("/users/")
async def read_users(pagination: Pagination = Depends()):
    return {"skip": pagination.skip, "limit": pagination.limit}

數(shù)據(jù)庫會話依賴

from sqlalchemy.orm import Session

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

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

認證和授權(quán)

JWT認證示例

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Security
import jwt

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="無效的token")

@app.get("/protected/")
async def protected_route(payload: dict = Depends(verify_token)):
    return {"user": payload["sub"]}

OAuth2密碼流

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 驗證用戶名和密碼
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="用戶名或密碼錯誤")
    
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

數(shù)據(jù)庫集成

SQLAlchemy集成

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    name = Column(String)

Base.metadata.create_all(bind=engine)

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

中間件

添加CORS中間件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

自定義中間件

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

后臺任務(wù)

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message + "\n")

@app.post("/send-notification/")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(write_log, f"通知發(fā)送到: {email}")
    return {"message": "通知將在后臺發(fā)送"}

文件上傳和下載

文件上傳

from fastapi import File, UploadFile

@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)
    }

多文件上傳

@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile] = File(...)):
    return [{"filename": file.filename} for file in files]

測試

使用TestClient

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "測試商品", "price": 99.99}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "測試商品"

性能優(yōu)化

使用異步操作

import httpx

@app.get("/external-api/")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

數(shù)據(jù)庫查詢優(yōu)化

from sqlalchemy.orm import joinedload

@app.get("/users/{user_id}/posts/")
def get_user_posts(user_id: int, db: Session = Depends(get_db)):
    # 使用eager loading避免N+1查詢問題
    user = db.query(User).options(
        joinedload(User.posts)
    ).filter(User.id == user_id).first()
    return user.posts

響應(yīng)緩存

from functools import lru_cache

@lru_cache(maxsize=128)
def get_settings():
    return Settings()

@app.get("/settings/")
def read_settings(settings: Settings = Depends(get_settings)):
    return settings

部署

使用Docker

創(chuàng)建Dockerfile:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

生產(chǎn)環(huán)境配置

# 使用Gunicorn + Uvicorn workers
# 命令行運行:
# gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

環(huán)境變量配置

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI應(yīng)用"
    database_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"

settings = Settings()

最佳實踐

項目結(jié)構(gòu)

project/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── routers/
│ │ ├── __init__.py
│ │ └── users.py
│ ├── dependencies.py
│ └── database.py
├── tests/
├── requirements.txt
└── .env

使用APIRouter組織路由

from fastapi import APIRouter

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}}
)

@router.get("/")
async def read_users():
    return [{"username": "user1"}]

@router.get("/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

# 在main.py中
app.include_router(router)

版本控制

from fastapi import APIRouter

api_v1 = APIRouter(prefix="/api/v1")
api_v2 = APIRouter(prefix="/api/v2")

@api_v1.get("/items/")
async def get_items_v1():
    return {"version": "1.0"}

@api_v2.get("/items/")
async def get_items_v2():
    return {"version": "2.0", "data": []}

app.include_router(api_v1)
app.include_router(api_v2)

總結(jié)

FastAPI是一個強大而現(xiàn)代的Python Web框架,它結(jié)合了高性能、開發(fā)效率和優(yōu)秀的開發(fā)者體驗。通過本指南,你已經(jīng)學(xué)習(xí)了:

FastAPI的核心概念和基礎(chǔ)用法
數(shù)據(jù)驗證和序列化
依賴注入系統(tǒng)
認證和授權(quán)
數(shù)據(jù)庫集成
性能優(yōu)化技巧
生產(chǎn)環(huán)境部署

繼續(xù)深入學(xué)習(xí)FastAPI的官方文檔,并在實際項目中應(yīng)用這些知識,你將能夠構(gòu)建出高性能、可維護的RESTful API應(yīng)用。

以上就是使用FastAPI構(gòu)建高性能RESTful API的完整指南的詳細內(nèi)容，更多關(guān)于FastAPI構(gòu)建高性能API的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

FastAPI
API

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线