Method lpush

tests/scheduler/test_splitwise_scheduler.py:97–105

(self, key: str, *values: Any)

        # ------------------------------- list operations used by the scheduler
        def lpush(self, key: str, *values: Any) -> int:
            self._check_expiration(key)
            items = list(values)
            if not items:
                return 0
            bucket = self.storage.setdefault(key, [])
            for value in items:
                bucket.insert(0, value)
            return len(bucket)

        def rpop(self, key: str, count: Optional[int] = None) -> Optional[Any]:
            self._check_expiration(key)