We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
User can't add another dashboard after creating one without refreshing


--- 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
9 from redash.utils import gen_query_hash
10 from redash.worker import celery
11 from redash.data.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):
51         query_hash = gen_query_hash(query)
52         logging.info("[Manager][%s] Inserting job", query_hash)
53         try_count = 0
54         job = None
55 
56         while try_count < cls.MAX_RETRIES:
57             try_count += 1
58 
59             pipe = redis_connection.pipeline()
60             try:
61                 pipe.watch('query_hash_job:%s' % query_hash)
62                 job_id = pipe.get('query_hash_job:%s' % query_hash)
63                 if job_id:
64                     logging.info("[Manager][%s] Found existing job: %s", query_hash, job_id)
65 
66                     job = cls(job_id=job_id)
67                 else:
68                     pipe.multi()
69 
70                     if scheduled:
71                         queue_name = data_source.queue_name
72                     else:
73                         queue_name = data_source.scheduled_queue_name
74 
75                     result = execute_query.apply_async(args=(query, data_source.id), queue=queue_name)
76                     job = cls(async_result=result)
77                     logging.info("[Manager][%s] Created new job: %s", query_hash, job.id)
78                     pipe.set('query_hash_job:%s' % query_hash, job.id)
79                     pipe.execute()
80                 break
81 
82             except redis.WatchError:
83                 continue
84 
85         if not job:
86             logging.error("[Manager][%s] Failed adding job for query.", query_hash)
87 
88         return job
89 
90     def to_dict(self):
91         if self._async_result.status == 'STARTED':
92             updated_at = self._async_result.result.get('start_time', 0)
93         else:
94             updated_at = 0
95 
96         if self._async_result.failed() and isinstance(self._async_result.result, Exception):
97             error = self._async_result.result.message
98         elif self._async_result.status == 'REVOKED':
99             error = 'Query execution cancelled.'
100         else:
101             error = ''
102 
103         if self._async_result.successful():
104             query_result_id = self._async_result.result
105         else:
106             query_result_id = None
107 
108         return {
109             'id': self._async_result.id,
110             'updated_at': updated_at,
111             'status': self.STATUSES[self._async_result.status],
112             'error': error,
113             'query_result_id': query_result_id,
114         }
115 
116     def cancel(self):
117         return self._async_result.revoke(terminate=True)
118 
119 
120 @celery.task(base=BaseTask)
121 def refresh_queries():
122     # self.status['last_refresh_at'] = time.time()
123     # self._save_status()
124 
125     logger.info("Refreshing queries...")
126 
127     outdated_queries_count = 0
128     for query in models.Query.outdated_queries():
129         # TODO: this should go into lower priority
130         QueryTask.add_task(query.query, query.data_source, scheduled=True)
131         outdated_queries_count += 1
132 
133     statsd_client.gauge('manager.outdated_queries', outdated_queries_count)
134     # TODO: decide if we still need this
135     # statsd_client.gauge('manager.queue_size', self.redis_connection.zcard('jobs'))
136 
137     logger.info("Done refreshing queries. Found %d outdated queries." % outdated_queries_count)
138 
139     status = redis_connection.hgetall('redash:status')
140     now = time.time()
141 
142     redis_connection.hmset('redash:status', {
143         'outdated_queries_count': outdated_queries_count,
144         'last_refresh_at': now
145     })
146 
147     statsd_client.gauge('manager.seconds_since_refresh', now - float(status.get('last_refresh_at', now)))
148 
149 @celery.task(bind=True, base=BaseTask, track_started=True)
150 def execute_query(self, query, data_source_id):
151     # TODO: maybe this should be a class?
152     start_time = time.time()
153 
154     logger.info("Loading data source (%d)...", data_source_id)
155 
156     # TODO: we should probably cache data sources in Redis
157     data_source = models.DataSource.get_by_id(data_source_id)
158 
159     self.update_state(state='STARTED', meta={'start_time': start_time, 'custom_message': ''})
160 
161     logger.info("Executing query:\n%s", query)
162 
163     query_hash = gen_query_hash(query)
164     query_runner = get_query_runner(data_source.type, data_source.options)
165 
166     if getattr(query_runner, 'annotate_query', True):
167         # TODO: anotate with queu ename
168         annotated_query = "/* Task Id: %s, Query hash: %s */ %s" % \
169                           (self.request.id, query_hash, query)
170     else:
171         annotated_query = query
172 
173     with statsd_client.timer('query_runner.{}.{}.run_time'.format(data_source.type, data_source.name)):
174         data, error = query_runner(annotated_query)
175 
176     run_time = time.time() - start_time
177     logger.info("Query finished... data length=%s, error=%s", data and len(data), error)
178 
179     self.update_state(state='STARTED', meta={'start_time': start_time, 'error': error, 'custom_message': ''})
180 
181     # TODO: it is possible that storing the data will fail, and we will need to retry
182     # while we already marked the job as done
183     # Delete query_hash
184     redis_connection.delete('query_hash_job:%s' % query_hash)
185 
186     if not error:
187         query_result = models.QueryResult.store_result(data_source.id, query_hash, query, data, run_time, datetime.datetime.utcnow())
188     else:
189         raise Exception(error)
190 
191     return query_result.id
192 
193
```
--- 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
@@ -58,8 +58,8 @@
 
             pipe = redis_connection.pipeline()
             try:
-                pipe.watch('query_hash_job:%s' % query_hash)
-                job_id = pipe.get('query_hash_job:%s' % query_hash)
+                pipe.watch(cls._job_lock_id(query_hash, data_source.id))
+                job_id = pipe.get(cls._job_lock_id(query_hash, data_source.id))
                 if job_id:
                     logging.info("[Manager][%s] Found existing job: %s", query_hash, job_id)
 
@@ -75,7 +75,7 @@
                     result = execute_query.apply_async(args=(query, data_source.id), queue=queue_name)
                     job = cls(async_result=result)
                     logging.info("[Manager][%s] Created new job: %s", query_hash, job.id)
-                    pipe.set('query_hash_job:%s' % query_hash, job.id)
+                    pipe.set(cls._job_lock_id(query_hash, data_source.id), job.id)
                     pipe.execute()
                 break
 
@@ -116,6 +116,9 @@
     def cancel(self):
         return self._async_result.revoke(terminate=True)
 
+    @staticmethod
+    def _job_lock_id(query_hash, data_source_id):
+        return "query_hash_job:%s:%s" % (data_source_id, query_hash)
 
 @celery.task(base=BaseTask)
 def refresh_queries():
@@ -178,11 +181,11 @@
 
     self.update_state(state='STARTED', meta={'start_time': start_time, 'error': error, 'custom_message': ''})
 
-    # TODO: it is possible that storing the data will fail, and we will need to retry
-    # while we already marked the job as done
     # Delete query_hash
-    redis_connection.delete('query_hash_job:%s' % query_hash)
+    redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))
 
+    # TODO: it is possible that storing the data will fail, and we will need to retry
+    # while we already marked the job as done
     if not error:
         query_result = models.QueryResult.store_result(data_source.id, query_hash, query, data, run_time, datetime.datetime.utcnow())
     else:

{"golden_diff": "diff --git a/redash/tasks.py b/redash/tasks.py\n--- a/redash/tasks.py\n+++ b/redash/tasks.py\n@@ -58,8 +58,8 @@\n \n             pipe = redis_connection.pipeline()\n             try:\n-                pipe.watch('query_hash_job:%s' % query_hash)\n-                job_id = pipe.get('query_hash_job:%s' % query_hash)\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@@ -75,7 +75,7 @@\n                     result = execute_query.apply_async(args=(query, data_source.id), queue=queue_name)\n                     job = cls(async_result=result)\n                     logging.info(\"[Manager][%s] Created new job: %s\", query_hash, job.id)\n-                    pipe.set('query_hash_job:%s' % query_hash, job.id)\n+                    pipe.set(cls._job_lock_id(query_hash, data_source.id), job.id)\n                     pipe.execute()\n                 break\n \n@@ -116,6 +116,9 @@\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 @celery.task(base=BaseTask)\n def refresh_queries():\n@@ -178,11 +181,11 @@\n \n     self.update_state(state='STARTED', meta={'start_time': start_time, 'error': error, 'custom_message': ''})\n \n-    # TODO: it is possible that storing the data will fail, and we will need to retry\n-    # while we already marked the job as done\n     # Delete query_hash\n-    redis_connection.delete('query_hash_job:%s' % query_hash)\n+    redis_connection.delete(QueryTask._job_lock_id(query_hash, data_source.id))\n \n+    # TODO: it is possible that storing the data will fail, and we will need to retry\n+    # while we already marked the job as done\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", "issue": "User can't add another dashboard after creating one without refreshing\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\nfrom redash.utils import gen_query_hash\nfrom redash.worker import celery\nfrom redash.data.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):\n        query_hash = gen_query_hash(query)\n        logging.info(\"[Manager][%s] Inserting job\", query_hash)\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('query_hash_job:%s' % query_hash)\n                job_id = pipe.get('query_hash_job:%s' % query_hash)\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                else:\n                    pipe.multi()\n\n                    if scheduled:\n                        queue_name = data_source.queue_name\n                    else:\n                        queue_name = data_source.scheduled_queue_name\n\n                    result = execute_query.apply_async(args=(query, data_source.id), queue=queue_name)\n                    job = cls(async_result=result)\n                    logging.info(\"[Manager][%s] Created new job: %s\", query_hash, job.id)\n                    pipe.set('query_hash_job:%s' % query_hash, job.id)\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    def cancel(self):\n        return self._async_result.revoke(terminate=True)\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        # TODO: this should go into lower priority\n        QueryTask.add_task(query.query, query.data_source, scheduled=True)\n        outdated_queries_count += 1\n\n    statsd_client.gauge('manager.outdated_queries', outdated_queries_count)\n    # TODO: decide if we still need this\n    # statsd_client.gauge('manager.queue_size', self.redis_connection.zcard('jobs'))\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@celery.task(bind=True, base=BaseTask, track_started=True)\ndef execute_query(self, query, data_source_id):\n    # TODO: maybe this should be a class?\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 getattr(query_runner, 'annotate_query', True):\n        # TODO: anotate with queu ename\n        annotated_query = \"/* Task Id: %s, Query hash: %s */ %s\" % \\\n                          (self.request.id, query_hash, 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(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    # TODO: it is possible that storing the data will fail, and we will need to retry\n    # while we already marked the job as done\n    # Delete query_hash\n    redis_connection.delete('query_hash_job:%s' % query_hash)\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", "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\nfrom redash.utils import gen_query_hash\nfrom redash.worker import celery\nfrom redash.data.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):\n        query_hash = gen_query_hash(query)\n        logging.info(\"[Manager][%s] Inserting job\", query_hash)\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                else:\n                    pipe.multi()\n\n                    if scheduled:\n                        queue_name = data_source.queue_name\n                    else:\n                        queue_name = data_source.scheduled_queue_name\n\n                    result = execute_query.apply_async(args=(query, data_source.id), queue=queue_name)\n                    job = cls(async_result=result)\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)\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    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@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        # TODO: this should go into lower priority\n        QueryTask.add_task(query.query, query.data_source, scheduled=True)\n        outdated_queries_count += 1\n\n    statsd_client.gauge('manager.outdated_queries', outdated_queries_count)\n    # TODO: decide if we still need this\n    # statsd_client.gauge('manager.queue_size', self.redis_connection.zcard('jobs'))\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@celery.task(bind=True, base=BaseTask, track_started=True)\ndef execute_query(self, query, data_source_id):\n    # TODO: maybe this should be a class?\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 getattr(query_runner, 'annotate_query', True):\n        # TODO: anotate with queu ename\n        annotated_query = \"/* Task Id: %s, Query hash: %s */ %s\" % \\\n                          (self.request.id, query_hash, 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(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    # TODO: it is possible that storing the data will fail, and we will need to retry\n    # while we already marked the job as done\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", "path": "redash/tasks.py"}]}