python walrus implements caching with expiration time and full-text search

 

In the previous article, I introduced you to using Python's own LRU cache to implement a cache with expiration time: One Day One Technique: Implementing an LRU Cache with Expiration Time. I also talked about inverted indexing: using inverted indexing to improve string search efficiency extremely fast. However, these codes are difficult for beginners and can be written with errors.

In fact, all of these functions can actually be implemented using Redis, and they can be done in just one minute each. The full-text search feature can even intelligently identify spelling errors when searching for English.

To achieve these functions, only two things need to be done：

1. Installing Redis

2. Python installation of third-party libraries: walrus

After the installation is complete, let's see how easy it is：

Cache decorator with expiration time

We want to implement a decorator which decorates a function. Let me use the cached data when accessing the function multiple times within 1 minute; re-execute the function's internal code only after 1 minute：

import time
import datetime
from walrus import Database

db = Database()
cache = db.cache()

@cache.cached(timeout=60)
def test():
    print('run')
    now = datetime.datetime.now()
    return now

now = test()
print('The data returned by the function is：', now)
time.sleep(10) # Wait 10 seconds, at this point the cache will be used
print('The data returned by the function is：', test())
time.sleep(5) # Wait 5 seconds, at this point the cache is still in use
print('The data returned by the function is：', test())

time.sleep(50)  # Let the time exceed the cache time
print('The data returned by the function is：', test())

 

Full text search

Let's take a look at the full-text search function, which is also very simple to implement：

from walrus import Database

db = Database()
search = db.Index('xxx')  # The name is arbitrary
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) # The first parameter cannot be repeated
search.add('docid2', poem2)
search.add('docid3', poem3)


for doc in search.search('end'):
    print(doc['content'])

The running effect is shown in the following figure：

If you want him to be compatible with spelling errors, then you can change search = db.Index('xxx') to search = db.Index('xxx', metaphone=True) and run the effect as shown below：

Unfortunately, however, this full-text search function only supports English.

Frequency limitation

Sometimes we want to limit the frequency of calls to a certain function, or we want to limit the frequency of IP access to a certain interface of a website. This can be easily done with walrus：

import time
from walrus import Database

db = Database()
rate = db.rate_limit('xxx', limit=5, per=60) # Only 5 calls per minute

for _ in range(35):
    if rate.limit('xxx'):
        print('Too frequent visits!')
    else:
        print('No access frequency limit triggered yet')
    time.sleep(2)

 

Where the parameter limit indicates how many times it can appear and per indicates over how long.

rate.limit will start checking how often this parameter appears in the set time as long as the same parameter is passed in.

You may think that this example doesn't tell us much, so let's combine it with FastAPI and use it to limit the frequency of IP access to the interface. Write the following code

：

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) # Only 5 calls per minute

app = FastAPI()


@app.exception_handler(RateLimitException)
def parse_rate_litmit_exception(request: Request, exc: RateLimitException):
    msg = {'success': False, 'msg': f'Your ip access is too fast！'}
    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 = 'Important Data'
    return {'success': True, 'data': data}

The above code defines a global exception interceptor：

@app.exception_handler(RateLimitException)
def parse_rate_litmit_exception(request: Request, exc: RateLimitException):
    msg = {'success': False, 'msg': f'Your ip access is too fast！'}
    return JSONResponse(status_code=429, content=msg)