Webhook

hq_superset/tasks.py

@@ -1,12 +1,9 @@
-import logging
 import os
 
 from superset.extensions import celery_app
 
 from .utils import AsyncImportHelper, refresh_hq_datasource
 
-logger = logging.getLogger(__name__)
-
 
 @celery_app.task(name='refresh_hq_datasource_task')
 def refresh_hq_datasource_task(domain, datasource_id, display_name, export_path, datasource_defn, user_id):
@@ -16,34 +13,3 @@ def refresh_hq_datasource_task(domain, datasource_id, display_name, export_path,
         AsyncImportHelper(domain, datasource_id).mark_as_complete()
         raise
     os.remove(export_path)
-
-
-@celery_app.task(name='subscribe_to_hq_datasource_task')
-def subscribe_to_hq_datasource_task(domain, datasource_id):
-    from superset.config import BASE_URL
-    from hq_superset.hq_requests import HQRequest, HqUrl
-    from hq_superset.models import HQClient
-
-    if HQClient.get_by_domain(domain) is None:
-        hq_request = HQRequest(
-            url=HqUrl.subscribe_to_datasource_url(domain, datasource_id)
-        )
-
-        client_id, client_secret = HQClient.create_domain_client(domain)
-
-        response = hq_request.post({
-            'webhook_url': f'{BASE_URL}/hq_webhook/change/',
-            'token_url': f'{BASE_URL}/oauth/token',
-            'client_id': client_id,
-            'client_secret': client_secret,
-        })
-        if response.status_code == 201:
-            return
-        if response.status_code < 500:
-            logger.error(
-                f"Failed to subscribe to data source {datasource_id} due to the following issue: {response.data}"
-            )
-        if response.status_code >= 500:
-            logger.exception(
-                f"Failed to subscribe to data source {datasource_id} due to a remote server error"
-            )

hq_superset/utils.py

@@ -1,4 +1,5 @@
 import ast
+import logging
 import os
 from contextlib import contextmanager
 from datetime import date, datetime
@@ -19,6 +20,8 @@
 SESSION_OAUTH_RESPONSE_KEY = "oauth_response"
 ASYNC_DATASOURCE_IMPORT_LIMIT_IN_BYTES = 5_000_000  # ~5MB
 
+logger = logging.getLogger(__name__)
+
 
 def get_datasource_export_url(domain, datasource_id):
     return f"a/{domain}/configurable_reports/data_sources/export/{datasource_id}/?format=csv"
@@ -191,9 +194,7 @@ def get_datasource_file(path):
 
 def download_datasource(domain, datasource_id):
     import superset
-
     from hq_superset.hq_requests import HQRequest, HqUrl
-    from hq_superset.tasks import subscribe_to_hq_datasource_task
 
     hq_request = HQRequest(
         url=HqUrl.datasource_export_url(domain, datasource_id),
@@ -204,15 +205,42 @@ def download_datasource(domain, datasource_id):
 
     filename = f"{datasource_id}_{datetime.now()}.zip"
     path = os.path.join(superset.config.SHARED_DIR, filename)
-
     with open(path, "wb") as f:
         f.write(response.content)
 
-    subscribe_to_hq_datasource_task.delay(domain, datasource_id)
-
     return path, len(response.content)
 
 
+def subscribe_to_hq_datasource(domain, datasource_id):
+    from superset.config import BASE_URL
+    from hq_superset.hq_requests import HQRequest, HqUrl
+    from hq_superset.models import HQClient
+
+    if HQClient.get_by_domain(domain) is None:
+        hq_request = HQRequest(
+            url=HqUrl.subscribe_to_datasource_url(domain, datasource_id)
+        )
+
+        client_id, client_secret = HQClient.create_domain_client(domain)
+
+        response = hq_request.post({
+            'webhook_url': f'{BASE_URL}/hq_webhook/change/',
+            'token_url': f'{BASE_URL}/oauth/token',
+            'client_id': client_id,
+            'client_secret': client_secret,
+        })
+        if response.status_code == 201:
+            return
+        if response.status_code < 500:
+            logger.error(
+                f"Failed to subscribe to data source {datasource_id} due to the following issue: {response.data}"
+            )
+        if response.status_code >= 500:
+            logger.exception(
+                f"Failed to subscribe to data source {datasource_id} due to a remote server error"
+            )
+
+
 def get_datasource_defn(domain, datasource_id):
     from hq_superset.hq_requests import HQRequest, HqUrl
 

hq_superset/views.py

@@ -28,6 +28,7 @@
     get_hq_database,
     get_schema_name_for_domain,
     refresh_hq_datasource,
+    subscribe_to_hq_datasource,
 )
 
 logger = logging.getLogger(__name__)
@@ -120,6 +121,7 @@ def trigger_datasource_refresh(domain, datasource_id, display_name):
         )
         return redirect("/tablemodelview/list/")
 
+    subscribe_to_hq_datasource(domain, datasource_id)
     path, size = download_datasource(domain, datasource_id)
     datasource_defn = get_datasource_defn(domain, datasource_id)
     if size < ASYNC_DATASOURCE_IMPORT_LIMIT_IN_BYTES: