From 119362ab2b0c3f00bb80b927641f478a847ac1d9 Mon Sep 17 00:00:00 2001
From: schneefux <schneefux+commit@schneefux.xyz>
Date: Sun, 26 Feb 2017 18:31:14 +0100
Subject: add job fails

---
 joblib.py      | 15 +++++++++++++++
 test_joblib.py | 14 ++++++++++++++
 2 files changed, 29 insertions(+)

diff --git a/joblib.py b/joblib.py
index a0ce985..48666a6 100644
--- a/joblib.py
+++ b/joblib.py
@@ -78,6 +78,21 @@ class JobQueue(object):
                 except asyncpg.exceptions.SerializationError:
                     pass
 
+    async def fail(self, jobid, reason):
+        """Mark a job as failed."""
+        async with self._pool.acquire() as con:
+            while True:
+                try:
+                    async with con.transaction(isolation="serializable"):
+                        await con.execute("""
+                            UPDATE jobs
+                            SET status='failed', payload=payload||$2::jsonb
+                            WHERE id=$1
+                        """, jobid, json.dumps({"error": reason}))
+                        return
+                except asyncpg.exceptions.SerializationError:
+                    pass
+
     async def cleanup(self):
         """Reopen all unfinished jobs."""
         async with self._pool.acquire() as con:
diff --git a/test_joblib.py b/test_joblib.py
index 47eedec..f9aa875 100644
--- a/test_joblib.py
+++ b/test_joblib.py
@@ -2,6 +2,7 @@
 
 import os
 import asyncio
+import json
 import asyncpg
 import pytest
 import joblib
@@ -63,6 +64,19 @@ class TestJoblib:
         await queue.request(jobtype="testing", payload=payload, priority=9)
         assert 9 == (await queue.acquire(jobtype="testing"))[2]
 
+    @pytest.mark.asyncio
+    async def test_fail(self, queue, payload):
+        err = "testing errors"
+        await queue.request(jobtype="testing", payload=payload)
+        jobid, _, _ = await queue.acquire(jobtype="testing")
+        await queue.fail(jobid, err)
+        async with queue._pool.acquire() as con:
+            jid, pl = await con.fetchrow(
+                "SELECT id, payload FROM jobs WHERE status='failed'")
+            assert jid == jobid
+            payload["error"] = err
+            assert json.loads(pl) == payload
+
     @pytest.mark.asyncio
     async def test_finish(self, queue, payload):
         await queue.request(jobtype="testing", payload=payload)
-- 
cgit v1.3.1