syntax

Author: JVickery-TBS

ckanext/datastore_search/backend/__init__.py

@@ -1,5 +1,5 @@
 from ckan.types import DataDict
-from typing import Any
+from typing import Any, Dict, Optional
 
 import ckan.plugins as plugins
 from ckan.common import CKANConfig
@@ -61,7 +61,7 @@ def get_active_backend(cls):
         return cls._active_backend
 
     @property
-    def field_type_map(self):
+    def field_type_map(self) -> Dict[str, str]:
         """
         Map of DataStore field types to their corresponding
         search index field types.
@@ -80,8 +80,8 @@ def configure(self, config: CKANConfig):
         return config
 
     def reindex(self,
-                resource_id: str,
-                connection: Any,
+                resource_id: Optional[str] = None,
+                connection: Optional[Any] = None,
                 only_missing: bool = False) -> Any:
         """Reindex/sync records between the database and the search engine.
         """

ckanext/datastore_search/backend/solr.py

@@ -79,10 +79,13 @@ def field_type_map(self):
             # TODO: map object/array types
         }
 
-    def _make_connection(self, resource_id: str) -> Optional[pysolr.Solr]:
+    def _make_connection(self,
+                         resource_id: Optional[str] = None) -> Optional[pysolr.Solr]:
         """
         Tries to make a SOLR connection to a core.
         """
+        if not resource_id:
+            return
         core_name = f'{self.prefix}{resource_id}'
         conn_string = f'{self.url}/solr/{core_name}'
         try:
@@ -122,20 +125,23 @@ def _get_site_context(self) -> Context:
         return cast(Context, {'user': site_user['name']})
 
     def reindex(self,
-                resource_id: str,
+                resource_id: Optional[str] = None,
                 connection: Optional[pysolr.Solr] = None,
                 only_missing: bool = False) -> Any:
         """
         Reindexes the SOLR core.
         """
-        #FIXME: put this in a background task as a larger
-        #       DS Resource could take a long time??
+        if not resource_id:
+            return
+        # FIXME: put this in a background task as a larger
+        #        DS Resource could take a long time??
         context = self._get_site_context()
         core_name = f'{self.prefix}{resource_id}'
         conn = self._make_connection(resource_id) if not connection else connection
 
         if not conn:
-            return
+            raise DatastoreSearchException(
+                _('SOLR core does not exist for DataStore Resource %s') % resource_id)
 
         errmsg = _('Could not reload SOLR core %s') % core_name
         resp = self._send_api_request(method='POST',
@@ -172,7 +178,8 @@ def reindex(self,
                 conn)
             if not solr_records:
                 gathering_solr_records = False
-            indexed_ids += [r['_id'] for r in solr_records]
+            # type_ignore_reason: checking solr_records
+            indexed_ids += [r['_id'] for r in solr_records]  # type: ignore
             offset += 1000
         gathering_ds_records = True
         offset = 0
@@ -222,25 +229,34 @@ def reindex(self,
         log.debug('Reindexed SOLR Core for DataStore Resource %s' % resource_id)
 
     def _check_counts(self,
-                      resource_id: str,
-                      connection: pysolr.Solr) -> Any:
+                      resource_id: Optional[str] = None,
+                      connection: Optional[pysolr.Solr] = None) -> Any:
         """
         Checks if the record counts match between the DataStore and SOLR.
         """
+        if not resource_id:
+            return
+
+        conn = self._make_connection(resource_id) if not connection else connection
+
+        if not conn:
+            raise DatastoreSearchException(
+                _('SOLR core does not exist for DataStore Resource %s') % resource_id)
+
         ds_result = get_action('datastore_search')(
             self._get_site_context(), {'resource_id': resource_id,
                                        'limit': 0,
                                        'include_total': True,
                                        'skip_search_engine': True})
         ds_total = ds_result['total']
-        solr_result = connection.search(q='*:*', rows=0)
+        solr_result = conn.search(q='*:*', rows=0)
         solr_total = solr_result.hits
 
         if int(ds_total) != int(solr_total):
             log.debug('SOLR (count: %s) and Postgres (count: %s) out of sync, '
                       'reindexing SOLR for DataStore Resource %s' %
                       (solr_total, ds_total, resource_id))
-            self.reindex(resource_id, connection, only_missing=True)
+            self.reindex(resource_id, connection=conn, only_missing=True)
 
     def create(self,
                data_dict: DataDict,
@@ -254,17 +270,19 @@ def create(self,
         if not conn:
             errmsg = _('Could not create SOLR core %s') % core_name
             callback_queue = add_queue_name_prefix(self.redis_callback_queue_name)
-            enqueue_job(fn='solr_utils.create_solr_core.proc.create_solr_core',
-                        kwargs={
-                            'core_name': core_name,
-                            'config_set': self.configset_name,
-                            'callback_fn': 'ckanext.datastore_search.logic.'
-                                           'action.datastore_search_create_callback',
-                            'callback_queue': callback_queue,
-                            'callback_timeout': config.get('ckan.jobs.timeout', 300)},
-                        title='SOLR Core creation %s' % core_name,
-                        queue=self.redis_queue_name,
-                        rq_kwargs={'timeout': 60})
+            enqueue_job(
+                # type_ignore_reason: incomplete typing
+                fn='solr_utils.create_solr_core.proc.create_solr_core',  # type: ignore
+                kwargs={
+                    'core_name': core_name,
+                    'config_set': self.configset_name,
+                    'callback_fn': 'ckanext.datastore_search.logic.'
+                                    'action.datastore_search_create_callback',
+                    'callback_queue': callback_queue,
+                    'callback_timeout': config.get('ckan.jobs.timeout', 300)},
+                title='SOLR Core creation %s' % core_name,
+                queue=self.redis_queue_name,
+                rq_kwargs={'timeout': 60})
             log.debug('Enqueued SOLR Core creation for DataStore Resource %s ' %
                       resource_id)
             return
@@ -276,6 +294,7 @@ def create(self,
             solr_fields = json.loads(conn._send_request(
                 method='GET', path='schema/fields'))['fields']
         except pysolr.SolrError as e:
+            errmsg = _('Could not get SOLR fields from core %s') % core_name
             raise DatastoreSearchException(
                 errmsg if not DEBUG else e.args[0][:MAX_ERR_LEN])
         keyed_solr_fields = {}
@@ -297,11 +316,13 @@ def create(self,
                     'stored': True,
                     'indexed': True})
                 continue
-            if self.field_type_map[ds_field['type']] == keyed_solr_fields[ds_field['id']]['type']:
+            if self.field_type_map[ds_field['type']] == \
+               keyed_solr_fields[ds_field['id']]['type']:
                 continue
             updated_fields.append(dict(keyed_solr_fields[ds_field['id']],
                                        type=self.field_type_map[ds_field['type']]))
-        for field_name in [i for i in keyed_solr_fields.keys() if i not in ds_field_ids]:
+        for field_name in [i for i in keyed_solr_fields.keys()
+                           if i not in ds_field_ids]:
             remove_fields.append({'name': field_name})
 
         for f in new_fields:
@@ -401,7 +422,8 @@ def upsert(self,
         conn = self._make_connection(resource_id) if not connection else connection
 
         if not conn:
-            return
+            raise DatastoreSearchException(
+                _('SOLR core does not exist for DataStore Resource %s') % resource_id)
 
         if data_dict['records']:
             for r in data_dict['records']:
@@ -420,7 +442,8 @@ def upsert(self,
 
     def search(self,
                data_dict: DataDict,
-               connection: Optional[pysolr.Solr] = None) -> Optional[List[Dict[str, Any]]]:
+               connection: Optional[pysolr.Solr] = None) \
+            -> Optional[List[Dict[str, Any]]]:
         """
         Searches the SOLR records.
         """
@@ -487,13 +510,15 @@ def delete(self,
         conn = self._make_connection(resource_id) if not connection else connection
 
         if not conn:
-            return
+            raise DatastoreSearchException(
+                _('SOLR core does not exist for DataStore Resource %s') % resource_id)
 
         if not data_dict.get('filters'):
             errmsg = _('Could not delete SOLR core %s') % core_name
             try:
                 conn.delete(q='*:*', commit=False)
-                log.debug('Unindexed all DataStore records for Resource %s' % resource_id)
+                log.debug('Unindexed all DataStore records for Resource %s' %
+                          resource_id)
             except pysolr.SolrError as e:
                 raise DatastoreSearchException(
                     errmsg if not DEBUG else e.args[0][:MAX_ERR_LEN])

ckanext/datastore_search/logic/action.py

@@ -1,3 +1,4 @@
+from typing import Dict, Any, Union
 from ckan.types import Context, DataDict, Action, ChainedAction
 
 from ckan.plugins import toolkit
@@ -64,7 +65,8 @@ def datastore_delete(up_func: Action,
 @toolkit.chained_action
 def datastore_search(up_func: Action,
                      context: Context,
-                     data_dict: DataDict) -> ChainedAction:
+                     data_dict: DataDict) -> Union[ChainedAction,
+                                                   Dict[str, Any]]:
     schema = context.get('schema', datastore_search_schema())
     schema['skip_search_engine'] = [ignore_missing, ignore_not_sysadmin]
     data_dict, errors = validate(data_dict, schema, context)

ckanext/datastore_search/plugin.py

@@ -1,7 +1,7 @@
 import ckan.plugins as plugins
 from ckan.common import CKANConfig
 
-from typing import Dict, Union
+from typing import Dict, Union, Type
 from ckan.types import (
     Context,
     DataDict,
@@ -16,10 +16,13 @@
 
 from ckanext.datapusher.interfaces import IDataPusher
 try:
-    from ckanext.xloader.interfaces import IXloader
-    HAS_XLOADER = True
+    # type_ignore_reason: catch ImportError
+    from ckanext.xloader.interfaces import IXloader  # type: ignore
+    # type_ignore_reason: not redifined if ImportError
+    HAS_XLOADER = True  # type: ignore
 except ImportError:
-    HAS_XLOADER = False
+    # type_ignore_reason: not redifined if ImportError
+    HAS_XLOADER = False  # type: ignore
 
 
 @plugins.toolkit.blanket.config_declarations
@@ -30,10 +33,11 @@ class DataStoreSearchPlugin(plugins.SingletonPlugin):
     plugins.implements(IDatastoreSearchBackend, inherit=True)
     plugins.implements(IDataPusher, inherit=True)
     if HAS_XLOADER:
-        plugins.implements(IXloader, inherit=True)
+        # type_ignore_reason: never unbound due to HAS_XLOADER
+        plugins.implements(IXloader, inherit=True)  # type: ignore
 
     # IDatastoreSearchBackend
-    def register_backends(self) -> Dict[str, DatastoreSearchBackend]:
+    def register_backends(self) -> Dict[str, Type[DatastoreSearchBackend]]:
         return {'solr': DatastoreSolrBackend}
 
     # IConfigurer
@@ -76,4 +80,4 @@ def after_upload(self,
             'resource_id': resource_dict.get('id'),
             'fields': [f for f in ds_result['fields'] if
                        f['id'] not in backend.default_search_fields]}
-        backend.create(context, create_dict)
+        backend.create(create_dict)