Merge branch 'master' into ps_add_hybrid_search

Author: petyaslavova

.github/workflows/codeql-analysis.yml

@@ -40,7 +40,7 @@ jobs:
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v3
+      uses: github/codeql-action/init@v4
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -51,7 +51,7 @@ jobs:
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: Autobuild
-      uses: github/codeql-action/autobuild@v3
+      uses: github/codeql-action/autobuild@v4
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -65,4 +65,4 @@ jobs:
     #   make release
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v3
+      uses: github/codeql-action/analyze@v4

redis/_parsers/helpers.py

@@ -137,11 +137,11 @@ def parse_sentinel_state(item):
     return result
 
 
-def parse_sentinel_master(response):
+def parse_sentinel_master(response, **options):
     return parse_sentinel_state(map(str_if_bytes, response))
 
 
-def parse_sentinel_state_resp3(response):
+def parse_sentinel_state_resp3(response, **options):
     result = {}
     for key in response:
         try:
@@ -154,27 +154,27 @@ def parse_sentinel_state_resp3(response):
     return result
 
 
-def parse_sentinel_masters(response):
+def parse_sentinel_masters(response, **options):
     result = {}
     for item in response:
         state = parse_sentinel_state(map(str_if_bytes, item))
         result[state["name"]] = state
     return result
 
 
-def parse_sentinel_masters_resp3(response):
-    return [parse_sentinel_state(master) for master in response]
+def parse_sentinel_masters_resp3(response, **options):
+    return [parse_sentinel_state_resp3(master) for master in response]
 
 
-def parse_sentinel_slaves_and_sentinels(response):
+def parse_sentinel_slaves_and_sentinels(response, **options):
     return [parse_sentinel_state(map(str_if_bytes, item)) for item in response]
 
 
-def parse_sentinel_slaves_and_sentinels_resp3(response):
-    return [parse_sentinel_state_resp3(item) for item in response]
+def parse_sentinel_slaves_and_sentinels_resp3(response, **options):
+    return [parse_sentinel_state_resp3(item, **options) for item in response]
 
 
-def parse_sentinel_get_master(response):
+def parse_sentinel_get_master(response, **options):
     return response and (response[0], int(response[1])) or None
 
 
@@ -268,13 +268,16 @@ def sort_return_tuples(response, **options):
     return list(zip(*[response[i::n] for i in range(n)]))
 
 
-def parse_stream_list(response):
+def parse_stream_list(response, **options):
     if response is None:
         return None
     data = []
     for r in response:
         if r is not None:
-            data.append((r[0], pairs_to_dict(r[1])))
+            if "claim_min_idle_time" in options:
+                data.append((r[0], pairs_to_dict(r[1]), *r[2:]))
+            else:
+                data.append((r[0], pairs_to_dict(r[1])))
         else:
             data.append((None, None))
     return data
@@ -332,16 +335,18 @@ def parse_xinfo_stream(response, **options):
     return data
 
 
-def parse_xread(response):
+def parse_xread(response, **options):
     if response is None:
         return []
-    return [[r[0], parse_stream_list(r[1])] for r in response]
+    return [[r[0], parse_stream_list(r[1], **options)] for r in response]
 
 
-def parse_xread_resp3(response):
+def parse_xread_resp3(response, **options):
     if response is None:
         return {}
-    return {key: [parse_stream_list(value)] for key, value in response.items()}
+    return {
+        key: [parse_stream_list(value, **options)] for key, value in response.items()
+    }
 
 
 def parse_xpending(response, **options):

redis/cluster.py

@@ -2358,6 +2358,7 @@ def inner(*args, **kwargs):
     "MGET NONATOMIC",
     "MOVE",
     "MSET",
+    "MSETEX",
     "MSET NONATOMIC",
     "MSETNX",
     "PFCOUNT",

redis/commands/core.py

@@ -1559,6 +1559,16 @@ def execute(self) -> ResponseT:
         return self.client.execute_command(*command)
 
 
+class DataPersistOptions(Enum):
+    # set the value for each provided key to each
+    # provided value only if all do not already exist.
+    NX = "NX"
+
+    # set the value for each provided key to each
+    # provided value only if all already exist.
+    XX = "XX"
+
+
 class BasicKeyCommands(CommandsProtocol):
     """
     Redis basic key-based commands
@@ -1987,6 +1997,10 @@ def mget(self, keys: KeysT, *args: EncodableT) -> ResponseT:
         """
         Returns a list of values ordered identically to ``keys``
 
+        ** Important ** When this method is used with Cluster clients, all keys
+                must be in the same hash slot, otherwise a RedisClusterException
+                will be raised.
+
         For more information, see https://redis.io/commands/mget
         """
         from redis.client import EMPTY_RESPONSE
@@ -2004,20 +2018,94 @@ def mset(self, mapping: Mapping[AnyKeyT, EncodableT]) -> ResponseT:
         key/value pairs. Both keys and values should be strings or types that
         can be cast to a string via str().
 
+        ** Important ** When this method is used with Cluster clients, all keys
+                must be in the same hash slot, otherwise a RedisClusterException
+                will be raised.
+
         For more information, see https://redis.io/commands/mset
         """
         items = []
         for pair in mapping.items():
             items.extend(pair)
         return self.execute_command("MSET", *items)
 
