如何使用Python创建高性能的API网关-DEV365 开发者社区

如何使用Python创建高性能的API网关

在当今世界，API已成为现代应用的重要组成部分。他们允许不同的服务互相交流和交换数据，从而可以创建复杂的系统。但是，管理API请求可能具有挑战性，尤其是在处理多个API服务器时。在本文中，我们将探索如何构建可以处理API请求并将其分配给其他服务器的网关服务器。

简介

网关服务器是一家中央服务器，可从客户端接收请求并将其分发给其他API服务器。网关服务器充当客户端和API服务器之间的中介，处理响应的身份验证，速率限制和缓存。使用网关服务器的优点是，它简化了API请求的管理并为客户提供了单个入口。

在本文中，我们将专注于使用Python和FastApi构建网关服务器。我们将使用Asyncio处理异步请求，并重新用于缓存。

构建网关服务器

要构建我们的网关服务器，我们将使用FastApi，这是一个现代的Web框架，用于使用Python构建API。我们将定义一个接收API请求并将其分发给其他API服务器的端点。我们还将实施响应的利率限制，身份验证和缓存。

我们的端点定义如下：


@app.get("/api/v1/{path:path}", include_in_schema=False)
@auth_and_rate_limit
async def v1_gateway(request: Request, path: str):
    """
    NOTE: In order for the gateway server to work properly it needs at least 2 GIG or RAM
        master router
    :param request:
    :param path:
    :return:
    """

此端点接收带有路径参数的请求，这是客户端要访问的API端点的URL。我们使用auth_and_rate_limit装饰器来强制限制速率和身份验证。

接下来，我们检查请求是否包含API密钥。如果是这样，我们将其传递给Create_take_credit_args函数，该功能处理API键的身份验证。

api_key: dict = request.query_params.get('api_key')
_path = f"/api/v1/{path}"
await create_take_credit_args(api_key=api_key, path=_path)

认证完成后，我们创建了要将请求分配给的API服务器列表。

api_urls = [f'{api_server_url}/api/v1/{path}' for api_server_url in api_server_urls]

然后，我们检查请求的响应是否在REDIS中缓存。如果是这样，我们返回缓存的响应。否则，我们将请求发送给所有API服务器并等待其响应。

tasks = [redis_cache.get(key=api_url, timeout=60*5) for api_url in api_urls]
cached_responses = await asyncio.gather(*tasks)

for i, response in enumerate(cached_responses):
    if response is not None:
        app_logger.info(msg=f"Found cached response from {api_urls[i]}")
        return JSONResponse(content=response, status_code=200, headers={"Content-Type": "application/json"})

try:

    # 5 minutes timeout on resource fetching from backend - some resources may take very long
    tasks = [requester(api_url=api_url, timeout=300) for api_url in api_urls]
    responses = await asyncio.gather(*tasks)

except asyncio.CancelledError:
    responses = []
except httpx.HTTPError as http_err:
    responses = []

如果我们收到任何API服务器的响应


    app_logger.info(msg=f"Request Responses returned : {len(responses)}")
    for i, response in enumerate(responses):
        if response and response.get("status", False):
            api_url = api_urls[i]
            # NOTE, Cache is being set to a ttl of one hour here
            await redis_cache.set(key=api_url, value=response, ttl=60 * 60)
            app_logger.info(msg=f"Server Responded for this Resource {api_url}")
            return JSONResponse(content=response, status_code=200, headers={"Content-Type": "application/json"})
        else:
            # The reason for this algorithm is because sometimes the cron server is busy this way no matter
            # what happens a response is returned
            app_logger.warning(msg=f"""
            Server Failed To Respond - Or Data Not Found
                Original Request URL : {api_urls[i]}
                Actual Response : {response}          
            """)

    mess = "All API Servers failed to respond - Or there is no Data for the requested resource and parameters"
    app_logger.warning(msg=mess)
    # TODO - send Notifications to developers that the API Servers are down - or something requests coming up empty handed
    _time = datetime.datetime.now().isoformat(sep="-")

    # TODO - create Dev Message Types - Like Fatal Errors, and etc also create Priority Levels
    _args = dict(message_type="resource_not_found", request=request, api_key=api_key)
    await email_process.send_message_to_devs(**_args)
    return JSONResponse(content={"status": False, "message": mess}, status_code=404,
                        headers={"Content-Type": "application/json"})

如果未从任何后端服务器收到任何响应，则该函数将记录警告消息，表明所有API服务器都无法响应，或者没有请求资源和参数的数据。然后，它使用“ email_process.send_message_to_devs”功能向开发人员发送通知，并返回具有404个状态代码的JSON响应和“ application/json”的“ content-type”标题。该响应包含一个“状态”字段设置为false和一个带有错误消息的“消息”字段。

if response and response.get("status", False):

如果从后端服务器收到响应，并且包含一个“状态”字段设置为true，则使用“ redis_cache.set”函数缓存响应，并立即以200个状态代码和“ content-”返回键入“应用程序/JSON”的标题。否则，该函数会记录警告消息，表明服务器无法响应或找不到请求的数据。

*结论 *

总而言之，the v1 gateway API endpoint provides a robust and reliable way to retrieve data from multiple API servers.

它利用缓存来减少API服务器上的负载，并改善经常要求的资源的响应时间。

它还实现了限制速率和身份验证，以确保系统的安全性和稳定性。

如果所有API服务器都无法响应或返回任何数据，则系统将通知向开发人员发送通知并将适当的错误消息返回给客户端。

总的来说，此API端点是如何设计和实现现代Web应用程序的可扩展和弹性的API网关的一个很好的例子。

The Above Gateway was implemented as a Gateway TO EOD Stock API which can be found on here.

并为您的网站和应用提供可扩展的解决方案，如果您正在寻找：

Exchange Information
股票股票数据
一天结束（EOD）库存数据
基本数据
股票期权和拆分数据
Financial News API
股票的社交媒体趋势数据
新闻与社交媒体的情感分析