feat: add optional multi-key priority support for RedisSpider

- redis_key can now be str (unchanged behavior) or list[str]
- List mode: priority ordered (index 0 = highest)
- Drains current queue for throughput
- Optional redis_key_check_interval for periodic preemption
- Idle detection considers all configured keys
- Keys are templated, deduplicated, and validated
- Backward compatible: single-key behavior unchanged

Closes #310

Author: rmax

src/scrapy_redis/defaults.py

@@ -26,4 +26,5 @@
 START_URLS_KEY = "%(name)s:start_urls"
 START_URLS_AS_SET = False
 START_URLS_AS_ZSET = False
+REDIS_KEY_CHECK_INTERVAL = None
 MAX_IDLE_TIME = 0

src/scrapy_redis/spiders.py

@@ -19,13 +19,16 @@ class RedisMixin:
     redis_key = None
     redis_batch_size = None
     redis_encoding = None
+    redis_key_check_interval = None
 
     # Redis client placeholder.
     server = None
 
     # Idle start time
     spider_idle_start_time = int(time.time())
     max_idle_time = None
+    _redis_keys = None
+    _last_priority_scan = 0
 
     def start_requests(self):
         """Returns a batch of start requests from redis."""
@@ -56,10 +59,7 @@ def setup_redis(self, crawler=None):
                 defaults.START_URLS_KEY,
             )
 
-        self.redis_key = self.redis_key % {"name": self.name}
-
-        if not self.redis_key.strip():
-            raise ValueError("redis_key must not be empty")
+        self.redis_key, self._redis_keys = self._normalize_redis_key(self.redis_key)
 
         if self.redis_batch_size is None:
             self.redis_batch_size = settings.getint(
@@ -76,12 +76,6 @@ def setup_redis(self, crawler=None):
                 "REDIS_ENCODING", defaults.REDIS_ENCODING
             )
 
-        self.logger.info(
-            "Reading start URLs from redis key '%(redis_key)s' "
-            "(batch size: %(redis_batch_size)s, encoding: %(redis_encoding)s)",
-            self.__dict__,
-        )
-
         self.server = connection.from_settings(crawler.settings)
 
         if settings.getbool("REDIS_START_URLS_AS_SET", defaults.START_URLS_AS_SET):
@@ -94,6 +88,38 @@ def setup_redis(self, crawler=None):
             self.fetch_data = self.pop_list_queue
             self.count_size = self.server.llen
 
