Add support for MODULE LOADEX (#2146)

redis/commands/core.py

@@ -5592,6 +5592,27 @@ def module_load(self, path, *args) -> ResponseT:
         """
         return self.execute_command("MODULE LOAD", path, *args)
 
+    def module_loadex(
+        self,
+        path: str,
+        options: Optional[List[str]] = None,
+        args: Optional[List[str]] = None,
+    ) -> ResponseT:
+        """
+        Loads a module from a dynamic library at runtime with configuration directives.
+
+        For more information see https://redis.io/commands/module-loadex
+        """
+        pieces = []
+        if options is not None:
+            pieces.append("CONFIG")
+            pieces.extend(options)
+        if args is not None:
+            pieces.append("ARGS")
+            pieces.extend(args)
+
+        return self.execute_command("MODULE LOADEX", path, *pieces)
+
     def module_unload(self, name) -> ResponseT:
         """
         Unloads the module ``name``.

tests/test_commands.py

@@ -4513,6 +4513,18 @@ def test_module(self, r):
             r.module_load("/some/fake/path", "arg1", "arg2", "arg3", "arg4")
             assert "Error loading the extension." in str(excinfo.value)
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("7.0.0")
+    @skip_if_redis_enterprise()
+    def test_module_loadex(self, r: redis.Redis):
+        with pytest.raises(redis.exceptions.ModuleError) as excinfo:
+            r.module_loadex("/some/fake/path")
+            assert "Error loading the extension." in str(excinfo.value)
+
+        with pytest.raises(redis.exceptions.ModuleError) as excinfo:
+            r.module_loadex("/some/fake/path", ["name", "value"], ["arg1", "arg2"])
+            assert "Error loading the extension." in str(excinfo.value)
+
     @skip_if_server_version_lt("2.6.0")
     def test_restore(self, r):