[Draft] Proof of concept for xloader site url

ckanext/xloader/action.py

@@ -140,17 +140,18 @@ def xloader_submit(context, data_dict):
         qualified=True
     )
     data = {
-        'api_key': utils.get_xloader_user_apitoken(),
-        'job_type': 'xloader_to_datastore',
-        'result_url': callback_url,
-        'metadata': {
-            'ignore_hash': data_dict.get('ignore_hash', False),
-            'ckan_url': config['ckan.site_url'],
-            'resource_id': res_id,
-            'set_url_type': data_dict.get('set_url_type', False),
-            'task_created': task['last_updated'],
-            'original_url': resource_dict.get('url'),
-        }
+        "api_key": utils.get_xloader_user_apitoken(),
+        "job_type": "xloader_to_datastore",
+        "result_url": callback_url,
+        "metadata": {
+            "ignore_hash": data_dict.get("ignore_hash", False),
+            "ckan_url": config.get("ckanext.xloader.site_url")
+            or config["ckan.site_url"],
+            "resource_id": res_id,
+            "set_url_type": data_dict.get("set_url_type", False),
+            "task_created": task["last_updated"],
+            "original_url": resource_dict.get("url"),
+        },
     }
     if custom_queue != rq_jobs.DEFAULT_QUEUE_NAME:
         # Don't automatically retry if it's a custom run

ckanext/xloader/command.py

@@ -114,12 +114,11 @@ def _submit_resource(self, resource, user, indent=0, sync=False, queue=None):
             'ignore_hash': True,
         }
         if sync:
-            data_dict['ckan_url'] = tk.config.get('ckan.site_url')
-            input_dict = {
-                'metadata': data_dict,
-                'api_key': 'TODO'
-            }
-            logger = logging.getLogger('ckanext.xloader.cli')
+            data_dict["ckan_url"] = tk.config.get(
+                "ckanext.xloader.site_url"
+            ) or tk.config.get("ckan.site_url")
+            input_dict = {"metadata": data_dict, "api_key": "TODO"}
+            logger = logging.getLogger("ckanext.xloader.cli")
             xloader_data_into_datastore_(input_dict, None, logger)
         else:
             if queue:

ckanext/xloader/config_declaration.yaml

@@ -2,6 +2,14 @@ version: 1
 groups:
   - annotation: ckanext-xloader settings
     options:
+      - key: ckanext.xloader.site_url
+        example: http://ckan-dev:5000
+        default:
+        description: |
+            Provide an alternate site URL for the xloader_submit action.
+            This is useful, for example, when the site is running within a docker network.
+        validators: configured_default("ckan.site_url",None)
+        required: false
       - key: ckanext.xloader.jobs_db.uri
         default: sqlite:////tmp/xloader_jobs.db
         description: |
@@ -152,5 +160,3 @@ groups:
           they will also display "complete", "active", "inactive", and "unknown".
         type: bool
         required: false
-
-

ckanext/xloader/jobs.py

@@ -17,12 +17,14 @@
 from rq import get_current_job
 import sqlalchemy as sa
 
+from urllib.parse import urljoin, urlunsplit
+
 from ckan import model
 from ckan.plugins.toolkit import get_action, asbool, enqueue_job, ObjectNotFound, config
 
 from . import db, loader
 from .job_exceptions import JobError, HTTPError, DataTooBigError, FileCouldNotBeLoadedError
-from .utils import datastore_resource_exists, set_resource_metadata
+from .utils import datastore_resource_exists, set_resource_metadata, get_ckan_url
 
 try:
     from ckan.lib.api_token import get_user_from_token
@@ -79,9 +81,14 @@ def xloader_data_into_datastore(input):
     # First flag that this task is running, to indicate the job is not
     # stillborn, for when xloader_submit is deciding whether another job would
     # be a duplicate or not
+
+    callback_url = get_ckan_url()
+    callback_url = urljoin(
+        callback_url.rstrip('/'), '/api/3/action/xloader_hook')
+
     job_dict = dict(metadata=input['metadata'],
                     status='running')
