We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Admin shows n/a for started value in manager column
<!--

We use GitHub only for bug reports 🐛

Anything else should be posted to https://discuss.redash.io 👫

🚨For support, help & questions use https://discuss.redash.io/c/support
💡For feature requests & ideas use https://discuss.redash.io/c/feature-requests

**Found a security vulnerability?** Please email security@redash.io to report any security vulnerabilities. We will acknowledge receipt of your vulnerability and strive to send you regular updates about our progress. If you're curious about the status of your disclosure please feel free to email us again. If you want to encrypt your disclosure email, you can use this PGP key.

-->

### Issue Summary

When going to the admin section you can see in the manager column that the "Started" value is always "n/a".

<img width="630" alt="CleanShot 2020-05-15 at 01 58 09@2x" src="https://user-images.githubusercontent.com/1610/81997320-b31fa600-964f-11ea-90c0-e0c2988fdcce.png">

### Steps to Reproduce

I **think** this a regression from first https://github.com/getredash/redash/commit/26f0ce0749c1f683ba3b8f0b1d2a9e70cb6ea0e2#diff-5ca0dc869c9c3218d8a95ba7f99f5f4cL460 where it removed the QueryTaskTracker and by that the ability to store the `started_at` time and then later when Celery was removed, where this code path became no-op.

Originally this value was shown in v6.0.0 here: https://github.com/getredash/redash/blob/4780bd9c5ef212dd4c38bafb525abc991812d59b/client/app/pages/admin/status/status.html#L35

Now it's pulled in from the `/status.json` api in https://github.com/getredash/redash/blob/8010781f0d3d14260523e135a5abef0847714b9e/client/app/pages/admin/SystemStatus.jsx#L47 but the value behind is not set anymore in https://github.com/getredash/redash/blob/8010781f0d3d14260523e135a5abef0847714b9e/redash/tasks/queries/maintenance.py#L95-L101.

Any other info e.g. Why do you consider this to be a bug? What did you expect to happen instead?

### Technical details:

* Redash Version: 8010781f0d3d14260523e135a5abef0847714b9e
* Browser/OS: Firefox macOS
* How did you install Redash: Docker


