[[442671]]

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

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

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

安裝Redis

Python安裝第三方庫：walrus

安裝完成以后，我們來看看它有多簡單：

帶過期時間的緩存裝飾器

我們想實現(xiàn)一個裝飾器，它裝飾一個函數(shù)。讓我在1分鐘內(nèi)多次訪問函數(shù)的時候，使用緩存的數(shù)據(jù);超過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ù)真正運行起來') 
    now = datetime.datetime.now() 
    return now 
 
now = test() 
print('函數(shù)返回的數(shù)據(jù)是：', now) 
time.sleep(10) # 等待10秒，此時會使用緩存 
print('函數(shù)返回的數(shù)據(jù)是：', test()) 
time.sleep(5) # 等待5秒，此時依然使用緩存 
print('函數(shù)返回的數(shù)據(jù)是：', test()) 
 
time.sleep(50)  # 讓時間超過緩存的時間 
print('函數(shù)返回的數(shù)據(jù)是：', test()) 

運行效果如下圖所示：

全文搜索

我們再來看看全文搜索功能，實現(xiàn)起來也很簡單：

from walrus import Database 
 
db = Database() 
search = db.Index('xxx')  # 這個名字隨便取 
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) # 第一個參數(shù)不能重復 
search.add('docid2', poem2) 
search.add('docid3', poem3) 
 
 
for doc in search.search('end'): 
    print(doc['content']) 

運行效果如下圖所示：

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

不過遺憾的是，這個全文搜索功能只支持英文。

頻率限制

我們有時候要限制調(diào)用某個函數(shù)的頻率，或者網(wǎng)站的某個接口要限制IP的訪問頻率。這個時候，使用walrus也可以輕松實現(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('訪問頻率太高！') 
    else: 
        print('還沒有觸發(fā)訪問頻率限制') 
    time.sleep(2) 

運行效果如下圖所示：

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

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

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

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'請喝杯茶，休息一下，你的ip: {request.client.host}訪問太快了！'} 
    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} 

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

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

在整個代碼的任何地方拋出了RateLimitException異常，就會進入這里的邏輯中。

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

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

頻率超過設定的值以后，訪問頁面就會報錯，如下圖所示：

總結(jié)

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

參考文獻

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

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