当前位置：首页 > article >正文

使用 FastAPI 和 Async Python 打造高性能 API

article 2025/1/11 16:04:09

随着现代互联网应用的快速发展，构建高性能的 API 成为开发者的重要任务之一。在这篇文章中，我们将探讨如何利用 FastAPI 和 Python 的异步特性打造高效的 API 服务。FastAPI 是一种非常流行的 Python 框架，结合了易用性和强大的性能支持，非常适合构建需要并发处理的应用。

为什么选择 FastAPI？

快速开发：FastAPI 提供了简单、直观的 API 开发接口，并自动生成 OpenAPI 文档。
异步支持：基于 Python 的 asyncio，充分利用协程，提升性能。
高性能：由于 FastAPI 使用 Starlette 作为其 ASGI 框架，并采用 Pydantic 进行数据验证，其速度接近 Node.js 和 Go。
类型安全：Pydantic 提供基于类型提示的强大数据验证功能，帮助开发者更早发现错误。
丰富的功能：内置请求验证、数据验证、依赖注入等。
社区活跃：FastAPI 社区庞大，拥有丰富的资源和支持。

环境准备

首先，我们需要安装 FastAPI 和一个支持 ASGI 的服务器，如 uvicorn：

pip install fastapi uvicorn

另外，如果你的应用需要访问数据库或调用外部 API，可以考虑安装异步支持的库，如 asyncpg 或 httpx。为了简化开发工作，可以使用 dotenv 加载环境变量：

pip install python-dotenv

快速上手示例

以下是一个简单的 FastAPI 应用：

from fastapi import FastAPI

app = FastAPI()

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

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

运行应用程序：

uvicorn main:app --reload

访问 http://127.0.0.1:8000/docs，你将看到自动生成的交互式 API 文档，这使得调试和测试变得非常简单。

构建异步数据库访问

FastAPI 与异步数据库访问库如 asyncpg、databases 等完美兼容。在下面的例子中，我们将展示如何使用 asyncpg 连接和查询 PostgreSQL 数据库。

安装依赖

pip install asyncpg

配置数据库操作

import asyncpg
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup():
    app.state.db = await asyncpg.create_pool(dsn="postgresql://user:password@localhost:5432/mydatabase")

@app.on_event("shutdown")
async def shutdown():
    await app.state.db.close()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    query = "SELECT * FROM users WHERE id = $1"
    async with app.state.db.acquire() as connection:
        user = await connection.fetchrow(query, user_id)
        if user:
            return {"id": user["id"], "name": user["name"]}
        return {"error": "User not found"}

通过这种方式，我们可以轻松管理数据库连接池，减少连接开销，并确保应用性能稳定。

集成异步外部服务调用

很多应用需要调用外部 API，例如第三方支付服务、地图服务等。我们可以使用 httpx 完成异步 HTTP 请求：

安装 httpx

pip install httpx

示例代码

import httpx
from fastapi import FastAPI

app = FastAPI()

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

httpx 支持异步操作，与 FastAPI 协程机制无缝结合，可以避免阻塞主线程，提升吞吐量。

实现简单的用户认证

通过 FastAPI 内置的依赖注入机制，我们可以轻松地实现用户认证。

示例代码

from fastapi import FastAPI, Depends, HTTPException, status

app = FastAPI()

# 模拟用户数据
dummy_users = {"alice": "password123", "bob": "securepassword"}

def authenticate_user(username: str, password: str):
    if username in dummy_users and dummy_users[username] == password:
        return {"username": username}
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid credentials"
    )

@app.get("/secure-data")
async def get_secure_data(user=Depends(authenticate_user)):
    return {"message": f"Hello, {user['username']}!"}

这种方式简单明了，适合快速实现认证逻辑。

性能优化技巧

数据库连接池：如前面示例，通过连接池管理数据库连接。
缓存机制：使用 Redis 缓存高频访问的数据。
限流：限制每个 IP 或用户的请求次数，保护服务。
监控与日志：引入工具如 Prometheus 和 Grafana 实时监控。
负载均衡：在生产环境使用负载均衡器分发流量。

使用 Redis 实现简单缓存

pip install aioredis

import aioredis
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup():
    app.state.redis = await aioredis.from_url("redis://localhost")

@app.on_event("shutdown")
async def shutdown():
    await app.state.redis.close()

@app.get("/cached-item/{item_id}")
async def read_item(item_id: int):
    cached_item = await app.state.redis.get(f"item:{item_id}")
    if cached_item:
        return {"cached": True, "data": cached_item}

    # 模拟获取数据库数据
    data = {"id": item_id, "value": "This is a fresh item"}
    await app.state.redis.set(f"item:{item_id}", str(data), ex=60)  # 缓存 60 秒
    return {"cached": False, "data": data}

添加请求限流

pip install slowapi

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, lambda r, e: {"detail": "Too many requests!"})

@app.get("/limited")
@limiter.limit("5/minute")
async def limited_endpoint(request: Request):
    return {"message": "This is a rate-limited endpoint."}

通过这些优化，你可以进一步提升服务性能和可靠性。