↑?關注 + 星標?，每天學Python新技能
后臺回復【大禮包】送你Python自學大禮包

作者：kingname

來源：未聞 Code

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

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

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

1. 安裝 Redis
2. Python 安裝第三方庫：walrus

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

帶過期時間的緩存裝飾器

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

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ù)據是：',?now)
time.sleep(10)?#?等待10秒，此時會使用緩存
print('函數(shù)返回的數(shù)據是：',?test())
time.sleep(5)?#?等待5秒，此時依然使用緩存
print('函數(shù)返回的數(shù)據是：',?test())

time.sleep(50)??#?讓時間超過緩存的時間
print('函數(shù)返回的數(shù)據是：',?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)，運行效果如下圖所示：

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

頻率限制

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

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('還沒有觸發(fā)訪問頻率限制')
????time.sleep(2)

運行效果如下圖所示：

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

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

你可能覺得這個例子并不能說明什么問題，那么我們跟 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?=?'重要數(shù)據'
????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異常，攔截到以后返回我們定義好的報錯信息。

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

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

總結

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

參考文獻

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

推薦閱讀
冒死曝光這個軟件，希望不要被封殺！
用 Python 自動化下載抖X無水印短視頻

您看此文用???分??秒，轉發(fā)只需1秒哦~