Forced job_id to only allow str/unicode

rq/job.py

@@ -2,6 +2,7 @@
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
 
+import types
 import inspect
 import warnings
 from functools import partial
@@ -107,6 +108,8 @@ class Job(object):
             raise TypeError('{0!r} is not a valid args list.'.format(args))
         if not isinstance(kwargs, dict):
             raise TypeError('{0!r} is not a valid kwargs dict.'.format(kwargs))
+        if not isinstance(job_id, (str, unicode, types.NoneType)):
+            raise TypeError('job_id must be a str/unicode, not {}.'.format(type(job_id)))
 
         job = cls(connection=connection)
         if job_id is not None:
@@ -329,11 +332,7 @@ class Job(object):
 
     def set_id(self, value):
         """Sets a job ID for the given job."""
-        try:
-            self.key_for(text_type(value))
-        except:
-            raise ValueError("Job ID invalid, failed to encode to string")
-        self._id = text_type(value)
+        self._id = value
 
     id = property(get_id, set_id)
 

tests/test_job.py

@@ -342,10 +342,10 @@ class TestJob(RQTestCase):
         self.assertNotIn(job.id, queue.get_job_ids())
 
     def test_create_job_with_id(self):
-        # try a bunch of different ID types
+        """test creating jobs with a custom ID"""
         queue = Queue(connection=self.testconn)
-        ids = [1234, uuid4(), "somejobid"]
-        for job_id in ids:
-            job = queue.enqueue(say_hello, job_id=job_id)
-            self.assertEqual(job.id, str(job_id))
-            job.perform()
+        job = queue.enqueue(say_hello, job_id="1234")
+        self.assertEqual(job.id, "1234")
+        job.perform()
+        
+        self.assertRaises(TypeError, queue.enqueue, say_hello, job_id=1234)