Replace ensure_future with create_task (#2311)

* Replace `ensure_future` with `create_task`

* linters

Author: dvora-h

redis/asyncio/cluster.py

@@ -642,7 +642,7 @@ async def execute_command(self, *args: EncodableT, **kwargs: Any) -> Any:
                     keys = [node.name for node in target_nodes]
                     values = await asyncio.gather(
                         *(
-                            asyncio.ensure_future(
+                            asyncio.create_task(
                                 self._execute_command(node, *args, **kwargs)
                             )
                             for node in target_nodes
@@ -841,7 +841,7 @@ def __del__(self) -> None:
     async def disconnect(self) -> None:
         ret = await asyncio.gather(
             *(
-                asyncio.ensure_future(connection.disconnect())
+                asyncio.create_task(connection.disconnect())
                 for connection in self._connections
             ),
             return_exceptions=True,
@@ -979,13 +979,13 @@ def set_nodes(
         if remove_old:
             for name in list(old.keys()):
                 if name not in new:
-                    asyncio.ensure_future(old.pop(name).disconnect())
+                    asyncio.create_task(old.pop(name).disconnect())
 
         for name, node in new.items():
             if name in old:
                 if old[name] is node:
                     continue
-                asyncio.ensure_future(old[name].disconnect())
+                asyncio.create_task(old[name].disconnect())
             old[name] = node
 
     def _update_moved_slots(self) -> None:
@@ -1202,7 +1202,7 @@ async def close(self, attr: str = "nodes_cache") -> None:
         self.default_node = None
         await asyncio.gather(
             *(
-                asyncio.ensure_future(node.disconnect())
+                asyncio.create_task(node.disconnect())
                 for node in getattr(self, attr).values()
             )
         )
@@ -1381,7 +1381,7 @@ async def _execute(
 
         errors = await asyncio.gather(
             *(
-                asyncio.ensure_future(node[0].execute_pipeline(node[1]))
+                asyncio.create_task(node[0].execute_pipeline(node[1]))
                 for node in nodes.values()
             )
         )

redis/commands/cluster.py

@@ -673,7 +673,7 @@ async def cluster_delslots(self, *slots: EncodableT) -> List[bool]:
         """
         return await asyncio.gather(
             *(
-                asyncio.ensure_future(self.execute_command("CLUSTER DELSLOTS", slot))
+                asyncio.create_task(self.execute_command("CLUSTER DELSLOTS", slot))
                 for slot in slots
             )
         )

tests/test_asyncio/compat.py

@@ -1,5 +1,4 @@
 import asyncio
-import sys
 from unittest import mock
 
 try:
@@ -9,7 +8,4 @@
 
 
 def create_task(coroutine):
-    if sys.version_info[:2] >= (3, 7):
-        return asyncio.create_task(coroutine)
-    else:
-        return asyncio.ensure_future(coroutine)
+    return asyncio.create_task(coroutine)