用FastAPI实现简单的微服务API网关

“ 本次实现的API网关非常简单，但是我觉得这样的架构已经能解决80%以上的软件架构问题，尤其是做小型系统时，显然这种架构的综合成本会很低。”

本文阐述了基于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)

按理说在访问量大的时候，后面的异步转发性能应该更好一些。不过在我的开发机简单测试时，发现在访问量不大的情况下，二者的性能差别并不明显。

测试网关

启动翻译服务

在我的开发环境中，可以用命令行启动：

# 激活venv

.venv\Scripts\activate



# 启动翻译服务

python "06-2.langchain api with fastapi.py"

翻译服务的代码可以去本文末尾的gitee或者github中下载。

启动网关

# 激活venv

.venv\Scripts\activate



# 启动网关服务

python "07-1.fastapi gateway.py"

测试网关  

使用`Apifox`发送请求。

下载源代码

-[gitee](https://gitee.com/liupras/langchain-llama3-Chroma-RAG-demo)

-[github](https://github.com/liupras/langchain-llama3-Chroma-RAG-demo)

文章转自微信公众号@AI很有趣