We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Query execution fails with Google Apps accounts having non-ASCII characters in name
If the account used for Google Apps authentication has non-ASCII characters (ä in this case), execution of queries fails against Amazon Redshift datasource.

[2015-06-05 11:36:13,520][PID:7502][INFO][root] [Manager] Metadata: [{'Username': u'Raimo J\xe4rvenp\xe4\xe4', 'Query ID': 25}]
[2015-06-05 11:36:13,520][PID:7502][INFO][root] [Manager][a2a8974a74cfcb958afc2f843f9f896b] Found existing job: ed40d887-8dae-4e2d-a3a4-439bfb4f01a6
[2015-06-05 11:36:13,521][PID:7502][INFO][root] [a2a8974a74cfcb958afc2f843f9f896b] job found is ready (FAILURE), removing lock


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `redash/tasks.py`
Content:
```
1 import time
2 import datetime
3 import logging
4 import redis
5 from celery import Task
6 from celery.result import AsyncResult
7 from celery.utils.log import get_task_logger
8 from redash import redis_connection, models, statsd_client, settings
9 from redash.utils import gen_query_hash
10 from redash.worker import celery
11 from redash.query_runner import get_query_runner
12 
13 logger = get_task_logger(__name__)
14 
15 
16 class BaseTask(Task):
17     abstract = True
18 
19     def after_return(self, *args, **kwargs):
20         models.db.close_db(None)
21 
22     def __call__(self, *args, **kwargs):
23         models.db.connect_db()
24         return super(BaseTask, self).__call__(*args, **kwargs)
25 
26 
27 class QueryTask(object):
28     MAX_RETRIES = 5
29 
30     # TODO: this is mapping to the old Job class statuses. Need to update the client side and remove this
31     STATUSES = {
32         'PENDING': 1,
33         'STARTED': 2,
34         'SUCCESS': 3,
35         'FAILURE': 4,
36         'REVOKED': 4
37     }
38 
39     def __init__(self, job_id=None, async_result=None):
40         if async_result:
41             self._async_result = async_result
42         else:
43             self._async_result = AsyncResult(job_id, app=celery)
44 
45     @property
46     def id(self):
47         return self._async_result.id
48 
49     @classmethod
50     def add_task(cls, query, data_source, scheduled=False, metadata={}):
51         query_hash = gen_query_hash(query)
52         logging.info("[Manager][%s] Inserting job", query_hash)
53         logging.info("[Manager] Metadata: [%s]", metadata)
54         try_count = 0
55         job = None
56         
57         while try_count < cls.MAX_RETRIES:
58             try_count += 1
59 
60             pipe = redis_connection.pipeline()
61             try:
62                 pipe.watch(cls._job_lock_id(query_hash, data_source.id))
63                 job_id = pipe.get(cls._job_lock_id(query_hash, data_source.id))
64                 if job_id:
65                     logging.info("[Manager][%s] Found existing job: %s", query_hash, job_id)
66 
67                     job = cls(job_id=job_id)
68                     if job.ready():
69                         logging.info("[%s] job found is ready (%s), removing lock", query_hash, job.celery_status)
70                         redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))
71                         job = None
72 
73                 if not job:
74                     pipe.multi()
75 
76                     if scheduled:
77                         queue_name = data_source.scheduled_queue_name
78                     else:
79                         queue_name = data_source.queue_name
80 
81                     result = execute_query.apply_async(args=(query, data_source.id, metadata), queue=queue_name)
82                     job = cls(async_result=result)
83                     
84                     logging.info("[Manager][%s] Created new job: %s", query_hash, job.id)
85                     pipe.set(cls._job_lock_id(query_hash, data_source.id), job.id, settings.JOB_EXPIRY_TIME)
86                     pipe.execute()
87                 break
88 
89             except redis.WatchError:
90                 continue
91 
92         if not job:
93             logging.error("[Manager][%s] Failed adding job for query.", query_hash)
94 
95         return job
96 
97     def to_dict(self):
98         if self._async_result.status == 'STARTED':
99             updated_at = self._async_result.result.get('start_time', 0)
100         else:
101             updated_at = 0
102 
103         if self._async_result.failed() and isinstance(self._async_result.result, Exception):
104             error = self._async_result.result.message
105         elif self._async_result.status == 'REVOKED':
106             error = 'Query execution cancelled.'
107         else:
108             error = ''
109 
110         if self._async_result.successful():
111             query_result_id = self._async_result.result
112         else:
113             query_result_id = None
114 
115         return {
116             'id': self._async_result.id,
117             'updated_at': updated_at,
118             'status': self.STATUSES[self._async_result.status],
119             'error': error,
120             'query_result_id': query_result_id,
121         }
122 
123     @property
124     def is_cancelled(self):
125         return self._async_result.status == 'REVOKED'
126 
127     @property
128     def celery_status(self):
129         return self._async_result.status
130 
131     def ready(self):
132         return self._async_result.ready()
133 
134     def cancel(self):
135         return self._async_result.revoke(terminate=True)
136 
137     @staticmethod
138     def _job_lock_id(query_hash, data_source_id):
139         return "query_hash_job:%s:%s" % (data_source_id, query_hash)
140 
141 
142 @celery.task(base=BaseTask)
143 def refresh_queries():
144     # self.status['last_refresh_at'] = time.time()
145     # self._save_status()
146 
147     logger.info("Refreshing queries...")
148 
149     outdated_queries_count = 0
150     for query in models.Query.outdated_queries():
151         QueryTask.add_task(query.query, query.data_source, scheduled=True,
152                            metadata={'Query ID': query.id, 'Username': 'Scheduled'})
153         outdated_queries_count += 1
154 
155     statsd_client.gauge('manager.outdated_queries', outdated_queries_count)
156 
157     logger.info("Done refreshing queries. Found %d outdated queries." % outdated_queries_count)
158 
159     status = redis_connection.hgetall('redash:status')
160     now = time.time()
161 
162     redis_connection.hmset('redash:status', {
163         'outdated_queries_count': outdated_queries_count,
164         'last_refresh_at': now
165     })
166 
167     statsd_client.gauge('manager.seconds_since_refresh', now - float(status.get('last_refresh_at', now)))
168 
169 
170 @celery.task(base=BaseTask)
171 def cleanup_tasks():
172     # in case of cold restart of the workers, there might be jobs that still have their "lock" object, but aren't really
173     # going to run. this job removes them.
174     lock_keys = redis_connection.keys("query_hash_job:*") # TODO: use set instead of keys command
175     if not lock_keys:
176         return
177     
178     query_tasks = [QueryTask(job_id=j) for j in redis_connection.mget(lock_keys)]
179 
180     logger.info("Found %d locks", len(query_tasks))
181 
182     inspect = celery.control.inspect()
183     active_tasks = inspect.active()
184     if active_tasks is None:
185         active_tasks = []
186     else:
187         active_tasks = active_tasks.values()
188 
189     all_tasks = set()
190     for task_list in active_tasks:
191         for task in task_list:
192             all_tasks.add(task['id'])
193 
194     logger.info("Active jobs count: %d", len(all_tasks))
195 
196     for i, t in enumerate(query_tasks):
197         if t.ready():
198             # if locked task is ready already (failed, finished, revoked), we don't need the lock anymore
199             logger.warning("%s is ready (%s), removing lock.", lock_keys[i], t.celery_status)
200             redis_connection.delete(lock_keys[i])
201 
202         # if t.celery_status == 'STARTED' and t.id not in all_tasks:
203         #     logger.warning("Couldn't find active job for: %s, removing lock.", lock_keys[i])
204         #     redis_connection.delete(lock_keys[i])
205 
206 
207 @celery.task(base=BaseTask)
208 def cleanup_query_results():
209     """
210     Job to cleanup unused query results -- such that no query links to them anymore, and older than a week (so it's less
211     likely to be open in someone's browser and be used).
212 
213     Each time the job deletes only 100 query results so it won't choke the database in case of many such results.
214     """
215 
216     unused_query_results = models.QueryResult.unused().limit(100)
217     total_unused_query_results = models.QueryResult.unused().count()
218     deleted_count = models.QueryResult.delete().where(models.QueryResult.id << unused_query_results).execute()
219 
220     logger.info("Deleted %d unused query results out of total of %d." % (deleted_count, total_unused_query_results))
221 
222 
223 @celery.task(base=BaseTask)
224 def refresh_schemas():
225     """
226     Refershs the datasources schema.
227     """
228 
229     for ds in models.DataSource.all():
230         logger.info("Refreshing schema for: {}".format(ds.name))
231         ds.get_schema(refresh=True)
232 
233 
234 @celery.task(bind=True, base=BaseTask, track_started=True)
235 def execute_query(self, query, data_source_id, metadata):
236     start_time = time.time()
237 
238     logger.info("Loading data source (%d)...", data_source_id)
239 
240     # TODO: we should probably cache data sources in Redis
241     data_source = models.DataSource.get_by_id(data_source_id)
242 
243     self.update_state(state='STARTED', meta={'start_time': start_time, 'custom_message': ''})
244 
245     logger.info("Executing query:\n%s", query)
246 
247     query_hash = gen_query_hash(query)
248     query_runner = get_query_runner(data_source.type, data_source.options)
249 
250     if query_runner.annotate_query():
251         metadata['Task ID'] = self.request.id
252         metadata['Query Hash'] = query_hash
253         metadata['Queue'] = self.request.delivery_info['routing_key']
254 
255         annotation = ", ".join(["{}: {}".format(k, v) for k, v in metadata.iteritems()])
256 
257         logging.debug("Annotation: %s", annotation)
258 
259         annotated_query = u"/* {} */ {}".format(annotation, query)
260     else:
261         annotated_query = query
262 
263     with statsd_client.timer('query_runner.{}.{}.run_time'.format(data_source.type, data_source.name)):
264         data, error = query_runner.run_query(annotated_query)
265 
266     run_time = time.time() - start_time
267     logger.info("Query finished... data length=%s, error=%s", data and len(data), error)
268 
269     self.update_state(state='STARTED', meta={'start_time': start_time, 'error': error, 'custom_message': ''})
270 
271     # Delete query_hash
272     redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))
273 
274     if not error:
275         query_result = models.QueryResult.store_result(data_source.id, query_hash, query, data, run_time, datetime.datetime.utcnow())
276     else:
277         raise Exception(error)
278 
279     return query_result.id
280 
281 
282 @celery.task(base=BaseTask)
283 def record_event(event):
284     models.Event.record(event)
285
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/redash/tasks.py b/redash/tasks.py
--- a/redash/tasks.py
+++ b/redash/tasks.py
@@ -252,9 +252,9 @@
         metadata['Query Hash'] = query_hash
         metadata['Queue'] = self.request.delivery_info['routing_key']
 
-        annotation = ", ".join(["{}: {}".format(k, v) for k, v in metadata.iteritems()])
+        annotation = u", ".join([u"{}: {}".format(k, v) for k, v in metadata.iteritems()])
 
-        logging.debug("Annotation: %s", annotation)
+        logging.debug(u"Annotation: %s", annotation)
 
         annotated_query = u"/* {} */ {}".format(annotation, query)
     else:

{"golden_diff": "diff --git a/redash/tasks.py b/redash/tasks.py\n--- a/redash/tasks.py\n+++ b/redash/tasks.py\n@@ -252,9 +252,9 @@\n         metadata['Query Hash'] = query_hash\n         metadata['Queue'] = self.request.delivery_info['routing_key']\n \n-        annotation = \", \".join([\"{}: {}\".format(k, v) for k, v in metadata.iteritems()])\n+        annotation = u\", \".join([u\"{}: {}\".format(k, v) for k, v in metadata.iteritems()])\n \n-        logging.debug(\"Annotation: %s\", annotation)\n+        logging.debug(u\"Annotation: %s\", annotation)\n \n         annotated_query = u\"/* {} */ {}\".format(annotation, query)\n     else:\n", "issue": "Query execution fails with Google Apps accounts having non-ASCII characters in name\nIf the account used for Google Apps authentication has non-ASCII characters (\u00e4 in this case), execution of queries fails against Amazon Redshift datasource.\n\n[2015-06-05 11:36:13,520][PID:7502][INFO][root] [Manager] Metadata: [{'Username': u'Raimo J\\xe4rvenp\\xe4\\xe4', 'Query ID': 25}]\n[2015-06-05 11:36:13,520][PID:7502][INFO][root] [Manager][a2a8974a74cfcb958afc2f843f9f896b] Found existing job: ed40d887-8dae-4e2d-a3a4-439bfb4f01a6\n[2015-06-05 11:36:13,521][PID:7502][INFO][root] [a2a8974a74cfcb958afc2f843f9f896b] job found is ready (FAILURE), removing lock\n\n", "before_files": [{"content": "import time\nimport datetime\nimport logging\nimport redis\nfrom celery import Task\nfrom celery.result import AsyncResult\nfrom celery.utils.log import get_task_logger\nfrom redash import redis_connection, models, statsd_client, settings\nfrom redash.utils import gen_query_hash\nfrom redash.worker import celery\nfrom redash.query_runner import get_query_runner\n\nlogger = get_task_logger(__name__)\n\n\nclass BaseTask(Task):\n    abstract = True\n\n    def after_return(self, *args, **kwargs):\n        models.db.close_db(None)\n\n    def __call__(self, *args, **kwargs):\n        models.db.connect_db()\n        return super(BaseTask, self).__call__(*args, **kwargs)\n\n\nclass QueryTask(object):\n    MAX_RETRIES = 5\n\n    # TODO: this is mapping to the old Job class statuses. Need to update the client side and remove this\n    STATUSES = {\n        'PENDING': 1,\n        'STARTED': 2,\n        'SUCCESS': 3,\n        'FAILURE': 4,\n        'REVOKED': 4\n    }\n\n    def __init__(self, job_id=None, async_result=None):\n        if async_result:\n            self._async_result = async_result\n        else:\n            self._async_result = AsyncResult(job_id, app=celery)\n\n    @property\n    def id(self):\n        return self._async_result.id\n\n    @classmethod\n    def add_task(cls, query, data_source, scheduled=False, metadata={}):\n        query_hash = gen_query_hash(query)\n        logging.info(\"[Manager][%s] Inserting job\", query_hash)\n        logging.info(\"[Manager] Metadata: [%s]\", metadata)\n        try_count = 0\n        job = None\n        \n        while try_count < cls.MAX_RETRIES:\n            try_count += 1\n\n            pipe = redis_connection.pipeline()\n            try:\n                pipe.watch(cls._job_lock_id(query_hash, data_source.id))\n                job_id = pipe.get(cls._job_lock_id(query_hash, data_source.id))\n                if job_id:\n                    logging.info(\"[Manager][%s] Found existing job: %s\", query_hash, job_id)\n\n                    job = cls(job_id=job_id)\n                    if job.ready():\n                        logging.info(\"[%s] job found is ready (%s), removing lock\", query_hash, job.celery_status)\n                        redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))\n                        job = None\n\n                if not job:\n                    pipe.multi()\n\n                    if scheduled:\n                        queue_name = data_source.scheduled_queue_name\n                    else:\n                        queue_name = data_source.queue_name\n\n                    result = execute_query.apply_async(args=(query, data_source.id, metadata), queue=queue_name)\n                    job = cls(async_result=result)\n                    \n                    logging.info(\"[Manager][%s] Created new job: %s\", query_hash, job.id)\n                    pipe.set(cls._job_lock_id(query_hash, data_source.id), job.id, settings.JOB_EXPIRY_TIME)\n                    pipe.execute()\n                break\n\n            except redis.WatchError:\n                continue\n\n        if not job:\n            logging.error(\"[Manager][%s] Failed adding job for query.\", query_hash)\n\n        return job\n\n    def to_dict(self):\n        if self._async_result.status == 'STARTED':\n            updated_at = self._async_result.result.get('start_time', 0)\n        else:\n            updated_at = 0\n\n        if self._async_result.failed() and isinstance(self._async_result.result, Exception):\n            error = self._async_result.result.message\n        elif self._async_result.status == 'REVOKED':\n            error = 'Query execution cancelled.'\n        else:\n            error = ''\n\n        if self._async_result.successful():\n            query_result_id = self._async_result.result\n        else:\n            query_result_id = None\n\n        return {\n            'id': self._async_result.id,\n            'updated_at': updated_at,\n            'status': self.STATUSES[self._async_result.status],\n            'error': error,\n            'query_result_id': query_result_id,\n        }\n\n    @property\n    def is_cancelled(self):\n        return self._async_result.status == 'REVOKED'\n\n    @property\n    def celery_status(self):\n        return self._async_result.status\n\n    def ready(self):\n        return self._async_result.ready()\n\n    def cancel(self):\n        return self._async_result.revoke(terminate=True)\n\n    @staticmethod\n    def _job_lock_id(query_hash, data_source_id):\n        return \"query_hash_job:%s:%s\" % (data_source_id, query_hash)\n\n\n@celery.task(base=BaseTask)\ndef refresh_queries():\n    # self.status['last_refresh_at'] = time.time()\n    # self._save_status()\n\n    logger.info(\"Refreshing queries...\")\n\n    outdated_queries_count = 0\n    for query in models.Query.outdated_queries():\n        QueryTask.add_task(query.query, query.data_source, scheduled=True,\n                           metadata={'Query ID': query.id, 'Username': 'Scheduled'})\n        outdated_queries_count += 1\n\n    statsd_client.gauge('manager.outdated_queries', outdated_queries_count)\n\n    logger.info(\"Done refreshing queries. Found %d outdated queries.\" % outdated_queries_count)\n\n    status = redis_connection.hgetall('redash:status')\n    now = time.time()\n\n    redis_connection.hmset('redash:status', {\n        'outdated_queries_count': outdated_queries_count,\n        'last_refresh_at': now\n    })\n\n    statsd_client.gauge('manager.seconds_since_refresh', now - float(status.get('last_refresh_at', now)))\n\n\n@celery.task(base=BaseTask)\ndef cleanup_tasks():\n    # in case of cold restart of the workers, there might be jobs that still have their \"lock\" object, but aren't really\n    # going to run. this job removes them.\n    lock_keys = redis_connection.keys(\"query_hash_job:*\") # TODO: use set instead of keys command\n    if not lock_keys:\n        return\n    \n    query_tasks = [QueryTask(job_id=j) for j in redis_connection.mget(lock_keys)]\n\n    logger.info(\"Found %d locks\", len(query_tasks))\n\n    inspect = celery.control.inspect()\n    active_tasks = inspect.active()\n    if active_tasks is None:\n        active_tasks = []\n    else:\n        active_tasks = active_tasks.values()\n\n    all_tasks = set()\n    for task_list in active_tasks:\n        for task in task_list:\n            all_tasks.add(task['id'])\n\n    logger.info(\"Active jobs count: %d\", len(all_tasks))\n\n    for i, t in enumerate(query_tasks):\n        if t.ready():\n            # if locked task is ready already (failed, finished, revoked), we don't need the lock anymore\n            logger.warning(\"%s is ready (%s), removing lock.\", lock_keys[i], t.celery_status)\n            redis_connection.delete(lock_keys[i])\n\n        # if t.celery_status == 'STARTED' and t.id not in all_tasks:\n        #     logger.warning(\"Couldn't find active job for: %s, removing lock.\", lock_keys[i])\n        #     redis_connection.delete(lock_keys[i])\n\n\n@celery.task(base=BaseTask)\ndef cleanup_query_results():\n    \"\"\"\n    Job to cleanup unused query results -- such that no query links to them anymore, and older than a week (so it's less\n    likely to be open in someone's browser and be used).\n\n    Each time the job deletes only 100 query results so it won't choke the database in case of many such results.\n    \"\"\"\n\n    unused_query_results = models.QueryResult.unused().limit(100)\n    total_unused_query_results = models.QueryResult.unused().count()\n    deleted_count = models.QueryResult.delete().where(models.QueryResult.id << unused_query_results).execute()\n\n    logger.info(\"Deleted %d unused query results out of total of %d.\" % (deleted_count, total_unused_query_results))\n\n\n@celery.task(base=BaseTask)\ndef refresh_schemas():\n    \"\"\"\n    Refershs the datasources schema.\n    \"\"\"\n\n    for ds in models.DataSource.all():\n        logger.info(\"Refreshing schema for: {}\".format(ds.name))\n        ds.get_schema(refresh=True)\n\n\n@celery.task(bind=True, base=BaseTask, track_started=True)\ndef execute_query(self, query, data_source_id, metadata):\n    start_time = time.time()\n\n    logger.info(\"Loading data source (%d)...\", data_source_id)\n\n    # TODO: we should probably cache data sources in Redis\n    data_source = models.DataSource.get_by_id(data_source_id)\n\n    self.update_state(state='STARTED', meta={'start_time': start_time, 'custom_message': ''})\n\n    logger.info(\"Executing query:\\n%s\", query)\n\n    query_hash = gen_query_hash(query)\n    query_runner = get_query_runner(data_source.type, data_source.options)\n\n    if query_runner.annotate_query():\n        metadata['Task ID'] = self.request.id\n        metadata['Query Hash'] = query_hash\n        metadata['Queue'] = self.request.delivery_info['routing_key']\n\n        annotation = \", \".join([\"{}: {}\".format(k, v) for k, v in metadata.iteritems()])\n\n        logging.debug(\"Annotation: %s\", annotation)\n\n        annotated_query = u\"/* {} */ {}\".format(annotation, query)\n    else:\n        annotated_query = query\n\n    with statsd_client.timer('query_runner.{}.{}.run_time'.format(data_source.type, data_source.name)):\n        data, error = query_runner.run_query(annotated_query)\n\n    run_time = time.time() - start_time\n    logger.info(\"Query finished... data length=%s, error=%s\", data and len(data), error)\n\n    self.update_state(state='STARTED', meta={'start_time': start_time, 'error': error, 'custom_message': ''})\n\n    # Delete query_hash\n    redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))\n\n    if not error:\n        query_result = models.QueryResult.store_result(data_source.id, query_hash, query, data, run_time, datetime.datetime.utcnow())\n    else:\n        raise Exception(error)\n\n    return query_result.id\n\n\n@celery.task(base=BaseTask)\ndef record_event(event):\n    models.Event.record(event)\n", "path": "redash/tasks.py"}], "after_files": [{"content": "import time\nimport datetime\nimport logging\nimport redis\nfrom celery import Task\nfrom celery.result import AsyncResult\nfrom celery.utils.log import get_task_logger\nfrom redash import redis_connection, models, statsd_client, settings\nfrom redash.utils import gen_query_hash\nfrom redash.worker import celery\nfrom redash.query_runner import get_query_runner\n\nlogger = get_task_logger(__name__)\n\n\nclass BaseTask(Task):\n    abstract = True\n\n    def after_return(self, *args, **kwargs):\n        models.db.close_db(None)\n\n    def __call__(self, *args, **kwargs):\n        models.db.connect_db()\n        return super(BaseTask, self).__call__(*args, **kwargs)\n\n\nclass QueryTask(object):\n    MAX_RETRIES = 5\n\n    # TODO: this is mapping to the old Job class statuses. Need to update the client side and remove this\n    STATUSES = {\n        'PENDING': 1,\n        'STARTED': 2,\n        'SUCCESS': 3,\n        'FAILURE': 4,\n        'REVOKED': 4\n    }\n\n    def __init__(self, job_id=None, async_result=None):\n        if async_result:\n            self._async_result = async_result\n        else:\n            self._async_result = AsyncResult(job_id, app=celery)\n\n    @property\n    def id(self):\n        return self._async_result.id\n\n    @classmethod\n    def add_task(cls, query, data_source, scheduled=False, metadata={}):\n        query_hash = gen_query_hash(query)\n        logging.info(\"[Manager][%s] Inserting job\", query_hash)\n        logging.info(\"[Manager] Metadata: [%s]\", metadata)\n        try_count = 0\n        job = None\n        \n        while try_count < cls.MAX_RETRIES:\n            try_count += 1\n\n            pipe = redis_connection.pipeline()\n            try:\n                pipe.watch(cls._job_lock_id(query_hash, data_source.id))\n                job_id = pipe.get(cls._job_lock_id(query_hash, data_source.id))\n                if job_id:\n                    logging.info(\"[Manager][%s] Found existing job: %s\", query_hash, job_id)\n\n                    job = cls(job_id=job_id)\n                    if job.ready():\n                        logging.info(\"[%s] job found is ready (%s), removing lock\", query_hash, job.celery_status)\n                        redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))\n                        job = None\n\n                if not job:\n                    pipe.multi()\n\n                    if scheduled:\n                        queue_name = data_source.scheduled_queue_name\n                    else:\n                        queue_name = data_source.queue_name\n\n                    result = execute_query.apply_async(args=(query, data_source.id, metadata), queue=queue_name)\n                    job = cls(async_result=result)\n                    \n                    logging.info(\"[Manager][%s] Created new job: %s\", query_hash, job.id)\n                    pipe.set(cls._job_lock_id(query_hash, data_source.id), job.id, settings.JOB_EXPIRY_TIME)\n                    pipe.execute()\n                break\n\n            except redis.WatchError:\n                continue\n\n        if not job:\n            logging.error(\"[Manager][%s] Failed adding job for query.\", query_hash)\n\n        return job\n\n    def to_dict(self):\n        if self._async_result.status == 'STARTED':\n            updated_at = self._async_result.result.get('start_time', 0)\n        else:\n            updated_at = 0\n\n        if self._async_result.failed() and isinstance(self._async_result.result, Exception):\n            error = self._async_result.result.message\n        elif self._async_result.status == 'REVOKED':\n            error = 'Query execution cancelled.'\n        else:\n            error = ''\n\n        if self._async_result.successful():\n            query_result_id = self._async_result.result\n        else:\n            query_result_id = None\n\n        return {\n            'id': self._async_result.id,\n            'updated_at': updated_at,\n            'status': self.STATUSES[self._async_result.status],\n            'error': error,\n            'query_result_id': query_result_id,\n        }\n\n    @property\n    def is_cancelled(self):\n        return self._async_result.status == 'REVOKED'\n\n    @property\n    def celery_status(self):\n        return self._async_result.status\n\n    def ready(self):\n        return self._async_result.ready()\n\n    def cancel(self):\n        return self._async_result.revoke(terminate=True)\n\n    @staticmethod\n    def _job_lock_id(query_hash, data_source_id):\n        return \"query_hash_job:%s:%s\" % (data_source_id, query_hash)\n\n\n@celery.task(base=BaseTask)\ndef refresh_queries():\n    # self.status['last_refresh_at'] = time.time()\n    # self._save_status()\n\n    logger.info(\"Refreshing queries...\")\n\n    outdated_queries_count = 0\n    for query in models.Query.outdated_queries():\n        QueryTask.add_task(query.query, query.data_source, scheduled=True,\n                           metadata={'Query ID': query.id, 'Username': 'Scheduled'})\n        outdated_queries_count += 1\n\n    statsd_client.gauge('manager.outdated_queries', outdated_queries_count)\n\n    logger.info(\"Done refreshing queries. Found %d outdated queries.\" % outdated_queries_count)\n\n    status = redis_connection.hgetall('redash:status')\n    now = time.time()\n\n    redis_connection.hmset('redash:status', {\n        'outdated_queries_count': outdated_queries_count,\n        'last_refresh_at': now\n    })\n\n    statsd_client.gauge('manager.seconds_since_refresh', now - float(status.get('last_refresh_at', now)))\n\n\n@celery.task(base=BaseTask)\ndef cleanup_tasks():\n    # in case of cold restart of the workers, there might be jobs that still have their \"lock\" object, but aren't really\n    # going to run. this job removes them.\n    lock_keys = redis_connection.keys(\"query_hash_job:*\") # TODO: use set instead of keys command\n    if not lock_keys:\n        return\n    \n    query_tasks = [QueryTask(job_id=j) for j in redis_connection.mget(lock_keys)]\n\n    logger.info(\"Found %d locks\", len(query_tasks))\n\n    inspect = celery.control.inspect()\n    active_tasks = inspect.active()\n    if active_tasks is None:\n        active_tasks = []\n    else:\n        active_tasks = active_tasks.values()\n\n    all_tasks = set()\n    for task_list in active_tasks:\n        for task in task_list:\n            all_tasks.add(task['id'])\n\n    logger.info(\"Active jobs count: %d\", len(all_tasks))\n\n    for i, t in enumerate(query_tasks):\n        if t.ready():\n            # if locked task is ready already (failed, finished, revoked), we don't need the lock anymore\n            logger.warning(\"%s is ready (%s), removing lock.\", lock_keys[i], t.celery_status)\n            redis_connection.delete(lock_keys[i])\n\n        # if t.celery_status == 'STARTED' and t.id not in all_tasks:\n        #     logger.warning(\"Couldn't find active job for: %s, removing lock.\", lock_keys[i])\n        #     redis_connection.delete(lock_keys[i])\n\n\n@celery.task(base=BaseTask)\ndef cleanup_query_results():\n    \"\"\"\n    Job to cleanup unused query results -- such that no query links to them anymore, and older than a week (so it's less\n    likely to be open in someone's browser and be used).\n\n    Each time the job deletes only 100 query results so it won't choke the database in case of many such results.\n    \"\"\"\n\n    unused_query_results = models.QueryResult.unused().limit(100)\n    total_unused_query_results = models.QueryResult.unused().count()\n    deleted_count = models.QueryResult.delete().where(models.QueryResult.id << unused_query_results).execute()\n\n    logger.info(\"Deleted %d unused query results out of total of %d.\" % (deleted_count, total_unused_query_results))\n\n\n@celery.task(base=BaseTask)\ndef refresh_schemas():\n    \"\"\"\n    Refershs the datasources schema.\n    \"\"\"\n\n    for ds in models.DataSource.all():\n        logger.info(\"Refreshing schema for: {}\".format(ds.name))\n        ds.get_schema(refresh=True)\n\n\n@celery.task(bind=True, base=BaseTask, track_started=True)\ndef execute_query(self, query, data_source_id, metadata):\n    start_time = time.time()\n\n    logger.info(\"Loading data source (%d)...\", data_source_id)\n\n    # TODO: we should probably cache data sources in Redis\n    data_source = models.DataSource.get_by_id(data_source_id)\n\n    self.update_state(state='STARTED', meta={'start_time': start_time, 'custom_message': ''})\n\n    logger.info(\"Executing query:\\n%s\", query)\n\n    query_hash = gen_query_hash(query)\n    query_runner = get_query_runner(data_source.type, data_source.options)\n\n    if query_runner.annotate_query():\n        metadata['Task ID'] = self.request.id\n        metadata['Query Hash'] = query_hash\n        metadata['Queue'] = self.request.delivery_info['routing_key']\n\n        annotation = u\", \".join([u\"{}: {}\".format(k, v) for k, v in metadata.iteritems()])\n\n        logging.debug(u\"Annotation: %s\", annotation)\n\n        annotated_query = u\"/* {} */ {}\".format(annotation, query)\n    else:\n        annotated_query = query\n\n    with statsd_client.timer('query_runner.{}.{}.run_time'.format(data_source.type, data_source.name)):\n        data, error = query_runner.run_query(annotated_query)\n\n    run_time = time.time() - start_time\n    logger.info(\"Query finished... data length=%s, error=%s\", data and len(data), error)\n\n    self.update_state(state='STARTED', meta={'start_time': start_time, 'error': error, 'custom_message': ''})\n\n    # Delete query_hash\n    redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))\n\n    if not error:\n        query_result = models.QueryResult.store_result(data_source.id, query_hash, query, data, run_time, datetime.datetime.utcnow())\n    else:\n        raise Exception(error)\n\n    return query_result.id\n\n\n@celery.task(base=BaseTask)\ndef record_event(event):\n    models.Event.record(event)\n", "path": "redash/tasks.py"}]}