Safely create_app in Celery code (try to fetch current_app first).

Closes #3181.

redash/__init__.py

@@ -3,7 +3,7 @@
 import urlparse
 import urllib
 import redis
-from flask import Flask
+from flask import Flask, current_app
 from flask_sslify import SSLify
 from werkzeug.contrib.fixers import ProxyFix
 from werkzeug.routing import BaseConverter
@@ -141,3 +141,11 @@ def create_app(load_admin=True):
     chrome_logger.init_app(app)
 
     return app
+
+
+def safe_create_app():
+    """Return current_app or create a new one."""
+    if current_app:
+        return current_app
+
+    return create_app()

redash/worker.py

@@ -8,7 +8,7 @@
 from celery import Celery
 from celery.schedules import crontab
 from celery.signals import worker_process_init
-from redash import __version__, create_app, settings
+from redash import __version__, safe_create_app, settings
 from redash.metrics import celery as celery_metrics
 
 celery = Celery('redash',
@@ -77,14 +77,14 @@ def __call__(self, *args, **kwargs):
 # Create Flask app after forking a new worker, to make sure no resources are shared between processes.
 @worker_process_init.connect
 def init_celery_flask_app(**kwargs):
-    app = create_app()
+    app = safe_create_app()
     app.app_context().push()
 
 
 # Hook for extensions to add periodic tasks.
 @celery.on_after_configure.connect
 def add_periodic_tasks(sender, **kwargs):
-    app = create_app()
+    app = safe_create_app()
     periodic_tasks = getattr(app, 'periodic_tasks', {})
     for params in periodic_tasks.values():
         sender.add_periodic_task(**params)