In previous articles, I introduced the use of

In fact, these functions can be implemented using Redis, and each function can be done in only 1 minute. The full-text search function can even intelligently identify spelling errors when searching English.

To achieve these functions, you only need to do two things:

1. Install Redis
2. Python installation third party Library: walrus

After the installation, let's see how simple it is:

Cache decorator with expiration time

We want to implement a decorator that decorates a function. Let me use the cached data when I access the function multiple times in 1 minute; Re execute the internal code of the function after more than 1 minute:

import time
import datetime
from walrus import Database

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

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

now = test()
print('The data returned by the function is:', now)
time.sleep(10) # Wait 10 seconds and the cache will be used
print('The data returned by the function is:', test())
time.sleep(5) # Wait 5 seconds and the cache is still used
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())

The operation effect is shown in the figure below:

Full text search

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

from walrus import Database

db = Database()
search = db.Index('xxx')  # Take the name casually
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 operation effect is shown in the figure below:

If you want him to be compatible with spelling mistakes, you can put search = dB Index ('xxx ') is changed to search = dB Index ('xxx ', metaphone=True). The operation effect is shown in the following figure:

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

Frequency limit

Sometimes we need to limit the frequency of calling a function, or the frequency of IP access to an interface of the website. At this time, walrus can also be easily implemented:

import time
from walrus import Database

db = Database()
rate = db.rate_limit('xxx', limit=5, per=60) # Can only be called 5 times per minute

for _ in range(35):
    if rate.limit('xxx'):
        print('Access frequency is too high!')
    else:
        print('The access frequency limit has not been triggered')
    time.sleep(2)

The operation effect is shown in the figure below:

The parameter limit indicates how many times it can occur, and per indicates how long it will occur.

rate.limit as long as the same parameter is passed in, it will start to check the frequency of this parameter within the set time.

You may think that this example does not illustrate any problem, so let's combine it with FastAPI to limit the frequency of IP access interfaces. 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) # Can only be called 5 times per minute

app = FastAPI()


@app.exception_handler(RateLimitException)
def parse_rate_litmit_exception(request: Request, exc: RateLimitException):
    msg = {'success': False, 'msg': f'Have a cup of tea and have a rest, please ip: {request.client.host}Visit 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'Have a cup of tea and have a rest, please ip: {request.client.host}Visit too fast!'}
    return JSONResponse(status_code=429, content=msg)

If a RateLimitException is thrown anywhere in the whole code, it will enter the logic here.

Use the decorator @ rate rate_ Limited decorates a routing function, and the decorator should be closer to the function. The routing function receives whatever parameters it receives. In the above example, we only received the request parameter to obtain the visitor's IP. If it is found that the access frequency of this IP exceeds the limit, a RateLimitException is thrown. Therefore, the previously defined global interceptor will intercept the RateLimitException exception, and then return the defined error message.

Access the page within the frequency range and return normal JSON data:

When the frequency exceeds the set value, an error will be reported when accessing the page, as shown in the following figure:

summary

walrus has a good secondary packaging of redis py, which is very easy to use. In addition to the three functions I mentioned above, it can also realize several lines of code to generate bloom filter, automatic completion function, simple graph database and so on. You can access its official documents for detailed instructions [1].

reference

[1] Official documents for detailed instructions: https://walrus.readthedocs.io/en/latest/getting-started.html

END

In service employees of first and second tier large factories

More than ten years old programming bird

College students at home and abroad

Newcomers to primary and secondary schools