+    def msetex(
+        self,
+        mapping: Mapping[AnyKeyT, EncodableT],
+        data_persist_option: Optional[DataPersistOptions] = None,
+        ex: Optional[ExpiryT] = None,
+        px: Optional[ExpiryT] = None,
+        exat: Optional[AbsExpiryT] = None,
+        pxat: Optional[AbsExpiryT] = None,
+        keepttl: bool = False,
+    ) -> Union[Awaitable[int], int]:
+        """
+        Sets key/values based on the provided ``mapping`` items.
+
+        ** Important ** When this method is used with Cluster clients, all keys
+                        must be in the same hash slot, otherwise a RedisClusterException
+                        will be raised.
+
+        ``mapping`` accepts a dict of key/value pairs that will be added to the database.
+
+        ``data_persist_option`` can be set to ``NX`` or ``XX`` to control the
+            behavior of the command.
+            ``NX`` will set the value for each provided key to each
+                provided value only if all do not already exist.
+            ``XX`` will set the value for each provided key to each
+                provided value only if all already exist.
+
+        ``ex`` sets an expire flag on the keys in ``mapping`` for ``ex`` seconds.
+
+        ``px`` sets an expire flag on the keys in ``mapping`` for ``px`` milliseconds.
+
+        ``exat`` sets an expire flag on the keys in ``mapping`` for ``exat`` seconds,
+            specified in unix time.
+
+        ``pxat`` sets an expire flag on the keys in ``mapping`` for ``pxat`` milliseconds,
+            specified in unix time.
+
+        ``keepttl`` if True, retain the time to live associated with the keys.
+
+        Returns the number of fields that were added.
+
+        Available since Redis 8.4
+        For more information, see https://redis.io/commands/msetex
+        """
+        opset = {ex, px, exat, pxat}
+        if len(opset) > 2 or len(opset) > 1 and keepttl:
+            raise DataError(
+                "``ex``, ``px``, ``exat``, ``pxat``, "
+                "and ``keepttl`` are mutually exclusive."
+            )
+
+        exp_options: list[EncodableT] = []
+        if data_persist_option:
+            exp_options.append(data_persist_option.value)
+
+        exp_options.extend(extract_expire_flags(ex, px, exat, pxat))
+
+        if keepttl:
+            exp_options.append("KEEPTTL")
+
+        pieces = ["MSETEX", len(mapping)]
+
+        for pair in mapping.items():
+            pieces.extend(pair)
+
+        return self.execute_command(*pieces, *exp_options)
+
     def msetnx(self, mapping: Mapping[AnyKeyT, EncodableT]) -> ResponseT:
         """
         Sets key/values based on a mapping if none of the keys are already set.
         Mapping is a dictionary of key/value pairs. Both keys and values
         should be strings or types that can be cast to a string via str().
         Returns a boolean indicating if the operation was successful.
 
+        ** Important ** When this method is used with Cluster clients, all keys
+                        must be in the same hash slot, otherwise a RedisClusterException
+                        will be raised.
+
         For more information, see https://redis.io/commands/msetnx
         """
         items = []
@@ -4013,6 +4101,7 @@ def xreadgroup(
         count: Optional[int] = None,
         block: Optional[int] = None,
         noack: bool = False,
+        claim_min_idle_time: Optional[int] = None,
     ) -> ResponseT:
         """
         Read from a stream via a consumer group.
@@ -4030,8 +4119,12 @@ def xreadgroup(
         block: number of milliseconds to wait, if nothing already present.
         noack: do not add messages to the PEL
 
+        claim_min_idle_time: accepts an integer type and represents a
+                             time interval in milliseconds
+
         For more information, see https://redis.io/commands/xreadgroup
         """
+        options = {}
         pieces: list[EncodableT] = [b"GROUP", groupname, consumername]
         if count is not None:
             if not isinstance(count, int) or count < 1:
@@ -4045,12 +4138,20 @@ def xreadgroup(
             pieces.append(str(block))
         if noack:
             pieces.append(b"NOACK")
+        if claim_min_idle_time is not None:
+            if not isinstance(claim_min_idle_time, int) or claim_min_idle_time < 0:
+                raise DataError(
+                    "XREADGROUP claim_min_idle_time must be a non-negative integer"
+                )
+            pieces.append(b"CLAIM")
+            pieces.append(claim_min_idle_time)
+            options["claim_min_idle_time"] = claim_min_idle_time
         if not isinstance(streams, dict) or len(streams) == 0:
             raise DataError("XREADGROUP streams must be a non empty dict")
         pieces.append(b"STREAMS")
         pieces.extend(streams.keys())
         pieces.extend(streams.values())
-        return self.execute_command("XREADGROUP", *pieces)
+        return self.execute_command("XREADGROUP", *pieces, **options)
 
     def xrevrange(
         self,