[[442671]]

在以前的文章里面，我給大家介紹了使用Python自帶的LRU緩存實(shí)現(xiàn)帶有過(guò)期時(shí)間的緩存：一日一技：實(shí)現(xiàn)有過(guò)期時(shí)間的LRU緩存。也講過(guò)倒排索引：使用倒排索引極速提高字符串搜索效率。但這些代碼對(duì)初學(xué)者來(lái)說(shuō)比較難，寫(xiě)起來(lái)可能會(huì)出錯(cuò)。

實(shí)際上，這些功能其實(shí)都可以使用Redis來(lái)實(shí)現(xiàn)，而且每個(gè)功能只需要1分鐘就能做出來(lái)。全文搜索功能在搜索英文的時(shí)候，甚至可以智能識(shí)別拼寫(xiě)錯(cuò)誤的問(wèn)題。

要實(shí)現(xiàn)這些功能，只需要做兩件事：

安裝Redis

Python安裝第三方庫(kù)：walrus

安裝完成以后，我們來(lái)看看它有多簡(jiǎn)單：

帶過(guò)期時(shí)間的緩存裝飾器

我們想實(shí)現(xiàn)一個(gè)裝飾器，它裝飾一個(gè)函數(shù)。讓我在1分鐘內(nèi)多次訪問(wèn)函數(shù)的時(shí)候，使用緩存的數(shù)據(jù);超過(guò)1分鐘以后才重新執(zhí)行函數(shù)的內(nèi)部代碼：

import time 
import datetime 
from walrus import Database 
 
db = Database() 
cache = db.cache() 
 
@cache.cached(timeout=60) 
def test(): 
    print('函數(shù)真正運(yùn)行起來(lái)') 
    now = datetime.datetime.now() 
    return now 
 
now = test() 
print('函數(shù)返回的數(shù)據(jù)是：', now) 
time.sleep(10) # 等待10秒，此時(shí)會(huì)使用緩存 
print('函數(shù)返回的數(shù)據(jù)是：', test()) 
time.sleep(5) # 等待5秒，此時(shí)依然使用緩存 
print('函數(shù)返回的數(shù)據(jù)是：', test()) 
 
time.sleep(50)  # 讓時(shí)間超過(guò)緩存的時(shí)間 
print('函數(shù)返回的數(shù)據(jù)是：', test()) 

運(yùn)行效果如下圖所示：

全文搜索

我們?cè)賮?lái)看看全文搜索功能，實(shí)現(xiàn)起來(lái)也很簡(jiǎn)單：

from walrus import Database 
 
db = Database() 
search = db.Index('xxx')  # 這個(gè)名字隨便取 
poem1 = 'Early in the day it was whispered that we should sail in a boat, only thou and I, and never a soul in the world would know of this our pilgrimage to no country and to no end.' 
poem2 = 'Had I the heavens’ embroidered cloths,Enwrought with golden and silver light' 
poem3 = 'to be or not to be, that is a question.' 
 
search.add('docid1', poem1) # 第一個(gè)參數(shù)不能重復(fù) 
search.add('docid2', poem2) 
search.add('docid3', poem3) 
 
 
for doc in search.search('end'): 
    print(doc['content']) 

運(yùn)行效果如下圖所示：

