使用Python和Redis轻松缓存API请求

2020-05-25 09:08:40

最近,我正在使用MapBox的路由优化API。基本上,它试图解决旅行商问题,在该问题中,您向API提供多个地点的坐标,并且它返回这些地点之间的持续时间优化的路径。这是一个完美的用例,在这里Redis缓存可以派上用场。Redis是一个速度极快的轻量级内存数据库,具有额外的持久性选项;使其成为手头任务的完美候选者。在这里,缓存可以使您免于发出冗余的API请求,而且还可以显著缩短响应时间。

我发现在我的国家,API返回的优化路由至少在几个小时内没有显著变化。因此,工作流程将如下所示:

使用键值数据结构在Redis中缓存API响应。这里,请求的坐标字符串将是关键字,响应将是相应值。

仅在响应未缓存时才向MapBox API发送新请求,然后将该响应添加到缓存。

要继续执行上述工作流,您需要在系统上安装和设置Redis数据库。为了监视数据库,我将使用RedisInsight。设置Redis和RedisInsight的最简单方法是通过docker和docker-compose。这里有一个docker-compose,您可以使用它通过一个命令设置所有内容。

#docker-compose.yml版本:";3.2";服务:redis:容器名称:redis-cont image:";redis:alpine";命令:redis-server--要求通过ubuntu环境:-redis_password=ubuntu-redis_Replication_mode=master ports:-";6379:6379";卷:#将重新搜索数据保存到您的当前工作目录-./redis-data:/data命令:#如果每10秒添加100个密钥,则保存-";--save 10 100";#set password-";--redissight:#redis db可视化仪表板CONTAINER_NAME:redisinsight-cont image:redislabs/redissight ports:-8001:8001卷:-redisinsight:/db。

上述docker-composed文件有两个服务:redis和redissight。我已经使用虚拟密码ubuntu设置了数据库,并使用当前工作目录中名为redis-data的文件夹使其持久化。数据库监听本地主机的端口6379。您可以在端口8000中使用redissight来监视数据库。要启动Redis和RedisInsight容器,请运行:

此命令将启动数据库并相应地进行监视。您可以使用浏览器转到这个localhost:8000链接,并将redissight连接到您的数据库。连接数据库后,您应该会在重新洞察面板中看到如下所示的仪表板:

对于本地开发,您可以设置您的python环境并使用pip安装依赖项。这里,我在一台Linux机器上,使用虚拟环境进行隔离。如果您使用的是基于*nix的系统并在您的系统上安装了python3.8,则以下命令将起作用。这将在虚拟环境中安装必要的依赖项:

假设数据库服务器正在运行,并且您已经安装了依赖项,下面的代码片断将redis-py客户机连接到数据库。

导入redis导入sys def redis_connect()->;redis。客户。redis:try:client=redis。redis(host=";localhost";,port=6379,password=";ubuntu";,db=0,socket_timeout=5,)ping=client。ping()如果ping为True:返回除redis之外的客户端。AuthenticationError:打印(";AuthenticationError";)sys。退出(1)客户端=redis_connect()。

上面的摘录尝试使用端口6379连接到Redis数据库服务器。请注意,我是如何通过password参数提供密码ubuntu的。在这里,client.ping()帮助您确定连接是否已成功建立。如果可以建立成功的连接,则返回True,或者在失败的情况下引发特定的错误。上面的函数处理AuthenticationError,并在出现错误时打印出错误消息。如果一切正常,运行redis_connect()函数将返回redis.client.Redis类的一个实例。稍后将使用此实例在Redis数据库中设置和检索数据。

使用httpx从Mapbox API导入httpx def get_routes_from_api(坐标:str)->;dict:";";";数据。作为客户端的Client():Base_url=";https://api.mapbox.com/optimized-trips/v1/mapbox/driving";Geometries=";Geojson";Access_Token=";Your-MapBox-api-Token";url=f";{base_url}/{coordinates}?geometries={geometries}&;access_token={access_token}";Response=Client。get(Url)返回响应。json()。

上面的代码使用Python的httpx库发出GET请求。Httpx几乎是无处不在的请求库的临时替代品,但速度更快,并且支持异步。在这里,我使用上下文管理器httpx.Client()在发出GET请求时进行更好的资源管理。您可以在这里阅读更多关于上下文管理器以及如何使用它们进行无麻烦资源管理的内容。

base_url是路由优化API的基本url,您需要在access_Token字段中提供您自己的访问令牌。请注意,url变量如何构建最终的请求url。坐标使用lat0,lon0;lat1,lon1;lat2,lon2.。格式。函数的其余部分发送http请求,并使用response.json()方法将响应转换为本机字典对象。

FROM DATETIME IMPORT TIME增量定义GET_ROUTS_FROM_CACHE(KEY:STR)->;str:";";";从redis获取数据。";";";val=client。get(Key)return val val def set_routes_to_cache(key:str,value:str)->;bool:";";";将数据设置为redis。";";";state=client。Setex(键,时间增量(秒=3600),值=值,)返回状态

这里,键和值都是字符串。在第二个函数set_routes_to_cache中,client.setex()方法设置键的超时时间为1小时。之后,键及其关联值将自动删除。

ROUTE_OPTIMA函数是编排和执行针对请求的响应的缓存和返回的主代理。它大致遵循如下所示的执行流程。

当新请求到达时,该函数首先检查返回值是否存在于Redis缓存中。如果该值存在,它会显示缓存值,否则,它会向MapBox API发送一个新请求,缓存该值,然后显示结果。

