CodehubRedis in Python
docker-compose.yml
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:
- cacheyml
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: bridgeredisClient.py
python
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)python
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
py
HOST = "127.0.0.1"
PORT = 63791
PASSWORD = ""pyproject.toml
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"