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 aioredis
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