selfprivacy-rest-api/selfprivacy_api/utils/redis_pool.py

42 lines
1 KiB
Python
Raw Normal View History

2022-10-28 08:09:36 +00:00
"""
Redis pool module for selfprivacy_api
"""
2023-07-20 15:24:26 +00:00
from os import environ
import redis
2022-10-28 08:09:36 +00:00
from selfprivacy_api.utils.singleton_metaclass import SingletonMetaclass
REDIS_SOCKET = "/run/redis-sp-api/redis.sock"
class RedisPool(metaclass=SingletonMetaclass):
"""
Redis connection pool singleton.
"""
def __init__(self):
2023-07-20 15:24:26 +00:00
if "USE_REDIS_PORT" in environ:
self._pool = redis.ConnectionPool(
host="127.0.0.1",
port=int(environ["USE_REDIS_PORT"]),
decode_responses=True,
)
else:
self._pool = redis.ConnectionPool.from_url(
f"unix://{REDIS_SOCKET}",
decode_responses=True,
)
2022-10-28 08:09:36 +00:00
self._pubsub_connection = self.get_connection()
def get_connection(self):
"""
Get a connection from the pool.
"""
return redis.Redis(connection_pool=self._pool)
def get_pubsub(self):
"""
Get a pubsub connection from the pool.
"""
return self._pubsub_connection.pubsub()