-    callback_xloader_hook(result_url=input['result_url'],
+    callback_xloader_hook(result_url=callback_url,
                           api_key=input['api_key'],
                           job_dict=job_dict)
 
@@ -143,7 +150,7 @@ def xloader_data_into_datastore(input):
         errored = True
     finally:
         # job_dict is defined in xloader_hook's docstring
-        is_saved_ok = callback_xloader_hook(result_url=input['result_url'],
+        is_saved_ok = callback_xloader_hook(result_url=callback_url,
                                             api_key=input['api_key'],
                                             job_dict=job_dict)
         errored = errored or not is_saved_ok
@@ -204,7 +211,10 @@ def direct_load():
         set_datastore_active(data, resource, logger)
         if 'result_url' in input:
             job_dict['status'] = 'running_but_viewable'
-            callback_xloader_hook(result_url=input['result_url'],
+            callback_url = get_ckan_url()
+            callback_url = urljoin(
+                callback_url.rstrip('/'), '/api/3/action/xloader_hook')
+            callback_xloader_hook(result_url=callback_url,
                                   api_key=api_key,
                                   job_dict=job_dict)
         logger.info('Data now available to users: %s', resource_ckan_url)
@@ -285,6 +295,13 @@ def _download_resource_data(resource, data, api_key, logger):
             'Only http, https, and ftp resources may be fetched.'
         )
 
+    resource_uri = urlunsplit(('', '', url_parts.path, url_parts.query, url_parts.fragment))
+    callback_url = get_ckan_url()
+    url = urljoin(
+        callback_url.rstrip('/'), resource_uri)
+
+    url_parts = urlsplit(url) # reparse the url after the callback_url is set
+
     # fetch the resource data
     logger.info('Fetching from: {0}'.format(url))
     tmp_file = get_tmp_file(url)

ckanext/xloader/plugin.py

@@ -61,13 +61,11 @@ def configure(self, config_):
         else:
             self.ignore_hash = False
 
-        for config_option in ("ckan.site_url",):
-            if not config_.get(config_option):
-                raise Exception(
-                    "Config option `{0}` must be set to use ckanext-xloader.".format(
-                        config_option
-                    )
-                )
+        site_url_configs = ("ckan.site_url", "ckanext.xloader.site_url")
+        if not any(site_url_configs):
+            raise Exception(
+                f"One of config options {site_url_configs} must be set to use ckanext-xloader."
+            )
 
     # IDomainObjectModification
 

ckanext/xloader/tests/test_jobs.py

@@ -59,17 +59,18 @@ def data(create_with_upload, apikey):
         "api.action", ver=3, logic_function="xloader_hook", qualified=True
     )
     return {
-        'api_key': apikey,
-        'job_type': 'xloader_to_datastore',
-        'result_url': callback_url,
-        'metadata': {
-            'ignore_hash': True,
-            'ckan_url': toolkit.config.get('ckan.site_url'),
-            'resource_id': resource["id"],
-            'set_url_type': False,
-            'task_created': datetime.utcnow().isoformat(),
-            'original_url': resource["url"],
-        }
+        "api_key": apikey,
+        "job_type": "xloader_to_datastore",
+        "result_url": callback_url,
+        "metadata": {
+            "ignore_hash": True,
+            "ckan_url": toolkit.config.get("ckanext.xloader.site_url")
+            or toolkit.config.get("ckan.site_url"),
+            "resource_id": resource["id"],
+            "set_url_type": False,
+            "task_created": datetime.utcnow().isoformat(),
+            "original_url": resource["url"],
+        },
     }
 
 

ckanext/xloader/utils.py

@@ -107,6 +107,28 @@ def get_xloader_user_apitoken():
     return site_user["apikey"]
 
 
+def get_ckan_url():
+    """ Returns the CKAN URL.
+
+    ckan may be behind a proxy, or more likely, within a docker network.
+    This method returns the URL set in the config file for the CKAN instance.
+    Containers within the same network ie: XLoader will be able to communicate with CKAN using this URL.
+    """
+    ckan_url = config.get('ckanext.xloader.site_url', None)
+    if ckan_url:
+        return ckan_url
+
+    # Fall back to mandatory ckan.site_url
+    ckan_url = config.get('ckan.site_url')
+    if not ckan_url:
+        raise ValueError(
+            "The ckan.site_url configuration option is required but not set. "
+            "Please set this value in your CKAN configuration file."
+        )
+
+    return ckan_url
+
+
 def set_resource_metadata(update_dict):
     '''
     Set appropriate datastore_active flag on CKAN resource.