feat(dev): csv file to solr;

JVickery-TBS · JVickery-TBS · commit 1cfecd101541 · 2025-03-18T19:10:25.000Z
- Data pusher plugin support with direct csv to solr.
diff --git a/README.md b/README.md
@@ -97,7 +97,7 @@ To install ckanext-datastore-search:
 	# (optional, default: False).
 	ckanext.datastore_search.only_use_engine = True
 
-**ckanext.datastore_search.min_rows_for_index** sets the minimum number of database rows required to create and use a search index (mutually exclusive with only_use_engine):
+**ckanext.datastore_search.min_rows_for_index** sets the minimum number of database rows required to create and use a search index (mutually exclusive with `only_use_engine`):
 
 	# (optional, default: 100,000).
 	ckanext.datastore_search.min_rows_for_index = 40000
@@ -122,6 +122,21 @@ To install ckanext-datastore-search:
 	# (optional, default: datastore_resource).
 	ckanext.datastore_search.solr.configset = ckan_ds_resource
 
+**ckanext.datastore_search.download_proxy_address** defines the NETLOC address to download files from. This is for XLoader/DataPusher plugin integrations. This is only used when using `only_use_engine`.
+
+	# (optional, default: None).
+	ckanext.datastore_search.download_proxy_address = 127.0.0.1:5001
+
+**ckanext.datastore_search.download_verify_https** specifies whether or not to verify HTTPS when downloading files. This is for XLoader/DataPusher plugin integrations. This is only used when using `only_use_engine`.
+
+	# (optional, default: true).
+	ckanext.datastore_search.download_verify_https = false
+
+**ckanext.datastore_search.always_reupload_file** specifies whether or not to always ignore matching file hashes when downloading files. This is for XLoader/DataPusher plugin integrations.
+
+	# (optional, default: false).
+	ckanext.datastore_search.always_reupload_file = true
+
 ## Tests
 
 To run the tests, do:
diff --git a/ckanext/datastore_search/backend/__init__.py b/ckanext/datastore_search/backend/__init__.py
@@ -22,6 +22,7 @@ class DatastoreSearchBackend:
 
     _backends = {}
     _active_backend: "DatastoreSearchBackend"
+    supported_file_formats = ['csv', 'tsv']
     url = plugins.toolkit.config.get(
         'ckanext.datastore_search.url')
     prefix = plugins.toolkit.config.get(
@@ -38,6 +39,14 @@ class DatastoreSearchBackend:
             'ckanext.datastore_search.only_use_engine', False))
     min_rows_for_index = int(plugins.toolkit.config.get(
         'ckanext.datastore_search.min_rows_for_index', 100000))
+    download_proxy_address = plugins.toolkit.config.get(
+        'ckanext.datastore_search.download_proxy_address')
+    download_verify_https = plugins.toolkit.asbool(
+        plugins.toolkit.config.get(
+            'ckanext.datastore_search.download_verify_https', True))
+    always_reupload_file = plugins.toolkit.asbool(
+        plugins.toolkit.config.get(
+            'ckanext.datastore_search.always_reupload_file', False))
 
     @classmethod
     def register_backends(cls):
diff --git a/ckanext/datastore_search/backend/solr.py b/ckanext/datastore_search/backend/solr.py
@@ -3,6 +3,11 @@
 import requests
 import re
 from logging import getLogger
+from urllib.parse import urlsplit
+import time
+import hashlib
+import tempfile
+import math
 
 from typing import Any, Optional, Dict, cast, List
 from ckan.types import Context, DataDict
@@ -12,6 +17,7 @@
 
 from ckanext.datastore.logic.action import datastore_search_sql
 from ckanext.datastore.backend.postgres import identifier