--- 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/queries/maintenance.py`
Content:
```
1 import logging
2 import time
3 
4 from rq.timeouts import JobTimeoutException
5 
6 from redash import models, redis_connection, settings, statsd_client
7 from redash.models.parameterized_query import (
8     InvalidParameterError,
9     QueryDetachedFromDataSourceError,
10 )
11 from redash.monitor import rq_job_ids
12 from redash.tasks.failure_report import track_failure
13 from redash.utils import json_dumps, sentry
14 from redash.worker import get_job_logger, job
15 
16 from .execution import enqueue_query
17 
18 logger = get_job_logger(__name__)
19 
20 
21 def empty_schedules():
22     logger.info("Deleting schedules of past scheduled queries...")
23 
24     queries = models.Query.past_scheduled_queries()
25     for query in queries:
26         query.schedule = None
27     models.db.session.commit()
28 
29     logger.info("Deleted %d schedules.", len(queries))
30 
31 
32 def _should_refresh_query(query):
33     if settings.FEATURE_DISABLE_REFRESH_QUERIES:
34         logger.info("Disabled refresh queries.")
35         return False
36     elif query.org.is_disabled:
37         logger.debug("Skipping refresh of %s because org is disabled.", query.id)
38         return False
39     elif query.data_source is None:
40         logger.debug("Skipping refresh of %s because the datasource is none.", query.id)
41         return False
42     elif query.data_source.paused:
43         logger.debug(
44             "Skipping refresh of %s because datasource - %s is paused (%s).",
45             query.id,
46             query.data_source.name,
47             query.data_source.pause_reason,
48         )
49         return False
50     else:
51         return True
52 
53 
54 def _apply_default_parameters(query):
55     parameters = {p["name"]: p.get("value") for p in query.parameters}
56     if any(parameters):
57         try:
58             return query.parameterized.apply(parameters).query
59         except InvalidParameterError as e:
60             error = f"Skipping refresh of {query.id} because of invalid parameters: {str(e)}"
61             track_failure(query, error)
62             raise
63         except QueryDetachedFromDataSourceError as e:
64             error = (
65                 f"Skipping refresh of {query.id} because a related dropdown "
66                 f"query ({e.query_id}) is unattached to any datasource."
67             )
68             track_failure(query, error)
69             raise
70     else:
71         return query.query_text
72 
73 
74 class RefreshQueriesError(Exception):
75     pass
76 
77 
78 def _apply_auto_limit(query_text, query):
79     should_apply_auto_limit = query.options.get("apply_auto_limit", False)
80     return query.data_source.query_runner.apply_auto_limit(query_text, should_apply_auto_limit)
81 
82 
83 def refresh_queries():
84     logger.info("Refreshing queries...")
85     enqueued = []
86     for query in models.Query.outdated_queries():
87         if not _should_refresh_query(query):
88             continue
89 
90         try:
91             query_text = _apply_default_parameters(query)
92             query_text = _apply_auto_limit(query_text, query)
93             enqueue_query(
94                 query_text,
95                 query.data_source,
96                 query.user_id,
97                 scheduled_query=query,
98                 metadata={"query_id": query.id, "Username": "Scheduled"},
99             )
100             enqueued.append(query)
101         except Exception as e:
102             message = "Could not enqueue query %d due to %s" % (query.id, repr(e))
103             logging.info(message)
104             error = RefreshQueriesError(message).with_traceback(e.__traceback__)
105             sentry.capture_exception(error)
106 
107     status = {
108         "outdated_queries_count": len(enqueued),
109         "last_refresh_at": time.time(),
110         "query_ids": json_dumps([q.id for q in enqueued]),
111     }
112 
113     redis_connection.hmset("redash:status", status)
114     logger.info("Done refreshing queries: %s" % status)
115 
116 
117 def cleanup_query_results():
118     """
119     Job to cleanup unused query results -- such that no query links to them anymore, and older than
120     settings.QUERY_RESULTS_CLEANUP_MAX_AGE (a week by default, so it's less likely to be open in someone's browser and be used).
121 
122     Each time the job deletes only settings.QUERY_RESULTS_CLEANUP_COUNT (100 by default) query results so it won't choke
123     the database in case of many such results.
124     """
125 
126     logger.info(
127         "Running query results clean up (removing maximum of %d unused results, that are %d days old or more)",
128         settings.QUERY_RESULTS_CLEANUP_COUNT,
129         settings.QUERY_RESULTS_CLEANUP_MAX_AGE,
130     )
131 
132     unused_query_results = models.QueryResult.unused(settings.QUERY_RESULTS_CLEANUP_MAX_AGE)
133     deleted_count = models.QueryResult.query.filter(
134         models.QueryResult.id.in_(unused_query_results.limit(settings.QUERY_RESULTS_CLEANUP_COUNT).subquery())
135     ).delete(synchronize_session=False)
136     models.db.session.commit()
137     logger.info("Deleted %d unused query results.", deleted_count)
138 
139 
140 def remove_ghost_locks():
141     """
142     Removes query locks that reference a non existing RQ job.
143     """
144     keys = redis_connection.keys("query_hash_job:*")
145     locks = {k: redis_connection.get(k) for k in keys}
146     jobs = list(rq_job_ids())
147 
148     count = 0
149 
150     for lock, job_id in locks.items():
151         if job_id not in jobs:
152             redis_connection.delete(lock)
153             count += 1
154 
155     logger.info("Locks found: {}, Locks removed: {}".format(len(locks), count))
156 
157 
158 @job("schemas")
159 def refresh_schema(data_source_id):
160     ds = models.DataSource.get_by_id(data_source_id)
161     logger.info("task=refresh_schema state=start ds_id=%s", ds.id)
162     start_time = time.time()
163     try:
164         ds.get_schema(refresh=True)
165         logger.info(
166             "task=refresh_schema state=finished ds_id=%s runtime=%.2f",
167             ds.id,
168             time.time() - start_time,
169         )
170         statsd_client.incr("refresh_schema.success")
171     except JobTimeoutException:
172         logger.info(
173             "task=refresh_schema state=timeout ds_id=%s runtime=%.2f",
174             ds.id,
175             time.time() - start_time,
176         )
177         statsd_client.incr("refresh_schema.timeout")
178     except Exception:
179         logger.warning("Failed refreshing schema for the data source: %s", ds.name, exc_info=1)
180         statsd_client.incr("refresh_schema.error")
181         logger.info(
182             "task=refresh_schema state=failed ds_id=%s runtime=%.2f",
183             ds.id,
184             time.time() - start_time,
185         )
186 
187 
188 def refresh_schemas():
189     """
190     Refreshes the data sources schemas.
191     """
192     blacklist = [int(ds_id) for ds_id in redis_connection.smembers("data_sources:schema:blacklist") if ds_id]
193     global_start_time = time.time()
194 
195     logger.info("task=refresh_schemas state=start")
196 
197     for ds in models.DataSource.query:
198         if ds.paused:
199             logger.info(
200                 "task=refresh_schema state=skip ds_id=%s reason=paused(%s)",
201                 ds.id,
202                 ds.pause_reason,
203             )
204         elif ds.id in blacklist:
205             logger.info("task=refresh_schema state=skip ds_id=%s reason=blacklist", ds.id)
206         elif ds.org.is_disabled:
207             logger.info("task=refresh_schema state=skip ds_id=%s reason=org_disabled", ds.id)
208         else:
209             refresh_schema.delay(ds.id)
210 
211     logger.info(
212         "task=refresh_schemas state=finish total_runtime=%.2f",
213         time.time() - global_start_time,
214     )
215
```
--- 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/queries/maintenance.py b/redash/tasks/queries/maintenance.py
--- a/redash/tasks/queries/maintenance.py
+++ b/redash/tasks/queries/maintenance.py
@@ -81,6 +81,7 @@
 
 
 def refresh_queries():
+    started_at = time.time()
     logger.info("Refreshing queries...")
     enqueued = []
     for query in models.Query.outdated_queries():
@@ -105,6 +106,7 @@
             sentry.capture_exception(error)
 
     status = {
+        "started_at": started_at,
         "outdated_queries_count": len(enqueued),
         "last_refresh_at": time.time(),
         "query_ids": json_dumps([q.id for q in enqueued]),

{"golden_diff": "diff --git a/redash/tasks/queries/maintenance.py b/redash/tasks/queries/maintenance.py\n--- a/redash/tasks/queries/maintenance.py\n+++ b/redash/tasks/queries/maintenance.py\n@@ -81,6 +81,7 @@\n \n \n def refresh_queries():\n+    started_at = time.time()\n     logger.info(\"Refreshing queries...\")\n     enqueued = []\n     for query in models.Query.outdated_queries():\n@@ -105,6 +106,7 @@\n             sentry.capture_exception(error)\n \n     status = {\n+        \"started_at\": started_at,\n         \"outdated_queries_count\": len(enqueued),\n         \"last_refresh_at\": time.time(),\n         \"query_ids\": json_dumps([q.id for q in enqueued]),\n", "issue": "Admin shows n/a for started value in manager column\n<!--\r\n\r\nWe use GitHub only for bug reports \ud83d\udc1b\r\n\r\nAnything else should be posted to https://discuss.redash.io \ud83d\udc6b\r\n\r\n\ud83d\udea8For support, help & questions use https://discuss.redash.io/c/support\r\n\ud83d\udca1For feature requests & ideas use https://discuss.redash.io/c/feature-requests\r\n\r\n**Found a security vulnerability?** Please email security@redash.io to report any security vulnerabilities. We will acknowledge receipt of your vulnerability and strive to send you regular updates about our progress. If you're curious about the status of your disclosure please feel free to email us again. If you want to encrypt your disclosure email, you can use this PGP key.\r\n\r\n-->\r\n\r\n### Issue Summary\r\n\r\nWhen going to the admin section you can see in the manager column that the \"Started\" value is always \"n/a\".\r\n\r\n<img width=\"630\" alt=\"CleanShot 2020-05-15 at 01 58 09@2x\" src=\"https://user-images.githubusercontent.com/1610/81997320-b31fa600-964f-11ea-90c0-e0c2988fdcce.png\">\r\n\r\n### Steps to Reproduce\r\n\r\nI **think** this a regression from first https://github.com/getredash/redash/commit/26f0ce0749c1f683ba3b8f0b1d2a9e70cb6ea0e2#diff-5ca0dc869c9c3218d8a95ba7f99f5f4cL460 where it removed the QueryTaskTracker and by that the ability to store the `started_at` time and then later when Celery was removed, where this code path became no-op.\r\n\r\nOriginally this value was shown in v6.0.0 here: https://github.com/getredash/redash/blob/4780bd9c5ef212dd4c38bafb525abc991812d59b/client/app/pages/admin/status/status.html#L35\r\n\r\nNow it's pulled in from the `/status.json` api in https://github.com/getredash/redash/blob/8010781f0d3d14260523e135a5abef0847714b9e/client/app/pages/admin/SystemStatus.jsx#L47 but the value behind is not set anymore in https://github.com/getredash/redash/blob/8010781f0d3d14260523e135a5abef0847714b9e/redash/tasks/queries/maintenance.py#L95-L101.\r\n\r\nAny other info e.g. Why do you consider this to be a bug? What did you expect to happen instead?\r\n\r\n### Technical details:\r\n\r\n* Redash Version: 8010781f0d3d14260523e135a5abef0847714b9e\r\n* Browser/OS: Firefox macOS\r\n* How did you install Redash: Docker\r\n\n", "before_files": [{"content": "import logging\nimport time\n\nfrom rq.timeouts import JobTimeoutException\n\nfrom redash import models, redis_connection, settings, statsd_client\nfrom redash.models.parameterized_query import (\n    InvalidParameterError,\n    QueryDetachedFromDataSourceError,\n)\nfrom redash.monitor import rq_job_ids\nfrom redash.tasks.failure_report import track_failure\nfrom redash.utils import json_dumps, sentry\nfrom redash.worker import get_job_logger, job\n\nfrom .execution import enqueue_query\n\nlogger = get_job_logger(__name__)\n\n\ndef empty_schedules():\n    logger.info(\"Deleting schedules of past scheduled queries...\")\n\n    queries = models.Query.past_scheduled_queries()\n    for query in queries:\n        query.schedule = None\n    models.db.session.commit()\n\n    logger.info(\"Deleted %d schedules.\", len(queries))\n\n\ndef _should_refresh_query(query):\n    if settings.FEATURE_DISABLE_REFRESH_QUERIES:\n        logger.info(\"Disabled refresh queries.\")\n        return False\n    elif query.org.is_disabled:\n        logger.debug(\"Skipping refresh of %s because org is disabled.\", query.id)\n        return False\n    elif query.data_source is None:\n        logger.debug(\"Skipping refresh of %s because the datasource is none.\", query.id)\n        return False\n    elif query.data_source.paused:\n        logger.debug(\n            \"Skipping refresh of %s because datasource - %s is paused (%s).\",\n            query.id,\n            query.data_source.name,\n            query.data_source.pause_reason,\n        )\n        return False\n    else:\n        return True\n\n\ndef _apply_default_parameters(query):\n    parameters = {p[\"name\"]: p.get(\"value\") for p in query.parameters}\n    if any(parameters):\n        try:\n            return query.parameterized.apply(parameters).query\n        except InvalidParameterError as e:\n            error = f\"Skipping refresh of {query.id} because of invalid parameters: {str(e)}\"\n            track_failure(query, error)\n            raise\n        except QueryDetachedFromDataSourceError as e:\n            error = (\n                f\"Skipping refresh of {query.id} because a related dropdown \"\n                f\"query ({e.query_id}) is unattached to any datasource.\"\n            )\n            track_failure(query, error)\n            raise\n    else:\n        return query.query_text\n\n\nclass RefreshQueriesError(Exception):\n    pass\n\n\ndef _apply_auto_limit(query_text, query):\n    should_apply_auto_limit = query.options.get(\"apply_auto_limit\", False)\n    return query.data_source.query_runner.apply_auto_limit(query_text, should_apply_auto_limit)\n\n\ndef refresh_queries():\n    logger.info(\"Refreshing queries...\")\n    enqueued = []\n    for query in models.Query.outdated_queries():\n        if not _should_refresh_query(query):\n            continue\n\n        try:\n            query_text = _apply_default_parameters(query)\n            query_text = _apply_auto_limit(query_text, query)\n            enqueue_query(\n                query_text,\n                query.data_source,\n                query.user_id,\n                scheduled_query=query,\n                metadata={\"query_id\": query.id, \"Username\": \"Scheduled\"},\n            )\n            enqueued.append(query)\n        except Exception as e:\n            message = \"Could not enqueue query %d due to %s\" % (query.id, repr(e))\n            logging.info(message)\n            error = RefreshQueriesError(message).with_traceback(e.__traceback__)\n            sentry.capture_exception(error)\n\n    status = {\n        \"outdated_queries_count\": len(enqueued),\n        \"last_refresh_at\": time.time(),\n        \"query_ids\": json_dumps([q.id for q in enqueued]),\n    }\n\n    redis_connection.hmset(\"redash:status\", status)\n    logger.info(\"Done refreshing queries: %s\" % status)\n\n\ndef cleanup_query_results():\n    \"\"\"\n    Job to cleanup unused query results -- such that no query links to them anymore, and older than\n    settings.QUERY_RESULTS_CLEANUP_MAX_AGE (a week by default, so it's less likely to be open in someone's browser and be used).\n\n    Each time the job deletes only settings.QUERY_RESULTS_CLEANUP_COUNT (100 by default) query results so it won't choke\n    the database in case of many such results.\n    \"\"\"\n\n    logger.info(\n        \"Running query results clean up (removing maximum of %d unused results, that are %d days old or more)\",\n        settings.QUERY_RESULTS_CLEANUP_COUNT,\n        settings.QUERY_RESULTS_CLEANUP_MAX_AGE,\n    )\n\n    unused_query_results = models.QueryResult.unused(settings.QUERY_RESULTS_CLEANUP_MAX_AGE)\n    deleted_count = models.QueryResult.query.filter(\n        models.QueryResult.id.in_(unused_query_results.limit(settings.QUERY_RESULTS_CLEANUP_COUNT).subquery())\n    ).delete(synchronize_session=False)\n    models.db.session.commit()\n    logger.info(\"Deleted %d unused query results.\", deleted_count)\n\n\ndef remove_ghost_locks():\n    \"\"\"\n    Removes query locks that reference a non existing RQ job.\n    \"\"\"\n    keys = redis_connection.keys(\"query_hash_job:*\")\n    locks = {k: redis_connection.get(k) for k in keys}\n    jobs = list(rq_job_ids())\n\n    count = 0\n\n    for lock, job_id in locks.items():\n        if job_id not in jobs:\n            redis_connection.delete(lock)\n            count += 1\n\n    logger.info(\"Locks found: {}, Locks removed: {}\".format(len(locks), count))\n\n\n@job(\"schemas\")\ndef refresh_schema(data_source_id):\n    ds = models.DataSource.get_by_id(data_source_id)\n    logger.info(\"task=refresh_schema state=start ds_id=%s\", ds.id)\n    start_time = time.time()\n    try:\n        ds.get_schema(refresh=True)\n        logger.info(\n            \"task=refresh_schema state=finished ds_id=%s runtime=%.2f\",\n            ds.id,\n            time.time() - start_time,\n        )\n        statsd_client.incr(\"refresh_schema.success\")\n    except JobTimeoutException:\n        logger.info(\n            \"task=refresh_schema state=timeout ds_id=%s runtime=%.2f\",\n            ds.id,\n            time.time() - start_time,\n        )\n        statsd_client.incr(\"refresh_schema.timeout\")\n    except Exception:\n        logger.warning(\"Failed refreshing schema for the data source: %s\", ds.name, exc_info=1)\n        statsd_client.incr(\"refresh_schema.error\")\n        logger.info(\n            \"task=refresh_schema state=failed ds_id=%s runtime=%.2f\",\n            ds.id,\n            time.time() - start_time,\n        )\n\n\ndef refresh_schemas():\n    \"\"\"\n    Refreshes the data sources schemas.\n    \"\"\"\n    blacklist = [int(ds_id) for ds_id in redis_connection.smembers(\"data_sources:schema:blacklist\") if ds_id]\n    global_start_time = time.time()\n\n    logger.info(\"task=refresh_schemas state=start\")\n\n    for ds in models.DataSource.query:\n        if ds.paused:\n            logger.info(\n                \"task=refresh_schema state=skip ds_id=%s reason=paused(%s)\",\n                ds.id,\n                ds.pause_reason,\n            )\n        elif ds.id in blacklist:\n            logger.info(\"task=refresh_schema state=skip ds_id=%s reason=blacklist\", ds.id)\n        elif ds.org.is_disabled:\n            logger.info(\"task=refresh_schema state=skip ds_id=%s reason=org_disabled\", ds.id)\n        else:\n            refresh_schema.delay(ds.id)\n\n    logger.info(\n        \"task=refresh_schemas state=finish total_runtime=%.2f\",\n        time.time() - global_start_time,\n    )\n", "path": "redash/tasks/queries/maintenance.py"}], "after_files": [{"content": "import logging\nimport time\n\nfrom rq.timeouts import JobTimeoutException\n\nfrom redash import models, redis_connection, settings, statsd_client\nfrom redash.models.parameterized_query import (\n    InvalidParameterError,\n    QueryDetachedFromDataSourceError,\n)\nfrom redash.monitor import rq_job_ids\nfrom redash.tasks.failure_report import track_failure\nfrom redash.utils import json_dumps, sentry\nfrom redash.worker import get_job_logger, job\n\nfrom .execution import enqueue_query\n\nlogger = get_job_logger(__name__)\n\n\ndef empty_schedules():\n    logger.info(\"Deleting schedules of past scheduled queries...\")\n\n    queries = models.Query.past_scheduled_queries()\n    for query in queries:\n        query.schedule = None\n    models.db.session.commit()\n\n    logger.info(\"Deleted %d schedules.\", len(queries))\n\n\ndef _should_refresh_query(query):\n    if settings.FEATURE_DISABLE_REFRESH_QUERIES:\n        logger.info(\"Disabled refresh queries.\")\n        return False\n    elif query.org.is_disabled:\n        logger.debug(\"Skipping refresh of %s because org is disabled.\", query.id)\n        return False\n    elif query.data_source is None:\n        logger.debug(\"Skipping refresh of %s because the datasource is none.\", query.id)\n        return False\n    elif query.data_source.paused:\n        logger.debug(\n            \"Skipping refresh of %s because datasource - %s is paused (%s).\",\n            query.id,\n            query.data_source.name,\n            query.data_source.pause_reason,\n        )\n        return False\n    else:\n        return True\n\n\ndef _apply_default_parameters(query):\n    parameters = {p[\"name\"]: p.get(\"value\") for p in query.parameters}\n    if any(parameters):\n        try:\n            return query.parameterized.apply(parameters).query\n        except InvalidParameterError as e:\n            error = f\"Skipping refresh of {query.id} because of invalid parameters: {str(e)}\"\n            track_failure(query, error)\n            raise\n        except QueryDetachedFromDataSourceError as e:\n            error = (\n                f\"Skipping refresh of {query.id} because a related dropdown \"\n                f\"query ({e.query_id}) is unattached to any datasource.\"\n            )\n            track_failure(query, error)\n            raise\n    else:\n        return query.query_text\n\n\nclass RefreshQueriesError(Exception):\n    pass\n\n\ndef _apply_auto_limit(query_text, query):\n    should_apply_auto_limit = query.options.get(\"apply_auto_limit\", False)\n    return query.data_source.query_runner.apply_auto_limit(query_text, should_apply_auto_limit)\n\n\ndef refresh_queries():\n    started_at = time.time()\n    logger.info(\"Refreshing queries...\")\n    enqueued = []\n    for query in models.Query.outdated_queries():\n        if not _should_refresh_query(query):\n            continue\n\n        try:\n            query_text = _apply_default_parameters(query)\n            query_text = _apply_auto_limit(query_text, query)\n            enqueue_query(\n                query_text,\n                query.data_source,\n                query.user_id,\n                scheduled_query=query,\n                metadata={\"query_id\": query.id, \"Username\": \"Scheduled\"},\n            )\n            enqueued.append(query)\n        except Exception as e:\n            message = \"Could not enqueue query %d due to %s\" % (query.id, repr(e))\n            logging.info(message)\n            error = RefreshQueriesError(message).with_traceback(e.__traceback__)\n            sentry.capture_exception(error)\n\n    status = {\n        \"started_at\": started_at,\n        \"outdated_queries_count\": len(enqueued),\n        \"last_refresh_at\": time.time(),\n        \"query_ids\": json_dumps([q.id for q in enqueued]),\n    }\n\n    redis_connection.hmset(\"redash:status\", status)\n    logger.info(\"Done refreshing queries: %s\" % status)\n\n\ndef cleanup_query_results():\n    \"\"\"\n    Job to cleanup unused query results -- such that no query links to them anymore, and older than\n    settings.QUERY_RESULTS_CLEANUP_MAX_AGE (a week by default, so it's less likely to be open in someone's browser and be used).\n\n    Each time the job deletes only settings.QUERY_RESULTS_CLEANUP_COUNT (100 by default) query results so it won't choke\n    the database in case of many such results.\n    \"\"\"\n\n    logger.info(\n        \"Running query results clean up (removing maximum of %d unused results, that are %d days old or more)\",\n        settings.QUERY_RESULTS_CLEANUP_COUNT,\n        settings.QUERY_RESULTS_CLEANUP_MAX_AGE,\n    )\n\n    unused_query_results = models.QueryResult.unused(settings.QUERY_RESULTS_CLEANUP_MAX_AGE)\n    deleted_count = models.QueryResult.query.filter(\n        models.QueryResult.id.in_(unused_query_results.limit(settings.QUERY_RESULTS_CLEANUP_COUNT).subquery())\n    ).delete(synchronize_session=False)\n    models.db.session.commit()\n    logger.info(\"Deleted %d unused query results.\", deleted_count)\n\n\ndef remove_ghost_locks():\n    \"\"\"\n    Removes query locks that reference a non existing RQ job.\n    \"\"\"\n    keys = redis_connection.keys(\"query_hash_job:*\")\n    locks = {k: redis_connection.get(k) for k in keys}\n    jobs = list(rq_job_ids())\n\n    count = 0\n\n    for lock, job_id in locks.items():\n        if job_id not in jobs:\n            redis_connection.delete(lock)\n            count += 1\n\n    logger.info(\"Locks found: {}, Locks removed: {}\".format(len(locks), count))\n\n\n@job(\"schemas\")\ndef refresh_schema(data_source_id):\n    ds = models.DataSource.get_by_id(data_source_id)\n    logger.info(\"task=refresh_schema state=start ds_id=%s\", ds.id)\n    start_time = time.time()\n    try:\n        ds.get_schema(refresh=True)\n        logger.info(\n            \"task=refresh_schema state=finished ds_id=%s runtime=%.2f\",\n            ds.id,\n            time.time() - start_time,\n        )\n        statsd_client.incr(\"refresh_schema.success\")\n    except JobTimeoutException:\n        logger.info(\n            \"task=refresh_schema state=timeout ds_id=%s runtime=%.2f\",\n            ds.id,\n            time.time() - start_time,\n        )\n        statsd_client.incr(\"refresh_schema.timeout\")\n    except Exception:\n        logger.warning(\"Failed refreshing schema for the data source: %s\", ds.name, exc_info=1)\n        statsd_client.incr(\"refresh_schema.error\")\n        logger.info(\n            \"task=refresh_schema state=failed ds_id=%s runtime=%.2f\",\n            ds.id,\n            time.time() - start_time,\n        )\n\n\ndef refresh_schemas():\n    \"\"\"\n    Refreshes the data sources schemas.\n    \"\"\"\n    blacklist = [int(ds_id) for ds_id in redis_connection.smembers(\"data_sources:schema:blacklist\") if ds_id]\n    global_start_time = time.time()\n\n    logger.info(\"task=refresh_schemas state=start\")\n\n    for ds in models.DataSource.query:\n        if ds.paused:\n            logger.info(\n                \"task=refresh_schema state=skip ds_id=%s reason=paused(%s)\",\n                ds.id,\n                ds.pause_reason,\n            )\n        elif ds.id in blacklist:\n            logger.info(\"task=refresh_schema state=skip ds_id=%s reason=blacklist\", ds.id)\n        elif ds.org.is_disabled:\n            logger.info(\"task=refresh_schema state=skip ds_id=%s reason=org_disabled\", ds.id)\n        else:\n            refresh_schema.delay(ds.id)\n\n    logger.info(\n        \"task=refresh_schemas state=finish total_runtime=%.2f\",\n        time.time() - global_start_time,\n    )\n", "path": "redash/tasks/queries/maintenance.py"}]}