checks.py 文件源码

def check_redis(app_configs=None, **kwargs):
    from django_rq.queues import get_queue
    from django_rq.workers import Worker

    errors = []

    try:
        queue = get_queue()
        Worker.all(queue.connection)
    except Exception as e:
        conn_settings = queue.connection.connection_pool.connection_kwargs
        errors.append(checks.Critical(
            _("Could not connect to Redis (%s)", e),
            hint=_("Make sure Redis is running on "
                   "%(host)s:%(port)s") % conn_settings,
            id="pootle.C001",
        ))
    else:
        redis_version = tuple(int(x) for x
                              in (queue.connection
                                       .info()["redis_version"].split(".")))
        if redis_version < REDIS_MINIMUM_REQUIRED_VERSION:
            errors.append(checks.Critical(
                _("Your version of Redis is too old."),
                hint=_("Update your system's Redis server package to at least "
                       "version %s", str(REDIS_MINIMUM_REQUIRED_VERSION)),
                id="pootle.C007",
            ))

        if len(queue.connection.smembers(Worker.redis_workers_keys)) == 0:
            # We need to check we're not running manage.py rqworker right now..
            import sys
            if len(sys.argv) > 1 and sys.argv[1] in RQWORKER_WHITELIST:
                errors.append(checks.Warning(
                    _("No RQ Worker running."),
                    hint=_("Run new workers with manage.py rqworker"),
                    id="pootle.W001",
                ))

    return errors