[[442671]]

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

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

要實現這些功能，只需要做兩件事：

安裝Redis

Python安裝第三方庫：walrus

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

帶過期時間的緩存裝飾器

我們想實現一個裝飾器，它裝飾一個函數。讓我在1分鐘內多次訪問函數的時候，使用緩存的數據;超過1分鐘以后才重新執行函數的內部代碼：

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

運行效果如下圖所示：

全文搜索

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

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) # 第一個參數不能重復 
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)，運行效果如下圖所示：

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

頻率限制

我們有時候要限制調用某個函數的頻率，或者網站的某個接口要限制IP的訪問頻率。這個時候，使用walrus也可以輕松實現：

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

運行效果如下圖所示：

其中參數limit表示能出現多少次，per表示在多長時間內。

rate.limit只要傳入相同的參數，那么就會開始檢查這個參數在設定的時間內出現的頻率。

你可能覺得這個例子并不能說明什么問題，那么我們跟FastAPI結合一下，用來限制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) # 每分鐘只能調用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 = '重要數據' 
    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裝飾一個路由函數，并且這個裝飾器要更靠近函數。路由函數接收什么參數，它就接收什么參數。在上面的例子中，我們只接收了request參數，用于獲取訪問者的IP。發現這個IP的訪問頻率超過了限制，就拋出一個RateLimitException。于是前面定義好的全局攔截器就會攔截RateLimitException異常，攔截到以后返回我們定義好的報錯信息。

在頻率范圍內訪問頁面，返回正常的JSON數據：

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

總結

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

參考文獻

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

本文轉載自微信公眾號「未聞Code」，可以通過以下二維碼關注。轉載本文請聯系未聞Code公眾號。