Skip to content

FastAPI 入门:构建高性能 Python Web API

FastAPI 是一个现代、快速(高性能)的 Web 框架,用于基于标准 Python 类型提示构建 API。本文将带你深入了解 FastAPI 的核心特性和最佳实践。

为什么选择 FastAPI?

核心优势

  • 快速:与 NodeJS 和 Go 相当的高性能
  • 代码简洁:减少约 40% 的代码量
  • 自动文档:基于 OpenAPI 规范自动生成交互式文档
  • 类型安全:利用 Python 类型提示提供更好的 IDE 支持
  • 易于测试:内置测试支持

性能对比

python
# FastAPI 性能示例
from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

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

# 性能测试结果(请求/秒):
# Starlette: ~28,000
# FastAPI: ~26,000
# Flask: ~11,000

快速开始

安装

bash
pip install fastapi uvicorn

第一个 API

python
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def 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}

运行服务器

bash
uvicorn main:app --reload

访问 http://localhost:8000/docs 查看自动生成的 API 文档。

路径参数和验证

路径参数

python
from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., gt=0, description="项目 ID,必须大于 0")
):
    return {"item_id": item_id}

查询参数

python
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    skip: int = Query(0, ge=0, description="跳过的项目数"),
    limit: int = Query(10, le=100, description="返回的最大项目数")
):
    return {"skip": skip, "limit": limit}

请求体验证

python
from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    description: str = Field(None, max_length=200)
    price: float = Field(..., gt=0, description="价格必须大于 0")
    tax: float = Field(None, ge=0, le=1)

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

数据库集成

使用 SQLAlchemy

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

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(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)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

Base.metadata.create_all(bind=engine)

依赖注入

python
from fastapi import Depends
from sqlalchemy.orm import Session

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

@app.post("/users/")
async 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

最佳实践

1. 项目结构

myapp/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── endpoints.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── models/
│   ├── schemas/
│   └── services/

2. 配置管理

python
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    APP_NAME: str = "FastAPI App"
    DEBUG: bool = False
    DATABASE_URL: str
    SECRET_KEY: str

    class Config:
        env_file = ".env"

settings = Settings()

3. 错误处理

python
from fastapi import HTTPException, status

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

4. 异步操作

python
import aiohttp

@app.get("/external-api")
async def fetch_external_data():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as response:
            data = await response.json()
            return data

性能优化

使用缓存

python
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

@app.get("/expensive-operation")
@cache(expire=60)  # 缓存 60 秒
async def expensive_operation():
    # 耗时操作
    return result

数据库连接池

python
from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

总结

FastAPI 是一个强大而现代的 Python Web 框架,它结合了高性能和开发效率。通过本文的学习,你应该能够:

  • ✅ 创建基本的 FastAPI 应用
  • ✅ 实现路径参数和查询参数验证
  • ✅ 集成数据库
  • ✅ 应用最佳实践

开始构建你的第一个 FastAPI 项目吧!

相关文章

Released under MIT License.