Use namedtuple for PEL entries as requested in code review

Changed PEL (Pending Entries List) storage from plain tuple to PelEntry
namedtuple with fields: consumer_name, time_read, times_delivered.

This improves code readability by using named attributes instead of indices
when accessing PEL data throughout the codebase.

Addresses review feedback from maintainer.

Author: zzstoatzz

fakeredis/model/_stream.py

@@ -26,6 +26,13 @@ def parse_str(entry_key: AnyStr) -> "StreamEntryKey":
         return StreamEntryKey(timestamp, sequence)
 
 
+class PelEntry(NamedTuple):
+    """Pending Entry List entry: tracks consumer ownership and delivery count"""
+    consumer_name: bytes
+    time_read: int
+    times_delivered: int
+
+
 class StreamRangeTest:
     """Argument converter for sorted set LEX endpoints."""
 
@@ -92,8 +99,8 @@ def __init__(
         self.last_delivered_key = start_key
         self.last_ack_key = start_key
         # Pending entry List, see https://redis.io/commands/xreadgroup/
-        # msg_id -> (consumer_name, time_read, times_delivered)
-        self.pel: Dict[StreamEntryKey, Tuple[bytes, int, int]] = {}
+        # msg_id -> PelEntry(consumer_name, time_read, times_delivered)
+        self.pel: Dict[StreamEntryKey, PelEntry] = {}
 
     def set_id(self, last_delivered_str: bytes, entries_read: Optional[int]) -> None:
         """Set last_delivered_id for the group"""
@@ -152,7 +159,7 @@ def group_read(
         if not noack:
             for k in ids_read:
                 # Initialize with times_delivered=1 for new messages
-                self.pel[k] = (consumer_name, _time, 1)
+                self.pel[k] = PelEntry(consumer_name, _time, 1)
         if len(ids_read) > 0:
             self.last_delivered_key = max(self.last_delivered_key, ids_read[-1])
             self.entries_read = (self.entries_read or 0) + len(ids_read)
@@ -161,9 +168,9 @@ def group_read(
         return [self.stream.format_record(x) for x in ids_read]
 
     def _calc_consumer_last_time(self) -> None:
-        # pel values are now (consumer_name, time_read, times_delivered)
+        # pel values are PelEntry namedtuples
         # Extract just consumer_name and time_read for grouping
-        new_last_success_map = {k: min(v, key=lambda x: x[1])[1] for k, v in itertools.groupby(self.pel.values(), key=itemgetter(0))}
+        new_last_success_map = {k: min(v, key=lambda x: x.time_read).time_read for k, v in itertools.groupby(self.pel.values(), key=lambda x: x.consumer_name)}
         for consumer in new_last_success_map:
             if consumer not in self.consumers:
                 self.consumers[consumer] = StreamConsumerInfo(consumer)
@@ -178,7 +185,7 @@ def ack(self, args: Tuple[bytes]) -> int:
             except Exception:
                 continue
             if parsed in self.pel:
-                consumer_name = self.pel[parsed][0]
+                consumer_name = self.pel[parsed].consumer_name
                 self.consumers[consumer_name].pending -= 1
                 del self.pel[parsed]
                 res += 1
@@ -196,9 +203,9 @@ def pending(
         _time = current_time()
         relevant_ids = list(self.pel.keys())
         if consumer is not None:
-            relevant_ids = [k for k in relevant_ids if self.pel[k][0] == consumer]
+            relevant_ids = [k for k in relevant_ids if self.pel[k].consumer_name == consumer]
         if idle is not None:
-            relevant_ids = [k for k in relevant_ids if self.pel[k][1] + idle < _time]
+            relevant_ids = [k for k in relevant_ids if self.pel[k].time_read + idle < _time]
         if start is not None and end is not None:
             relevant_ids = [
                 k
@@ -212,10 +219,10 @@ def pending(
             relevant_ids = sorted(relevant_ids)[:count]
 
         # Return all 4 fields: message_id, consumer, time_since_delivered, times_delivered
-        return [[k.encode(), self.pel[k][0], _time - self.pel[k][1], self.pel[k][2]] for k in relevant_ids]
+        return [[k.encode(), self.pel[k].consumer_name, _time - self.pel[k].time_read, self.pel[k].times_delivered] for k in relevant_ids]
 
     def pending_summary(self) -> List[Any]:
-        counter = Counter([self.pel[k][0] for k in self.pel])
+        counter = Counter([self.pel[k].consumer_name for k in self.pel])
         data = [
             len(self.pel),
             min(self.pel).encode() if len(self.pel) > 0 else None,
@@ -245,18 +252,18 @@ def claim(
             if key not in self.pel:
                 if force:
                     # Force claim msg - initialize with times_delivered=1
-                    self.pel[key] = (consumer_name, _time, 1)
+                    self.pel[key] = PelEntry(consumer_name, _time, 1)
                     if key in self.stream:
                         claimed_msgs.append(key)
                     else:
                         deleted_msgs.append(key)
                         del self.pel[key]
                 continue
-            if curr_time - self.pel[key][1] < min_idle_ms:
+            if curr_time - self.pel[key].time_read < min_idle_ms:
                 continue  # Not idle enough time to be claimed
             # Increment times_delivered when claiming
-            old_times_delivered = self.pel[key][2]
-            self.pel[key] = (consumer_name, _time, old_times_delivered + 1)
+            old_times_delivered = self.pel[key].times_delivered
+            self.pel[key] = PelEntry(consumer_name, _time, old_times_delivered + 1)
             if key in self.stream:
                 claimed_msgs.append(key)
             else:
@@ -268,7 +275,7 @@ def claim(
     def read_pel_msgs(self, min_idle_ms: int, start: bytes, count: int) -> List[StreamEntryKey]:
         start_key = StreamEntryKey.parse_str(start)
         curr_time = current_time()
-        msgs = sorted([k for k in self.pel if (curr_time - self.pel[k][1] >= min_idle_ms) and k >= start_key])
+        msgs = sorted([k for k in self.pel if (curr_time - self.pel[k].time_read >= min_idle_ms) and k >= start_key])
         count = min(count, len(msgs))
         return msgs[:count]