Redis¶

Initial Redis Class¶

Redis is a good choice for caching, that why we need to use it to store the token or users data some configuration relate to cache.

In Python we use redis/redis-py library to connect to Redis.

Redis CRUD Operations¶

Redis CRUD is a simple CRUD operation, we can use it to store the data in Redis. we need to create some methods. (Get Key-Value, Set Key-Value, Delete Key, etc.)

async def get(self, key: str) -> str:
    return await self._redis.get(key)

async def delete(self, key: str) -> None:
    await self._redis.delete(key)
    return None

async def keys(self, match: str) -> Iterable[str]:
    return await self._redis.keys(match)

async def set(
    self, key: str, value: Union[str, bytes, int], expire: int = 0
) -> None:
    await self._redis.set(key, value, expire=expire)
    return None

Warning

It is very important not to await buffered command (ie self._redis.set('foo', '123')) as it will block forever.

Set String Value¶

Using a prebuilt functions using setnx and incr to set the value of the key.

async def setnx(self, key: str, value: Union[str, bytes, int], expire: int) -> None:
    await self._redis.setnx(key, value)
    await self._redis.expire(key, expire)
    return None

async def incr(self, key: str) -> str:
    return await self._redis.incr(key)

Dispatch Action¶

After we have the client we need to dispatch the action to the client, this gonna help to publish .json file to the client.

async def dispatch_action(
    self, channel: str,
    action: str,
    payload: dict) -> None:
        await self._redis.publish_json(channel,
        {"action": action, "payload": payload})
        return None