如果你想讓他兼容拼寫(xiě)錯(cuò)誤，那么可以把search = db.Index('xxx')改成search = db.Index('xxx’, metaphone=True)，運(yùn)行效果如下圖所示：

不過(guò)遺憾的是，這個(gè)全文搜索功能只支持英文。

頻率限制

我們有時(shí)候要限制調(diào)用某個(gè)函數(shù)的頻率，或者網(wǎng)站的某個(gè)接口要限制IP的訪問(wèn)頻率。這個(gè)時(shí)候，使用walrus也可以輕松實(shí)現(xiàn)：

import time 
from walrus import Database 
 
db = Database() 
rate = db.rate_limit('xxx', limit=5, per=60) # 每分鐘只能調(diào)用5次 
 
for _ in range(35): 
    if rate.limit('xxx'): 
        print('訪問(wèn)頻率太高！') 
    else: 
        print('還沒(méi)有觸發(fā)訪問(wèn)頻率限制') 
    time.sleep(2) 

運(yùn)行效果如下圖所示：

其中參數(shù)limit表示能出現(xiàn)多少次，per表示在多長(zhǎng)時(shí)間內(nèi)。

rate.limit只要傳入相同的參數(shù)，那么就會(huì)開(kāi)始檢查這個(gè)參數(shù)在設(shè)定的時(shí)間內(nèi)出現(xiàn)的頻率。

你可能覺(jué)得這個(gè)例子并不能說(shuō)明什么問(wèn)題，那么我們跟FastAPI結(jié)合一下，用來(lái)限制IP訪問(wèn)接口的頻率。編寫(xiě)如下代碼：

from walrus import Database, RateLimitException 
from fastapi import FastAPI, Request 
from fastapi.responses import JSONResponse 
 
db = Database() 
rate = db.rate_limit('xxx', limit=5, per=60) # 每分鐘只能調(diào)用5次 
 
app = FastAPI() 
 
 
@app.exception_handler(RateLimitException) 
def parse_rate_litmit_exception(request: Request, exc: RateLimitException): 
    msg = {'success': False, 'msg': f'請(qǐng)喝杯茶，休息一下，你的ip: {request.client.host}訪問(wèn)太快了！'} 
    return JSONResponse(status_code=429, content=msg) 
 
@app.get('/') 
def index(): 
    return {'success': True} 
 
 
@app.get('/important_api') 
@rate.rate_limited(lambda request: request.client.host) 
def query_important_data(request: Request): 
    data = '重要數(shù)據(jù)' 
    return {'success': True, 'data': data} 

上面代碼定義了一個(gè)全局的異常攔截器：

@app.exception_handler(RateLimitException) 
def parse_rate_litmit_exception(request: Request, exc: RateLimitException): 
    msg = {'success': False, 'msg': f'請(qǐng)喝杯茶，休息一下，你的ip: {request.client.host}訪問(wèn)太快了！'} 
    return JSONResponse(status_code=429, content=msg) 

在整個(gè)代碼的任何地方拋出了RateLimitException異常，就會(huì)進(jìn)入這里的邏輯中。

使用裝飾器@rate.rate_limited裝飾一個(gè)路由函數(shù)，并且這個(gè)裝飾器要更靠近函數(shù)。路由函數(shù)接收什么參數(shù)，它就接收什么參數(shù)。在上面的例子中，我們只接收了request參數(shù)，用于獲取訪問(wèn)者的IP。發(fā)現(xiàn)這個(gè)IP的訪問(wèn)頻率超過(guò)了限制，就拋出一個(gè)RateLimitException。于是前面定義好的全局?jǐn)r截器就會(huì)攔截RateLimitException異常，攔截到以后返回我們定義好的報(bào)錯(cuò)信息。

在頻率范圍內(nèi)訪問(wèn)頁(yè)面，返回正常的JSON數(shù)據(jù)：

頻率超過(guò)設(shè)定的值以后，訪問(wèn)頁(yè)面就會(huì)報(bào)錯(cuò)，如下圖所示：

總結(jié)

walrus對(duì)redis-py進(jìn)行了很好的二次封裝，用起來(lái)非常順手。除了上面我提到的三個(gè)功能外，它還可以實(shí)現(xiàn)幾行代碼生成布隆過(guò)濾器，實(shí)現(xiàn)自動(dòng)補(bǔ)全功能，實(shí)現(xiàn)簡(jiǎn)易圖數(shù)據(jù)庫(kù)等等。大家可以訪問(wèn)它的官方文檔了解詳細(xì)使用說(shuō)明[1]。

參考文獻(xiàn)

[1] 官方文檔了解詳細(xì)使用說(shuō)明: https://walrus.readthedocs.io/en/latest/getting-started.html

本文轉(zhuǎn)載自微信公眾號(hào)「未聞Code」，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系未聞Code公眾號(hào)。