Блог

Ищем надежный способ совместного использования объектов соединения в проекте django, аналогично подключениям к базе данных django.

 from django.db import connections
with connections['default'].cursor() as cursor:
    cursor.execute(sql, params)
  

Не обязательно с тем же API, но мне нужен аналогичный способ получения пулов соединений redis, который инициализируется при запуске сервера django, чтобы я мог повторно использовать один пул соединений из разных частей проекта.

У вас есть какие-либо предложения или идеи, как я могу подойти к этой проблеме?

Решение, которое сработало для меня, было найдено в django CacheHandler

В этом случае для построения соединения REDIS используется redis-sentinel-url.

settings.py — Повторный вывод определений строк подключения в

 REDIS = {
    "default": "redis:///",
    "secondary": "redis sentinel:///"
}
  

RedisPoolHandler реализация класса, аналогичная django CacheHandler

 from threading import local

import redis_sentinel_url
from django.conf import settings
from django.core.cache import InvalidCacheBackendError


class RedisClientHandler:
    """
    A Redis Client Handler to manage access to Redis instances.
    Ensure only one instance of each alias exists per thread.
    """
    def __init__(self):
        self._clients = local()

    def __getitem__(self, alias):
        try:
            return self._redis_clients.caches[alias]
        except AttributeError:
            self._redis_clients.caches = {}
        except KeyError:
            pass

        if alias not in settings.REDIS:
            raise InvalidCacheBackendError(
                "Could not find config for '%s' in settings.REDIS" % alias
            )

        value = settings.REDIS[alias]
        sentinel, redis_pool = redis_sentinel_url.connect(value, client_options={"decode_responses": True})

        self._redis_clients.caches[alias] = redis_pool
        return redis_pool

    def all(self):
        return getattr(self._redis_clients, '_redis_clients', {}).values()


redis_clients = RedisPoolHandler()
  

Пример использования ниже:

 from path.to.classfile import redis_clients

redis_client = redis_clients['default']