Merge branch 'master' of https://github.com/vainglorygame/joblib

2 files changed, 15 insertions, 10 deletions

diff --git a/joblib.py b/joblib.py
index 6891e11..a0ce985 100644
--- a/joblib.py
+++ b/joblib.py
@@ -36,29 +36,29 @@ class JobQueue(object):
             """, jobtype, json.dumps(payload), priority)
 
     async def acquire(self, jobtype):
-        """Mark a job as running, return payload and return id.
-        Return (None, None) if no job is available."""
+        """Mark a job as running, return id, payload and priority.
+        Return (None, None, None) if no job is available."""
         async with self._pool.acquire() as con:
             while True:
                 try:
                     # do not allow async access
                     async with con.transaction(isolation="serializable"):
                         result = await con.fetchrow("""
-                            SELECT id, payload
+                            SELECT id, payload, priority
                             FROM jobs WHERE
                             type=$1 AND status='open'
-                            ORDER BY priority ASC
+                            ORDER BY priority DESC
                         """, jobtype)
                         if result is None:
                             # no jobs available
-                            return None, None
-                        jobid, payload = result
+                            return None, None, None
+                        jobid, payload, priority = result
                         await con.execute("""
                             UPDATE jobs
                             SET status='running'
                             WHERE id=$1
                         """, jobid)
-                        return jobid, json.loads(payload)
+                        return jobid, json.loads(payload), priority
                 except asyncpg.exceptions.SerializationError:
                     # job is being picked up by another worker, try again
                     pass
diff --git a/test_joblib.py b/test_joblib.py
index 35ae90e..47eedec 100644
--- a/test_joblib.py
+++ b/test_joblib.py
@@ -51,17 +51,22 @@ class TestJoblib:
     async def test_cleanup(self, queue, payload):
         await queue.request(jobtype="testing", payload=payload)
         # mark job as processing
-        jobid_1, payload_1 = await queue.acquire(jobtype="testing")
+        jobid_1, payload_1, _ = await queue.acquire(jobtype="testing")
         assert payload_1 == payload
         await queue.cleanup()
         # same job should be available again
-        jobid_2, payload_2 = await queue.acquire(jobtype="testing")
+        jobid_2, payload_2, _ = await queue.acquire(jobtype="testing")
         assert jobid_1 == jobid_2 and payload_1 == payload_2
 
     @pytest.mark.asyncio
+    async def test_priority(self, queue, payload):
+        await queue.request(jobtype="testing", payload=payload, priority=9)
+        assert 9 == (await queue.acquire(jobtype="testing"))[2]
+
+    @pytest.mark.asyncio
     async def test_finish(self, queue, payload):
         await queue.request(jobtype="testing", payload=payload)
-        jobid, _ = await queue.acquire(jobtype="testing")
+        jobid, _, _ = await queue.acquire(jobtype="testing")
         await queue.finish(jobid)
         # job should not be available again
         assert None == (await queue.acquire(jobtype="testing"))[1]