Redis in Python

docker-compose.yml

docker-compose.yml

version: '3.8'
services:
  cache:
    image: redis:6.2-alpine
    restart: always
    ports:
      - '127.0.0.1:63791:6379'
    command: redis-server --save 300 1 --loglevel warning --requirepass my-pass-word
    volumes: 
      - ./cache/redis:/data
  fetcher:
    image: u2py
    build: .
    restart: always
    command: poetry run python main.py
    volumes: 
      - .:/root/project
      - ./cache:/root/.cache
    working_dir: /root/project
    network_mode: host
    depends_on:
      - cache

docker-compose.yml

version: '3'
services:
  redis:
    image: redis:alpine
    restart: always
    command: redis-server --loglevel warning --requirepass password
    volumes: 
      - ./__redis_cache__/redis:/data
    networks:
      - redis_network
  app:
    image: u20p39
    command: poetry run python app.py
    restart: always 
    volumes:
      - ./project:/root/project
      - ./__env_cache__:/root/.cache
    working_dir: /root/project
    networks:
      - redis_network
    depends_on:
      - redis    
networks:
  redis_network:
    driver: bridge

redisClient.py

redis_helper.py

from redis import StrictRedis
from redis_config import *
import json


# redis-cli -h 127.0.0.1 -p 6379 -n 0 -a <> 

class RedisHelper():

    def __init__(self, db=0):
        self.r = StrictRedis(
            host=HOST,
            port=PORT,
            password=PASSWORD,
            db=db,
            socket_timeout=None,
            connection_pool=None,
            charset='utf-8',
            errors='strict',
            unix_socket_path=None,
            decode_responses=True,
        )

    def hset_add(self, name, field, data):
        return self.r.hset(name, field, data)

    def hset_get(self, name, field):
        return self.r.hget(name, field)

    def hset_del(self, name, field):
        return self.r.hdel(name, field)

    def hset_exists(self, name, field):
        return self.r.hexists(name, field)
        
    def hset_all(self, name):
        return self.r.hgetall(name)

    def list_push(self, name, data):
        return self.r.lpush(name, data)

    def list_pop(self, name):
        return self.r.rpop(name)

    def list_len(self, name):
        return self.r.llen(name)

class HashData(RedisHelper):

    def __init__(self, db, name):
        super().__init__(db)
        self.name = name
        
    def delete(self, key_name):
        self.r.delete(self.name, key_name)

    def has(self, key_name):
        return self.hset_exists(self.name, key_name)

    def get(self, key_name):
        try:
            return json.loads(self.hset_get(self.name, key_name))
        except:
            return None

    def add(self, key_name, data):
        return self.hset_add(self.name, key_name, json.dumps(data))

    def patch(self, key_name, data):
        data_new = self.get(key_name)
        if data_new:
            data_new.update(data)
            return self.add(key_name, data_new)
    
    def all(self):
        d = dict()
        for i, v in self.hset_all(self.name).items():
            d[i] = json.loads(v)
        return d

class ListData(RedisHelper):

    def __init__(self, db, name):
        super().__init__(db)
        self.name = name

    def remove(self):
        self.r.delete(self.name)

    def push(self, data):
        return self.list_push(self.name, json.dumps(data))

    def pop(self):
        try:
            return json.loads(self.list_pop(self.name))
        except:
            return None

    def size(self):
        return self.list_len(self.name)

redis_helper.py

import redis.asyncio as aioredis
import json

HOST = 'redis'
PORT = 6379
PASSWORD = 'my-password'

class RedisHelper:

    def __init__(self, db=0):
        self.db = db
        self.redis = None

    async def initialize(self):
        self.redis = aioredis.Redis(
            host=HOST,
            port=PORT,
            password=PASSWORD,
            db=self.db,
            encoding='utf-8',
            decode_responses=True
        )
    
    # message list
    async def list_push(self, name, data):
        return await self.redis.rpush(name, data)

    async def list_pop(self, name):
        return await self.redis.lpop(name)

    async def list_len(self, name):
        return await self.redis.llen(name)

    # cache key
    async def set_cache_value(self, name, data_to_cache, expire):
        await self.redis.set(name, data_to_cache, ex=expire)

    async def get_cache_value(self, name):
        return await self.redis.get(name, decode_responses=False)

    async def close(self):
        await self.redis.close()

class ListData(RedisHelper):

    def __init__(self, db, name):
        super().__init__(db=0)
        self.name = name

    async def remove(self):
        await self.redis.delete(self.name)

    async def push(self, data):
        return await self.list_push(self.name, json.dumps(data))

    async def pop(self):
        try:
            data = await self.list_pop(self.name)
            return json.loads(data) if data else None
        except Exception as e:
            # Handle exception (logging, etc.)
            return None

    async def size(self):
        return await self.list_len(self.name)

class CacheData(RedisHelper):

    def __init__(self, db=1):
        super().__init__(db)
        self.expire = 24*60*60

    async def set_value(self, name, data):
        await self.set_cache_value(name, data, self.expire)

    async def get_value(self, name):
        return await self.get_cache_value(name)

    async def exec_cache_value(self, tag, func, args=()):
        content = await self.get_value(tag)
        if not content:
            content = await func(*args)
            await self.set_value(tag, content)
        return content
    
class SetData(RedisHelper):

    def __init__(self, name=None, db=2):
        super().__init__(db)
        # self.expire = 24*60*60
        self.name = name

    async def sadd(self, key):
        await self.redis.sadd(self.name, key)
        # await self.redis.expire(self.name, self.expire)

    async def size(self):
        return await self.redis.scard(self.name)

    async def is_exists(self, key):
        return await self.redis.sismember(self.name, key)

redis_conf.py

HOST = "127.0.0.1"
PORT = 63791
PASSWORD = ""

pyproject.toml

[tool.poetry]
name = "project"
version = "0.1.0"
description = ""
authors = ["Your Name <you@example.com>"]
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.8"
redis = "^4.3.4"


[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"