+from ckanext.datastore.blueprint import dump_to
 
 from ckanext.datastore_search.backend import (
     DatastoreSearchBackend,
@@ -21,6 +27,8 @@
 
 MAX_ERR_LEN = 1000
 PSQL_TO_SOLR_WILCARD_MATCH = re.compile('^_?|_?$')
+CHUNK_SIZE = 16 * 1024
+DOWNLOAD_TIMEOUT = 30
 
 log = getLogger(__name__)
 DEBUG = config.get('debug', False)
@@ -182,7 +190,8 @@ def reindex(self,
             # type_ignore_reason: checking solr_records
             indexed_ids += [r['_id'] for r in solr_records]  # type: ignore
             if self.only_use_engine:
-                indexed_records += solr_records
+                # type_ignore_reason: checking solr_records
+                indexed_records += solr_records  # type: ignore
             offset += 1000
         core_name = f'{self.prefix}{resource_id}'
         errmsg = _('Failed to reindex records for %s' % core_name)
@@ -193,16 +202,16 @@ def reindex(self,
             where_statement = 'WHERE _id NOT IN ({indexed_ids})'.format(
                 indexed_ids=','.join(indexed_ids)) if \
                 only_missing and indexed_ids and \
-                    int(solr_total) <= int(ds_total) else ''
+                int(solr_total) <= int(ds_total) else ''
             existing_ids = []
             while gathering_ds_records:
                 sql_string = '''
                     SELECT {columns} FROM {table} {where_statement}
                     LIMIT 1000 OFFSET {offset}
                 '''.format(columns=ds_columns,
-                        table=table_name,
-                        where_statement=where_statement,
-                        offset=offset)
+                           table=table_name,
+                           where_statement=where_statement,
+                           offset=offset)
                 ds_result = datastore_search_sql(
                     context, {'sql': sql_string})
                 if not ds_result['records']:
@@ -283,12 +292,12 @@ def create(self,
         Create or update & reload/reindex a core if the fields have changed.
         """
         resource_id = data_dict.get('resource_id')
+        upload_file = data_dict.pop('upload_file', False)
         core_name = f'{self.prefix}{resource_id}'
         conn = self._make_connection(resource_id) if not connection else connection
         if not conn:
             errmsg = _('Could not create SOLR core %s') % core_name
             callback_queue = add_queue_name_prefix(self.redis_callback_queue_name)
-            # TODO: add callback_extras to pass records if self.only_use_engine
             enqueue_job(
                 # type_ignore_reason: incomplete typing
                 fn='solr_utils.create_solr_core.proc.create_solr_core',  # type: ignore
@@ -299,8 +308,11 @@ def create(self,
                                    'action.datastore_search_create_callback',
                     'callback_queue': callback_queue,
                     'callback_timeout': config.get('ckan.jobs.timeout', 300),
-                    'callback_extras': {'records': data_dict.get('records', None)}
-                    if self.only_use_engine else None},
+                    'callback_extras': {
+                        'records': data_dict.get('records', None)
+                        if self.only_use_engine else None,
+                        'upload_file': upload_file}
+                    },
                 title='SOLR Core creation %s' % core_name,
                 queue=self.redis_queue_name,
                 rq_kwargs={'timeout': 60})
@@ -335,6 +347,7 @@ def create(self,
                     'name': ds_field['id'],
                     'type': self.field_type_map[ds_field['type']],
                     'stored': True,
+                    'multiValued': False,
                     'indexed': True})
                 continue
             if self.field_type_map[ds_field['type']] == \
@@ -400,6 +413,99 @@ def create(self,
             log.debug('Removed SOLR Field %s for DataStore Resource %s' %
                       (f['name'], resource_id))
 
+        if upload_file:
+            res_dict = get_action('resource_show')(
+                {'ignore_auth': True}, {'id': resource_id})
+            if res_dict.get('format', '').lower() not in self.supported_file_formats:
+                raise DatastoreSearchException('Unsupported file format')
+            url = res_dict.get('url')
+            url_parts = urlsplit(url)
+            scheme = url_parts.scheme
+            if scheme not in ('http', 'https', 'ftp'):
+                raise DatastoreSearchException('Only http, https, and ftp '
+                                               'resources may be fetched.')
+            download_uri = url
+            headers = {}
+            site_user = get_action('get_site_user')({'ignore_auth': True}, {})
+            if res_dict.get('url_type') == 'upload':
+                headers['Authorization'] = site_user['apikey']
+                download_uri = url_parts._replace(
+                    query='{}&nonce={}'.format(url_parts.query, time.time()),
+                    netloc=(self.download_proxy_address or url_parts.netloc)).geturl()
+            log.debug('Fetching from: %s' % download_uri)
+            tmp_file = tempfile.TemporaryFile()
+            length = 0
+            m = hashlib.md5()
+            errmsg = _('Could not load file to SOLR core %s') % core_name
+            try:
+                if self.only_use_engine:
+                    response = requests.get(
+                        download_uri,
+                        verify=self.download_verify_https,
+                        stream=True,
+                        headers=headers,
+                        timeout=DOWNLOAD_TIMEOUT)
+                    response.raise_for_status()
+                    for chunk in response.iter_content(CHUNK_SIZE):
+                        length += len(chunk)
+                        tmp_file.write(chunk)
+                        m.update(chunk)
+                else:
+                    # type_ignore_reason: incomplete typing
+                    for chunk in dump_to(resource_id,  # type: ignore
+                                         fmt=res_dict['format'].lower(),
+                                         offset=0,
+                                         limit=None,
+                                         options={'bom': False},
+                                         sort='_id',
+                                         search_params={'skip_search_engine': True},
+                                         user=site_user['name']):
+                        length += len(chunk)
+                        tmp_file.write(chunk)
+                        m.update(chunk)
+                human_length = '0 bytes'
+                if length != 0:
+                    size_name = ('bytes', 'KB', 'MB', 'GB', 'TB')
+                    i = int(math.floor(math.log(length, 1024)))
+                    p = math.pow(1024, i)
+                    s = round(float(length) / p, 1)
+                    human_length = "%s %s" % (s, size_name[i])
+                log.debug('Downloaded ok - %s', human_length)
+                file_hash = m.hexdigest()
+                tmp_file.seek(0)
+                if not self.always_reupload_file and res_dict.get('hash') == file_hash:
+                    log.debug('The file hash has not changed, skipping loading into '
+                              'SOLR index for DataStore Resource %s' % resource_id)
+                    return
+                log.debug('Emptying SOLR index for DataStore Resource %s' %
+                          resource_id)
+                conn.delete(q='*:*', commit=False)
+                log.debug('Unindexed all DataStore records for DataStore Resource %s' %
+                          resource_id)
+                log.debug('Uploading file to SOLR core %s for DataStore Resource %s' %
+                          (core_name, resource_id))
+                separator = ','
+                if res_dict['format'].lower() == 'tsv':
+                    separator = '%09'
+                solr_response = json.loads(conn._send_request(
+                    method='POST',
+                    path='update?commit=false&trim=true&overwrite=false'
+                    '&header=true&separator=%s' % separator,
+                    headers={'Content-Type': 'application/csv'},
+                    body=tmp_file.read()))
+                if solr_response.get('errors'):
+                    raise DatastoreSearchException(
+                        errmsg if not DEBUG else solr_response.get('errors'))
+            except pysolr.SolrError as e:
+                raise DatastoreSearchException(
+                    errmsg if not DEBUG else e.args[0][:MAX_ERR_LEN])
+            except Exception as e:
+                raise DatastoreSearchException(e)
+            finally:
+                tmp_file.close()
+            conn.commit(waitSearcher=False)
+            return
+
         if new_fields or updated_fields or remove_fields:
             self.reindex(resource_id, connection=conn)
 
@@ -422,6 +528,7 @@ def create_callback(self, data_dict: DataDict) -> Any:
             log.debug('SOLR core creation stderr: %s' % data_dict.get('stderr'))
 
         resource_id = data_dict.get('core_name', '').replace(self.prefix, '')
+        upload_file = data_dict.get('extras', {}).get('upload_file', False)
         context = self._get_site_context()
         ds_result = get_action('datastore_search')(
             context, {'resource_id': resource_id,
@@ -430,10 +537,17 @@ def create_callback(self, data_dict: DataDict) -> Any:
         create_dict = {
             'resource_id': resource_id,
             'fields': [f for f in ds_result['fields'] if
-                       f['id'] not in self.default_search_fields]}
+                       f['id'] not in self.default_search_fields],
+            'upload_file': upload_file}
+        # call create again to get datastore fields and data types to build schema
         self.create(create_dict)
 
-        if self.only_use_engine and data_dict.get('extras', {}).get('records'):
+        if (
+            not upload_file and
+            self.only_use_engine and
+            data_dict.get('extras', {}).get('records')
+        ):
+            # use datastore_upsert with insert to be able to use dry_run to get _id
             get_action('datastore_upsert')(
                 context, {'resource_id': resource_id,
                           'records': data_dict.get('extras', {}).get('records'),
@@ -534,7 +648,6 @@ def delete(self,
         """
         Removes records from the SOLR index, or deletes the core entirely.
         """
-        # FIXME: delete core if count less than min_rows_for_index??
         resource_id = data_dict.get('resource_id')
         core_name = f'{self.prefix}{resource_id}'
         conn = self._make_connection(resource_id) if not connection else connection
@@ -547,7 +660,7 @@ def delete(self,
             errmsg = _('Could not delete SOLR core %s') % core_name
             try:
                 conn.delete(q='*:*', commit=False)
-                log.debug('Unindexed all DataStore records for Resource %s' %
+                log.debug('Unindexed all DataStore records for DataStore Resource %s' %
                           resource_id)
             except pysolr.SolrError as e:
                 raise DatastoreSearchException(
@@ -564,16 +677,17 @@ def delete(self,
             return
 
         if self.only_use_engine:
+            # will not have deleted_records returned, so need to do this by SOLR query
             fq = []
             for key, value in data_dict.get('filters', {}).items():
                 fq.append('%s:%s' % (key, value))
             errmsg = _('Could not delete DataStore record(s) '
-                       'q=%s in SOLR core %s') % \
-                        (' AND '.join(fq), core_name)
+                       'q=%s in SOLR core %s') % (' AND '.join(fq), core_name)
             collecting_deleted_records = True
             offset = 0
             deleted_records = []
             try:
+                # do a search before deleting to emulate deleted_records
                 while collecting_deleted_records:
                     results = conn.search(q=' AND '.join(fq),
                                           start=offset,
diff --git a/ckanext/datastore_search/config_declaration.yaml b/ckanext/datastore_search/config_declaration.yaml
@@ -24,12 +24,13 @@ groups:
         required: false
       - key: ckanext.datastore_search.only_use_engine
         default: false
+        type: bool
         description: |
           Only use the search engine, never inserting records into the database.
           Because the DataDictionary saves data types and field descriptions into
           the database column comments, the database table and fields will still
           be created to maintain this functionality.
-        validators: ignore_missing
+        validators: ignore_missing boolean_validator
         required: false
       - key: ckanext.datastore_search.min_rows_for_index
         default: 100000
@@ -60,3 +61,29 @@ groups:
         validators: ignore_missing
         example: ckan_ds_resource
         required: false
+      - key: ckanext.datastore_search.download_proxy_address
+        default: None
+        description: |
+          NETLOC address for an internal proxy to download files.
+          This is only used in XLoader & DataPusher integration when using ckanext.datastore_search.only_use_engine.
+        validators: ignore_missing
+        example: 127.0.0.1:5001
+        required: false
+      - key: ckanext.datastore_search.download_verify_https
+        default: true
+        type: bool
+        description: |
+          Whether or not to verify HTTPS to download files.
+          This is only used in XLoader & DataPusher integration when using ckanext.datastore_search.only_use_engine.
+        validators: ignore_missing boolean_validator
+        example: false
+        required: false
+      - key: ckanext.datastore_search.always_reupload_file
+        default: false
+        type: bool
+        description: |
+          Whether or not to always ignore file hashes.
+          This is only used in XLoader & DataPusher integration.
+        validators: ignore_missing boolean_validator
+        example: true
+        required: false
diff --git a/ckanext/datastore_search/logic/action.py b/ckanext/datastore_search/logic/action.py
diff --git a/ckanext/datastore_search/plugin.py b/ckanext/datastore_search/plugin.py
