Merge pull request #151 from EverythingMe/feature_data_source

Fix issue with serializing unicode queries.

redash/data/manager.py

@@ -78,6 +78,8 @@ class Manager(object):
     def refresh_queries(self):
         # TODO: this will only execute scheduled queries that were executed before. I think this is
         # a reasonable assumption, but worth revisiting.
+
+        # TODO: move this logic to the model.
         outdated_queries = models.Query.select(peewee.Func('first_value', models.Query.id)\
             .over(partition_by=[models.Query.query_hash, models.Query.data_source]))\
             .join(models.QueryResult)\
@@ -116,6 +118,7 @@ class Manager(object):
 
         logging.info("[Manager][%s] Inserted query data; id=%s", query_hash, query_result.id)
 
+        # TODO: move this logic to the model?
         updated_count = models.Query.update(latest_query_data=query_result).\
             where(models.Query.query_hash==query_hash, models.Query.data_source==data_source_id).\
             execute()

redash/data/worker.py

@@ -80,6 +80,13 @@ class RedisObject(object):
         return obj
 
 
+def fix_unicode(string):
+    if isinstance(string, unicode):
+        return string
+
+    return string.decode('utf-8')
+
+
 class Job(RedisObject):
     HIGH_PRIORITY = 1
     LOW_PRIORITY = 2
@@ -108,7 +115,7 @@ class Job(RedisObject):
     }
 
     conversions = {
-        'query': lambda query: query.decode('utf-8'),
+        'query': fix_unicode,
         'priority': int,
         'updated_at': float,
         'status': int,
@@ -121,7 +128,7 @@ class Job(RedisObject):
     name = 'job'
 
     def __init__(self, redis_connection, query, priority, **kwargs):
-        kwargs['query'] = query
+        kwargs['query'] = fix_unicode(query)
         kwargs['priority'] = priority
         kwargs['query_hash'] = gen_query_hash(kwargs['query'])
         self.new_job = 'id' not in kwargs

tests/test_job.py

@@ -1,3 +1,5 @@
+# coding=utf-8
+
 import time
 from unittest import TestCase
 from mock import patch
@@ -86,5 +88,13 @@ class TestJob(TestCase):
         self.assertEquals(now - updated_at, job.query_time)
         self.assertIsNone(job.error)
 
-    def test_done_failed(self):
-        pass
+    def test_unicode_serialization(self):
+        unicode_query = u"יוניקוד"
+        job = Job(redis_connection, query=unicode_query, priority=self.priority)
+
+        self.assertEquals(job.query, unicode_query)
+
+        job.save()
+        loaded_job = Job.load(redis_connection, job.id)
+        self.assertEquals(loaded_job.query, unicode_query)
+