Django, RQ и FakeRedis
Я часто в своих проектах использую связку Django + RQ вместо Celery. RQ удобный и максимально простой инструмент среди популярных Task Queue решений в экосистеме Python. Пару месяцев назад возникла необходимость тестировать код с сигналами в Django. Схема простая: в ответ на какое-то событие (создание объекта в БД, кастомный сигнал и т.д.) вызывался RQ Job через delay. Дело в том, что такое событие транслировалось ко всем получателям (receivers) как только объект удалялся из базы. Я активный пользователь pytest и создаю промежуточные объекты через стандартные фикстуры. Одно из решений — патчить/mockать job-функции во всех местах, где такие объекты создаются. Но это неудобно и непрактично, с развитием системы количество получателей может расти. Я нашел выход в подмене connection-класса в зависимости от условий. Условие в моём случае это наличие переменной FAKE_REDIS в Django settings.py. Когда FAKE_REDIS=True, то мы заменяем соединение с redis на инстанс класса FakeRedis из пакета fakeredis.
Чтобы этот подход сносно работал необходимо переписать job-декоратор. Вот что получилось у меня:
from rq.decorators import job as rq_job_decorator
from django_rq.queues import get_queue
def async_job(queue_name: str, *args: t.Any, **kwargs: t.Any) -> t.Any:
"""
The same as RQ's job decorator, but it automatically replaces the
``connection`` argument with a fake one if ``settings.FAKE_REDIS`` is set to ``True``.
"""
class LazyAsyncJob:
def __init__(self, f: t.Callable[..., t.Any]) -> None:
self.f = f
self.job: t.Optional[t.Callable[..., t.Any]] = None
def setup_connection(self) -> t.Callable[..., t.Any]:
if self.job:
return self.job
if settings.FAKE_REDIS:
from fakeredis import FakeRedis
queue = get_queue(queue_name, connection=FakeRedis()) # type: ignore
else:
queue = get_queue(queue_name)
RQ = getattr(settings, 'RQ', {})
default_result_ttl = RQ.get('DEFAULT_RESULT_TTL')
if default_result_ttl is not None:
kwargs.setdefault('result_ttl', default_result_ttl)
return rq_job_decorator(queue, *args, **kwargs)(self.f)
def delay(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
self.job = self.setup_connection()
return self.job.delay(*args, **kwargs) # type: ignore
def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
self.job = self.setup_connection()
return self.job(*args, **kwargs)
return LazyAsyncJob
И все rq таски необходимо декорировать через async_job:
@async_job('default')
def process_image():
pass
В Django тестах я использую фикстуры pytest:
@pytest.fixture(autouse=True)
def fake_redis(settings: t.Any) -> None:
settings.FAKE_REDIS = True
Параметр autouse=True нужен для того, чтобы абсолютно во всех тестах использовался fake_redis без явного на то указания.