This is the 17th day of my participation in the First Challenge 2022.

In a previous article, I showed you how to implement caches with expiration times using Python’s native LRU cache: Implement LRU caches with expiration times. I also talked about inverted indexes: using inverted indexes to speed up string searches. But this code is difficult for beginners and can make mistakes.

In fact, all of these features can be implemented using Redis, and each feature takes only 1 minute to create. The full text search function can even intelligently identify spelling errors when searching English.

To implement these features, you only need to do two things:

  1. Install Redis
  2. Python install third-party libraries:walrus

Once installed, 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 cached data when I access the function multiple times in a minute; Re-executing the internal code of a function after more than a minute:

import time import datetime from walrus import Database db = Database() cache = db.cache() @cache.cached(timeout=60) def  test(): Now = datetime.datetime.now() return now now = test() print(' function actually run ') now = datetime.datetime.now() return now now = test() print(' function actually run ') ', now) time.sleep(10) # print(' print(') ') time.sleep(10) # print(' print(') ') time.sleep(10) # print(') ' ', test()) time.sleep(50) # print()Copy the code

The operating effect is shown in the figure below:

Full-text search

Let’s take a look at full-text search, which is also easy to implement:

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' renewables with golden and silver light' poem3 = 'to be or not to be Add ('docid2', poem2) search. Add ('docid2', poem2) search. Add ('docid3', poem2) search. poem3) for doc in search.search('end'): print(doc['content'])Copy the code

The operating effect is shown in the figure below:

If you want to make it misspell-compatible, change search = db.Index(‘ XXX ‘) to search = db.Index(‘ XXX ‘, Metaphone =True) as shown below:

Unfortunately, the full-text search is only available in English.

Frequency limit

Sometimes we want to limit how often a function is called, or how often an interface on a website can access an IP. At this point, it is also easy to implement using Walrus:

import time from walrus import Database db = Database() rate = db.rate_limit('xxx', limit=5, Per =60) # for _ in range(35): if rate.limit(' XXX '): print(' XXX ') ') else: print(' not triggered yet ') time.sleep(2)Copy the code

The operating effect is shown in the figure below:

Limit indicates the number of occurrences, and per indicates the duration. \

Rate. limit As soon as the same parameter is passed in, it will start checking how often that parameter occurs within the specified time.

If you think this example doesn’t prove anything, let’s combine it with FastAPI 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, App = FastAPI() @app.exception_handler(RateLimitException) def parse_rate_litmit_exception(request: Request, exc: RateLimitException): MSG = {'success': False, 'MSG ': f' please have a cup of tea and rest, your IP: {request.client.host} 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}Copy the code

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' please have a cup of tea and rest, your IP: {request.client.host} is too fast! '} return JSONResponse(status_code=429, content=msg)Copy the code

A RateLimitException thrown anywhere in the entire code goes into the logic here.

Decorate a routing function with the decorator @rate.rate_limited closer to the function. The routing function receives whatever arguments it receives. In the example above, we only receive the Request parameter to get the visitor’s IP.

Finding that this IP is accessed more frequently than the limit, it throws a RateLimitException. The previously defined global interceptor intercepts the RateLimitException and returns our defined error message.

Visit the page in 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 below:

conclusion

Walrus has a good secondary encapsulation of Redis-py, making it very handy to use. In addition to the three features mentioned above, it can also implement several lines of code to generate bloom filters, implement auto-completion, implement a simple graph database, and more. You can visit its official documentation for detailed instructions [1].