用FastAPI实现微服务API网关

用FastAPI实现微服务API网关

本文将介绍如何使用FastAPI实现一个简单的API网关。通过本文，读者将学习到如何使用同步和异步两种方式实现HTTP请求转发，并了解如何测试网关功能。

本文阐述了基于FastAPI实现一个API网关的详细步骤。这样未来可以不断的在服务端像搭积木一样添加各种服务。 我们即将实现下面的简单的微服务架构，目前它只实现了请求转发功能。

代码实现

下面使用FastAPI的同步和异步两种方式实现http请求转发。
其中services是在本网关中定义的服务名称，path是服务的路径，网关将把这个路径转发到后端服务。
例如：前端使用URL:http://127.0.0.1:8000/translation/trans/v1访问本网关时，@api_route将会把server的值填充为translation，将path的值填充为trans/v1，这样API网关调用的后端服务的API地址为：http://127.0.0.1:5001/trans/v1。

同步转发

from fastapi import FastAPI, Request,HTTPException
import requests

# 创建一个FastAPI实例
app = FastAPI()

# 定义服务
services = {
    "translation": "http://127.0.0.1:5001",
    # 可以在这里添加其它服务地址
}
'''
services的key是服务名称，客户端在请求时传入服务名称，本网关再根据服务名称找到对应的服务地址
'''

# 接收客户端请求并转发到后端服务
@app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def gateway(service: str, path: str, request: Request):
    '''
    !注意：网关并未将header转发给后端服务，这样比较简单。
    '''
    if service not in services:
        raise HTTPException(status_code=401, detail="未找到该服务")
    # 根据服务名称找到对应的服务地址
    service_url = services[service]
    body = await request.json() if request.method in ["POST", "PUT", "PATCH"] else None 
    url = f"{service_url}/{path}"
    # 同步调用，这里会阻塞
    response = requests.post(url, json = body)
    return response.json() 

# 启动网关
if __name__ == "__main__":
    import uvicorn
    # 交互式API文档地址：
    # http://127.0.0.1:8000/docs/ 
    # http://127.0.0.1:8000/redoc/
    uvicorn.run(app, host="0.0.0.0", port=8000)

异步转发

from fastapi import FastAPI, Request,HTTPException
import httpx

# 定义超时时间，单位：秒
time_out = 30

# 创建一个FastAPI实例
app = FastAPI()

# 定义服务
services = {
    "translation": "http://127.0.0.1:5001",
    # 可以在这里添加其它服务地址
}
'''
services的key是服务名称，客户端在请求时传入服务名称，本网关再根据服务名称找到对应的服务地址
'''

# 接收客户端请求并转发到后端服务
@app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def gateway(service: str, path: str, request: Request):
    '''
    !注意：网关并未将header转发给后端服务，这样比较简单。
    '''
    if service not in services:
        raise HTTPException(status_code=401, detail="未找到该服务")
    #headers = dict(request.headers)
    # 从客户端请求中获取数据
    client_request_data = await request.json()
    service_url = services[service]
    url = f"{service_url}/{path}" 
    # 使用 httpx 将请求转发到后端服务，非阻塞，不过在我的配置一般的开发机上没有发现和阻塞式调用在性能上有多少区别。
    async with httpx.AsyncClient() as client:
        '''
        !注意：httpx.AsyncClient默认的timeout为5秒，在调用基于大模型的后端服务时经常超时，所以这里设置超时时间为30秒
        '''
        response = await client.post(url=url, json=client_request_data,timeout=time_out)
        #print(response)
        return response.json()

# 启动网关
if __name__ == "__main__":
    import uvicorn
    # 交互式API文档地址：
    # http://127.0.0.1:8000/docs/ 
    # http://127.0.0.1:8000/redoc/
    uvicorn.run(app, host="0.0.0.0", port=8000)

测试网关

1. 启动翻译服务

在开发环境中，可以用以下命令启动翻译服务：

# 激活虚拟环境
.venv\Scripts\activate

# 启动翻译服务
python "06-2.langchain api with fastapi.py"

2. 启动网关

# 激活虚拟环境
.venv\Scripts\activate

# 启动网关服务
python "07-1.fastapi gateway.py"

3. 测试网关

可以使用Apifox或其他API测试工具发送请求进行测试。