use pipeline rather than gather

python-arq · Apr 17, 2019 · 5d67dc4 · 5d67dc4
1 parent 2105f14
commit 5d67dc4
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 4 deletions.
diff --git a/HISTORY.rst b/HISTORY.rst
@@ -6,6 +6,7 @@ History
 v0.16.0a4 (unreleased)
 ......................
 * add ``Worker.run_check``, fix #115
+* use ``pipeline`` in ``enqueue_job``
 
 v0.16.0a3 (2010-03-12)
 ......................

diff --git a/Makefile b/Makefile
@@ -8,7 +8,7 @@ install:
 	pip install -r requirements.txt
 	pip install -e .[watch]
 
-.PHONY: isort
+.PHONY: format
 format:
 	$(isort)
 	$(black)

diff --git a/arq/connections.py b/arq/connections.py
@@ -78,8 +78,12 @@ async def enqueue_job(
         expires_ms = to_ms(_expires)
 
         with await self as conn:
-            _, _, job_exists = await asyncio.gather(conn.unwatch(), conn.watch(job_key), conn.exists(job_key))
-            if job_exists:
+            pipe = conn.pipeline()
+            pipe.unwatch()
+            pipe.watch(job_key)
+            job_exists = pipe.exists(job_key)
+            await pipe.execute()
+            if await job_exists:
                 return
 
             enqueue_time_ms = timestamp_ms()
@@ -99,7 +103,7 @@ async def enqueue_job(
             try:
                 await tr.execute()
             except MultiExecError:
-                # job got enqueued since we got 'job_exists'
+                # job got enqueued since we checked 'job_exists'
                 return
         return Job(job_id, self)