这是我参与2022首次更文挑战的第17天,活动详情查看:2022首次更文挑战」

在以前的文章里面,我给大家介绍了使用 Python 自带的 LRU 缓存实现带有过期时间的缓存:实现有过期时间的 LRU 缓存。也讲过倒排索引:使用倒排索引极速提高字符串搜索效率。但这些代码对初学者来说比较难,写起来可能会出错。

实际上,这些功能其实都可以使用 Redis 来实现,而且每个功能只需要 1 分钟就能做出来。全文搜索功能在搜索英文的时候,甚至可以智能识别拼写错误的问题。

要实现这些功能,只需要做两件事:

  1. 安装 Redis
  2. Python 安装第三方库:walrus

安装完成以后,我们来看看它有多简单:

带过期时间的缓存装饰器

我们想实现一个装饰器,它装饰一个函数。让我在 1 分appstore钟内多次访问函数的时候,使用缓存的数据;超过 1 分钟以后才重新执行函数的内部代码:

importtime
importdatetime
fromwalrusimportDatabase
db=Database()
cache=db.cache()
@cache.cached(timeout=60)
deftest():
print('函数真正运行起来')
now=datetime.datetime.now()
returnnow
now=test()
print('函数返回的数据是:',now)
time.sleep(10)#等待10秒,此时会使用缓存
print('函数返回的数据是:',test())
time.sleep(5)#等待5秒,此时依然使用缓存
print('函数返回的数据是:',test())
time.sleep(50)#让时间超过缓存的时间
print('函数返回的数据是:',test())

运行效果如下图所示:

有了这个,网页反爬限制请求频率易如反掌!

全文搜索

我们再来看看全文搜索功能,实现起来也很简单:

fromwalrusimportDatabase
db=Database()
search=db.Index('xxx')#这个名字随便取
poem1='Earlyinthedayitwaswhisperedthatweshouldsailinaboat,onlythouandI,andneverasoulintheworldwouldknowofthisourpilgrimagetonocountryandtonoend.'
poem2='HadItheheavens’embroideredcloths,Enwroughtwithgoldenandsilverlight'
poem3='tobeornottobe,thatisaquestion.'
search.add('docid1',poem1)#第一个参数不能重复
search.add('docid2',poem2)
search.add('docid3',poem3)
fordocinsearch.search('end'):
print(doc['content'])

运行效果如下图所示:

有了这个,网页反爬限制请求频率易如反掌!

如果你想让他兼容拼写错appstore误,那么可以把search = db.Index('xxx'redistribute)改成search = db.Index('xxx’, metaphone=True),运行效果如下图所示:

有了这个,网页反爬限制请求频率易如反掌!

不过遗憾的appointment是,这个全文搜索功能只支持英文。

频率限制

我们有时候要限制调用某个函python是什么意思数的频率,或者网站的python123平台登录某个接口要限制 IP 的访问频率。这个时候,使用walrus也可以轻松实现:

importtime
fromwalrusimportDatabase
db=Database()
rate=db.rate_limit('xxx',limit=5,per=60)#每分钟只能调用5次
for_inrange(35):
ifrate.limit('xxx'):
print('访问频率太高!')
else:
print('还没有触发访问频率限制')
time.sleep(2)

运行效果如下图所示:

有了这个,网页反爬限制请求频率易如反掌!

其中参数limit表示能出现多少次,per表示在多长时间内。\

rate.limit只要传入python代码大全相同的参数,那么就会开始检查这个参数在设定的时间内出现的频率。

你可能觉得这个例子并不能说明什么问题,那么我们跟 FastAPI 结合一下,用来限制 IP 访问接口的频率。编索引写如下代码:

fromwalrusimportDatabase,RateLimitException
fromfastapiimportFastAPI,Request
fromfastapi.responsesimportJSONResponse
db=Database()
rate=db.rate_limit('xxx',limit=5,per=60)#每分钟只能调用5次
app=FastAPI()
@app.exception_handler(RateLimitException)
defparse_rate_litmit_exception(request:Request,exc:RateLimitException):
msg={'success':False,'msg':f'请喝杯茶,休息一下,你的ip:{request.client.host}访问太快了!'}
returnJSONResponse(status_code=429,content=msg)
@app.get('/')
defindex():
return{'success':True}
@app.get('/important_api')
@rate.rate_limited(lambdarequest:request.client.host)
defquery_important_data(request:Request):
data='重要数据'
return{'success':True,'data':data}

上面python代码大全代码定义了一个全局的异常拦截器:

@app.exception_handler(RateLimitException)
defparse_rate_litmit_exception(request:Request,exc:RateLimitException):
msg={'success':False,'msg':f'请喝杯茶,休息一下,你的ip:{request.client.host}访问太快了!'}
returnJSONResponse(status_code=429,content=msg)

在整个代码的任何地方抛出了RateLimitExcep数据库管理系统tion异常,就会进入这里的逻辑中。

使用装饰器@ratpython代码画樱花e.rate_limited装饰一个路由函数,并索引超出矩阵维度且这个装饰器要更靠近函数。路由函数数据库接收什么redis数据库参数,它就接收什么参数。在上面的例子中,我们只接收了requpython是什么意思est参数,用于获取访问者的 IP。

发现这个 IP 的访问频率超过了限制,就抛出一个RateLimitException。于是前面定义好的全局拦截器就会拦截RateLimitE数据库xception异常,拦截到以后返回我们定义好的报错信息。

在频率范围内访问页面,返回正常的 JSON 数据:

有了这个,网页反爬限制请求频率易如反掌!

频率超过设定的值以后,访问页面就会报错,如下图所示:

有了这个,网页反爬限制请求频率易如反掌!

总结

walrusredis-py进行了很好的数据库系统二次封装索引符号和详图符号,用起来非常顺手。除了上面提到的三个功能外,它还可以实现几行代码生成布隆过滤器,redistribute实现自动补全功能,实现简易图数据库等等。大家可以访问它的官方文档了解详细使用说明[1]。