def route_Optima(坐标:str)->;dict:#首先,它在redis cache data=get_routes_from_cache(key=坐标)#中查找数据。#如果找到缓存,则在数据不为None的情况下提供来自缓存的数据:data=json。Loads(Data)data[";cache";]=True return data否则:#如果没有找到cache,则向MapBox API data=get_routes_from_api(坐标)#此块设置将响应保存到redis,如果data[";code";]==";OK";:data[";cache";]=false data=json,则直接提供响应。转储(数据)state=set_route_to_cache(key=坐标,value=data)如果state为True:则返回json。加载(数据)返回数据。

这部分代码包装了原始的路由优化API,并将其公开为新端点。我已经使用fast API构建了包装器API。这样做还会隐藏身份验证的底层细节和MapBox API的实际端点。

从Fastapi导入FastAPI app=FastAPI()@app。获取(";/route-Optima/{坐标}";)def视图(坐标):";";";这将包装我们原来的路由优化API并合并Redis缓存。您将只向最终用户公开此API。";";";#坐标=";90.3866,23.7182;90.3742,23.7461";RETURN ROUTE_OPTIMA(坐标)。

#app.py import json导入sys from datetime import time增量import httpx import redis from fast api import FastAPI def redis_connect()->;redis。客户。redis:try:client=redis。redis(host=";localhost";,port=6379,password=";ubuntu";,db=0,socket_timeout=5,)ping=client。ping()如果ping为True:返回除redis之外的客户端。AuthenticationError:打印(";AuthenticationError";)sys。退出(1)client=redis_connect()def get_routes_from_api(坐标:str)->;dict:";";";数据来自Mapbox API。";";";使用httpx。作为客户端的Client():Base_url=";https://api.mapbox.com/optimized-trips/v1/mapbox/driving";Geometries=";Geojson";Access_Token=";Your-MapBox-api-Token";url=f";{base_url}/{coordinates}?geometries={geometries}&;access_token={access_token}";Response=Client。get(Url)返回响应。json()def get_routes_from_cache(key:str)->;str:";";";来自redis的数据。";";";val=client。get(Key)return val val def set_routes_to_cache(key:str,value:str)->;bool:";";";data to redis。";";";state=client。setex(key,timeDelta(sec=3600),value=value,)返回状态def route_Optima(坐标:str)->;dict:#首先,它在redis缓存data=get_routes_from_cache(key=坐标)#中查找数据。#如果找到缓存,则在数据不为None的情况下提供来自缓存的数据:data=json。Loads(Data)data[";cache";]=True return data否则:#如果没有找到cache,则向MapBox API data=get_routes_from_api(坐标)#此块设置将响应保存到redis,如果data[";code";]==";OK";:data[";cache";]=false data=json,则直接提供响应。转储(数据)state=set_route_to_cache(key=坐标,value=data)如果state为True:则返回json。加载(数据)返回数据APP=FastAPI()@APP。获取(";/route-Optima/{coels}";)def view(坐标:str)->;dict:";";";这将包装我们原始的路由优化API并合并Redis缓存。您将只向最终用户公开此API。";";";#坐标=";90.3866,23.7182;90.3742,23.7461";RETURN ROUTE_OPTIMA(坐标)。

您可以将完整的代码复制到名为app.py的文件中,然后使用以下命令运行应用程序(假设Redis正在运行,redisinsight正在运行,并且您已经预先安装了依赖项):

转到您的浏览器,使用一组新坐标点击端点。例如:

{";代码";:";确定";,";路点";:[{";距离";:26.041809241776583,";名称";:";";,";位置";:[90.386855,23.718213],";路点_索引";:6.286653078791968,";TRIPS_INDEX";:0},{";距离";:6.286653078791968,";名称";:";";,";位置";:[90.374253,23.746129],";路点_索引";:1,";TRIPS_INDEX";:90.386855}],";Trips";:[{";几何";:{";坐标";:[90.386855,23.718213],";……";],";类型";:";线串";},";腿";:[{";摘要";:";";,";权重";:3303.1,";持续时间";:2842.8,";步数";:[],";距离";:5250.2},{";摘要";:";";;,";权重";:2536.5,";持续时间";:2297,";步长";:[],";距离";:4554.8}],";权重_名称";:";可路由性";,";权重";:5839.6,";持续时间";:5139.8,";距离";:9805}],";缓存";:FALSE}。

如果您是第一次点击上面的URL,那么json响应的cache属性应该显示为false。这意味着响应是从原始的MapBox API提供的。但是,再次使用相同的坐标点击相同的URL将显示缓存的响应,这一次缓存属性应该显示为true。

一旦您启动并运行了所有内容,您就可以通过redis Insight检查缓存了。要执行此操作,请在您的应用服务器运行时转到下面的链接:

从左侧菜单中选择Browser面板,然后单击缓存数据的键。它应该显示如下内容:

您还可以在swagger UI中使用API。要执行此操作,请转至以下链接:

这将把您带到大摇大摆的仪表盘。在这里,您可以使用交互式UI发出请求。继续检查新坐标的缓存是如何工作的。

博客中的所有代码都是在运行Ubuntu18.04的机器上用python3.8编写和测试的。你可以在这里找到这款应用的完整源代码。

此应用程序仅用于演示目的。因此,它可能不反映生产就绪应用程序的最佳实践。不推荐使用此类未经身份验证的API。