使用FastAPI构建高性能RESTful API的完整指南_Python

引言

fastapi是一个现代、高性能的python web框架,专为构建api而设计。它基于标准的python类型提示,提供了自动的api文档生成、数据验证和序列化等强大功能。本指南将带你从基础到高级,全面掌握fastapi的使用。

为什么选择fastapi

fastapi相比其他框架有以下显著优势:

高性能: 性能可与nodejs和go媲美,是最快的python框架之一
开发效率高: 比传统框架减少约40%的开发时间
自动文档: 自动生成交互式api文档(swagger ui和redoc)
类型安全: 基于python类型提示,提供编辑器自动补全和类型检查
标准化: 基于openapi和json schema标准
异步支持: 原生支持async/await语法

环境搭建

安装依赖

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # windows: venv\scripts\activate

# 安装fastapi和asgi服务器
pip install fastapi
pip install "uvicorn[standard]"

第一个api应用

创建main.py文件:

from fastapi import fastapi

app = fastapi()

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

运行应用:

uvicorn main:app --reload

访问http://localhost:8000即可看到返回结果,访问http://localhost:8000/docs可以看到自动生成的交互式文档。

核心概念

路径参数

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

fastapi会自动进行类型转换和验证,如果item_id不是整数,会返回清晰的错误信息。

查询参数

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

查询参数在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会自动验证请求数据,并提供清晰的错误提示。

数据验证

使用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}

响应模型

定义响应结构

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):
    # 假设保存到数据库后返回
    return {"id": 1, **item.dict()}

多种响应状态码

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} 时出错"}
    )

依赖注入

基础依赖

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}

数据库会话依赖

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

认证和授权

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"}

数据库集成

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

后台任务

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"通知发送到: {email}")
    return {"message": "通知将在后台发送"}

文件上传和下载

文件上传

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"] == "测试商品"

性能优化

使用异步操作

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()

数据库查询优化

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

响应缓存

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

创建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"]

生产环境配置

# 使用gunicorn + uvicorn workers
# 命令行运行:
# gunicorn main:app -w 4 -k uvicorn.workers.uvicornworker --bind 0.0.0.0:8000

环境变量配置

from pydantic_settings import basesettings

class settings(basesettings):
    app_name: str = "fastapi应用"
    database_url: str
    secret_key: str
    
    class config:
        env_file = ".env"

settings = settings()

最佳实践

项目结构

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)

总结

fastapi是一个强大而现代的python web框架,它结合了高性能、开发效率和优秀的开发者体验。通过本指南,你已经学习了:

fastapi的核心概念和基础用法
数据验证和序列化
依赖注入系统
认证和授权
数据库集成
性能优化技巧
生产环境部署

继续深入学习fastapi的官方文档,并在实际项目中应用这些知识,你将能够构建出高性能、可维护的restful api应用。

以上就是使用fastapi构建高性能restful api的完整指南的详细内容，更多关于fastapi构建高性能api的资料请关注代码网其它相关文章！


验证码：

使用FastAPI构建高性能RESTful API的完整指南

2025年11月04日 • Python •我要评论

引言