+        if self.redis_key_check_interval is None:
+            self.redis_key_check_interval = settings.get(
+                "REDIS_KEY_CHECK_INTERVAL", defaults.REDIS_KEY_CHECK_INTERVAL
+            )
+
+        if self.redis_key_check_interval in (None, 0, "0"):
+            self.redis_key_check_interval = 0
+        else:
+            try:
+                self.redis_key_check_interval = int(self.redis_key_check_interval)
+            except (TypeError, ValueError):
+                raise ValueError("redis_key_check_interval must be an integer or None")
+
+        self._last_priority_scan = 0
+
+        if self._redis_keys:
+            self.logger.info(
+                "Reading start URLs from redis keys '%(redis_keys)s' "
+                "(batch size: %(redis_batch_size)s, encoding: %(redis_encoding)s)",
+                {
+                    "redis_keys": self._redis_keys,
+                    "redis_batch_size": self.redis_batch_size,
+                    "redis_encoding": self.redis_encoding,
+                },
+            )
+        else:
+            self.logger.info(
+                "Reading start URLs from redis key '%(redis_key)s' "
+                "(batch size: %(redis_batch_size)s, encoding: %(redis_encoding)s)",
+                self.__dict__,
+            )
+
         if self.max_idle_time is None:
             self.max_idle_time = settings.get(
                 "MAX_IDLE_TIME_BEFORE_CLOSE", defaults.MAX_IDLE_TIME
@@ -108,6 +134,83 @@ def setup_redis(self, crawler=None):
         # that's when we will schedule new requests from redis queue
         crawler.signals.connect(self.spider_idle, signal=signals.spider_idle)
 
+    def _normalize_redis_key(self, redis_key):
+        if isinstance(redis_key, str):
+            redis_key = redis_key % {"name": self.name}
+            if not redis_key.strip():
+                raise ValueError("redis_key must not be empty")
+            return redis_key, None
+
+        if isinstance(redis_key, list):
+            redis_keys = []
+            seen = set()
+            for key in redis_key:
+                formatted_key = key % {"name": self.name}
+                stripped_key = formatted_key.strip()
+                if not stripped_key:
+                    raise ValueError("redis_key must not be empty")
+                if stripped_key not in seen:
+                    seen.add(stripped_key)
+                    redis_keys.append(stripped_key)
+
+            if not redis_keys:
+                raise ValueError("redis_key list is empty after normalization")
+
+            return redis_keys[0], redis_keys
+
+        return redis_key, None
+
+    def _is_multi_key_mode(self):
+        return bool(self._redis_keys)
+
+    def _any_key_has_items(self):
+        if not self._is_multi_key_mode():
+            return self.count_size(self.redis_key) > 0
+        return sum(self.count_size(key) for key in self._redis_keys) > 0
+
+    def _select_priority_key(self):
+        """Return the first non-empty key from _redis_keys in priority order."""
+        if not self._redis_keys:
+            return self.redis_key
+        for key in self._redis_keys:
+            if self.count_size(key) > 0:
+                return key
+        return None
+
+    def _maybe_switch_to_higher_priority_key(self):
+        if not self._redis_keys:
+            return
+
+        selected_key = self._select_priority_key()
+        if selected_key is None:
+            return
+
+        try:
+            current_index = self._redis_keys.index(self.redis_key)
+        except ValueError:
+            current_index = len(self._redis_keys)
+
+        selected_index = self._redis_keys.index(selected_key)
+        if selected_index < current_index:
+            self.redis_key = selected_key
+
+    def _maybe_check_priority_scan(self):
+        if not self._redis_keys:
+            return
+
+        if self.count_size(self.redis_key) == 0:
+            selected_key = self._select_priority_key()
+            if selected_key is not None:
+                self.redis_key = selected_key
+
+        if not self.redis_key_check_interval:
+            return
+
+        current_time = time.time()
+        if current_time - self._last_priority_scan >= self.redis_key_check_interval:
+            self._maybe_switch_to_higher_priority_key()
+            self._last_priority_scan = current_time
+
     def pop_list_queue(self, redis_key, batch_size):
         with self.server.pipeline() as pipe:
             pipe.lrange(redis_key, 0, batch_size - 1)
@@ -125,6 +228,7 @@ def pop_priority_queue(self, redis_key, batch_size):
     def next_requests(self):
         """Returns a request to be scheduled or none."""
         # XXX: Do we need to use a timeout here?
+        self._maybe_check_priority_scan()
         found = 0
         datas = self.fetch_data(self.redis_key, self.redis_batch_size)
         for data in datas:
@@ -208,6 +312,7 @@ def make_request_from_data(self, data):
     def schedule_next_requests(self):
         """Schedules a request if available"""
         # TODO: While there is capacity, schedule a batch of redis requests.
+        self._maybe_check_priority_scan()
         for req in self.next_requests():
             # see https://github.com/scrapy/scrapy/issues/5994
             if scrapy_version >= (2, 6):
@@ -221,7 +326,9 @@ def spider_idle(self):
         or close spider when waiting seconds > MAX_IDLE_TIME_BEFORE_CLOSE.
         MAX_IDLE_TIME_BEFORE_CLOSE will not affect SCHEDULER_IDLE_BEFORE_CLOSE.
         """
-        if self.server is not None and self.count_size(self.redis_key) > 0:
+        self._maybe_check_priority_scan()
+
+        if self.server is not None and self._any_key_has_items():
             self.spider_idle_start_time = int(time.time())
 
         self.schedule_next_requests()