Plugin

VersionedSearchPlugin

Bases: SingletonPlugin

Source code in ckanext/versioned_datastore/plugin.py

class VersionedSearchPlugin(SingletonPlugin):
    implements(interfaces.IActions)
    implements(interfaces.IAuthFunctions)
    implements(interfaces.ITemplateHelpers, inherit=True)
    implements(interfaces.IResourceController, inherit=True)
    implements(interfaces.IConfigurer)
    implements(interfaces.IConfigurable)
    implements(interfaces.IBlueprint, inherit=True)
    implements(IVersionedDatastoreQuerySchema)
    implements(interfaces.IClick)
    implements(interfaces.IPackageController, inherit=True)
    if status_available:
        implements(IStatus)

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # all these are set up during the configure method
        self.mongo_client: Optional[MongoClient] = None
        self.elasticsearch_client: Optional[Elasticsearch] = None
        self.sg_client: Optional[SplitgillClient] = None

    # IConfigurable
    def configure(self, ckan_config):
        # do the setup for Splitgill first, create Mongo client, Elasticsearch client,
        # and then the Splitgill client
        self.mongo_client = MongoClient(
            host=ckan_config.get('ckanext.versioned_datastore.mongo_host'),
            port=int(ckan_config.get('ckanext.versioned_datastore.mongo_port')),
        )
        es_hosts = ckan_config.get(
            'ckanext.versioned_datastore.elasticsearch_hosts'
        ).split(',')
        es_port = ckan_config.get('ckanext.versioned_datastore.elasticsearch_port')
        self.elasticsearch_client = Elasticsearch(
            hosts=[f'http://{host}:{es_port}/' for host in es_hosts],
            # todo: check these params
            sniff_on_start=True,
            sniff_on_node_failure=True,
            sniff_timeout=30,
            http_compress=False,
            request_timeout=60,
        )
        mongo_db_name = ckan_config.get('ckanext.versioned_datastore.mongo_database')
        self.sg_client = SplitgillClient(
            self.mongo_client, self.elasticsearch_client, mongo_db_name
        )

        # register all custom query schemas
        for plugin in utils.iqs_implementations():
            for query_version, query_schema in plugin.get_query_schemas():
                register_schema(query_version, query_schema)

        # reserve any requested slugs
        from .lib.query.slugs.slugs import reserve_slug

        for plugin in utils.ivds_implementations():
            slugs = plugin.vds_reserve_slugs()
            for reserved_pretty_slug, query_parameters in slugs.items():
                query = SchemaQuery(**query_parameters)
                with suppress(Exception):
                    reserve_slug(reserved_pretty_slug, query)

        # configure cache
        configure_cache(ckan_config, 'versioned_datastore', 'vds')

    def is_sg_configured(self) -> bool:
        """
        Returns whether Splitgill is configured and ready for use. This checks if the
        Mongo client, Elasticsearch client, and Splitgill client have all been created
        and stored against this plugin instance.

        :returns: True if it's ready, False if not
        """
        return (
            self.mongo_client is not None
            and self.elasticsearch_client is not None
            and self.sg_client is not None
        )

    # IActions
    def get_actions(self):
        return create_actions(
            basic_action,
            data_action,
            download_action,
            multi_action,
            options_action,
            resource_action,
            schema_action,
            slug_action,
            version_action,
        )

    # IAuthFunctions
    def get_auth_functions(self):
        return create_auth(
            basic_auth,
            data_auth,
            download_auth,
            multi_auth,
            options_auth,
            resource_auth,
            schema_auth,
            slug_auth,
            version_auth,
        )

    # IClick
    def get_commands(self):
        return cli.get_commands()

    # ITemplateHelpers
    def get_helpers(self):
        return {
            'is_datastore_resource': utils.is_datastore_resource,
            'is_duplicate_ingestion': helpers.is_duplicate_ingestion,
            'get_human_duration': helpers.get_human_duration,
            'get_stat_icon': helpers.get_stat_icon,
            'get_stat_activity_class': helpers.get_stat_activity_class,
            'get_stat_title': helpers.get_stat_title,
            'get_available_formats': helpers.get_available_formats,
            'pretty_print_json': helpers.pretty_print_json,
            'get_version_date': helpers.get_version_date,
            'latest_item_version': helpers.latest_item_version,
            'nav_slug': helpers.nav_slug,
            'multisearch': helpers.multisearch,
        }

    # IResourceController (<2.10 compatibility)
    def before_create(self, context, resource):
        # only in IResourceController, but renamed
        self.before_resource_create(context, resource)

    # IResourceController and IPackageController (<2.10 compatibility)
    def after_create(self, context, data_dict=None, resource=None, pkg_dict=None):
        data_dict = data_dict or resource or pkg_dict
        is_resource = (resource is not None) or ('resource_type' in data_dict)
        if is_resource:
            self.after_resource_create(context, data_dict)
        else:
            self.after_dataset_create(context, data_dict)

    # IResourceController (<2.10 compatibility)
    def before_show(self, resource_dict):
        # only in IResourceController, but renamed
        self.before_resource_show(resource_dict)

    # IResourceController (<2.10 compatibility)
    def before_update(self, context, current, resource):
        # only in IResourceController, but renamed
        self.before_resource_update(context, current, resource)

    # IResourceController and IPackageController (<2.10 compatibility)
    def after_update(self, context, data_dict=None, resource=None, pkg_dict=None):
        data_dict = data_dict or resource or pkg_dict
        is_resource = (resource is not None) or ('resource_type' in data_dict)
        if is_resource:
            self.after_resource_update(context, data_dict)
        else:
            self.after_dataset_update(context, data_dict)

    # IResourceController (<2.10 compatibility)
    def before_delete(self, context: dict, resource: dict, resources: List[dict]):
        # only in IResourceController, but renamed
        self.before_resource_delete(context, resource, resources)

    # IResourceController and IPackageController (<2.10 compatibility)
    def after_delete(self, context, data_obj=None, resources=None, pkg_dict=None):
        data_obj = data_obj or resources or pkg_dict
        is_resource = (resources is not None) or isinstance(data_obj, list)
        if is_resource:
            self.after_resource_delete(context, data_obj)
        else:
            self.after_dataset_delete(context, data_obj)

    # IResourceController
    def before_resource_create(self, context, resource):
        # only set disable_parsing if it's True (False by default)
        if toolkit.asbool(resource.pop('disable_parsing', False)):
            resource['disable_parsing'] = True
        # same for download_original_filenames
        if toolkit.asbool(resource.pop('download_original_filenames', False)):
            resource['download_original_filenames'] = True

    # IResourceController
    def after_resource_create(self, context: dict, resource: dict):
        # use replace to overwrite the existing data (this is what users would expect)
        data_dict = {'resource_id': resource['id'], 'replace': True}
        with suppress(utils.ReadOnlyResourceException), suppress(
            utils.RawResourceException
        ):
            toolkit.get_action('vds_data_add')(context, data_dict)
        try:
            clear_cache_region('versioned_datastore', utils, cache_name='vds')
        except CacheClearError as e:
            log.error(e)

    # IResourceController
    def before_resource_show(self, resource_dict):
        # ensure datastore_active is set where it should be
        resource_dict['datastore_active'] = utils.is_datastore_resource(
            resource_dict['id']
        )
        # theoretically a resource could be datastore_active and have parsing disabled
        # at the same time if the database and ES have gotten out of sync, which isn't
        # ideal, but the fixes are more annoying than the problem itself
        return resource_dict

    # IResourceController
    def before_resource_update(self, context, current, resource):
        # we can't automatically go from ingested to raw because it might be included
        # in queries and DOIs, but we can ingest a file that was previously raw
        was_raw = toolkit.asbool(current.get('disable_parsing', False))
        is_raw = toolkit.asbool(resource.pop('disable_parsing', False))
        if was_raw and is_raw:
            # only allow if it was already previously raw
            resource['disable_parsing'] = True
        # download_original_filenames is fine; we can change that whenever
        if toolkit.asbool(resource.pop('download_original_filenames', False)):
            resource['download_original_filenames'] = True

    # IResourceController
    def after_resource_update(self, context: dict, resource: dict):
        # use replace to overwrite the existing data (this is what users would expect)
        data_dict = {'resource_id': resource['id'], 'replace': True}
        with suppress(utils.ReadOnlyResourceException), suppress(
            utils.RawResourceException
        ):
            toolkit.get_action('vds_data_add')(context, data_dict)
        try:
            clear_cache_region('versioned_datastore', utils, cache_name='vds')
        except CacheClearError as e:
            log.error(e)

    # IResourceController
    def before_resource_delete(
        self, context: dict, resource: dict, resources: List[dict]
    ):
        toolkit.get_action('vds_data_delete')(context, {'resource_id': resource['id']})

    # IResourceController
    def after_resource_delete(self, context: dict, resources: List[dict]):
        try:
            clear_cache_region('versioned_datastore', utils, cache_name='vds')
        except CacheClearError as e:
            log.error(e)

    # IPackageController
    def after_dataset_create(self, context: dict, pkg_dict: dict):
        try:
            clear_cache_region('versioned_datastore', utils, cache_name='vds')
        except CacheClearError as e:
            log.error(e)

    # IPackageController
    def after_dataset_update(self, context: dict, pkg_dict: dict):
        try:
            clear_cache_region('versioned_datastore', utils, cache_name='vds')
        except CacheClearError as e:
            log.error(e)

    # IPackageController
    def after_dataset_delete(self, context: dict, pkg_dict: dict):
        try:
            clear_cache_region('versioned_datastore', utils, cache_name='vds')
        except CacheClearError as e:
            log.error(e)

    # IConfigurer
    def update_config(self, config):
        # add public folder containing schemas
        toolkit.add_public_directory(config, 'theme/public')
        # add templates
        toolkit.add_template_directory(config, 'theme/templates')
        toolkit.add_resource('theme/assets', 'ckanext-versioned-datastore')

    # IBlueprint
    def get_blueprint(self):
        return routes.blueprints

    # IVersionedDatastoreQuerySchema
    def get_query_schemas(self):
        return [(v1_0_0Schema.version, v1_0_0Schema())]

    # IStatus
    def modify_status_reports(self, status_reports):
        queued_downloads = get_queue_length('download')

        status_reports.append(
            {
                'label': toolkit._('Downloads'),
                'value': queued_downloads,
                'group': toolkit._('Queues'),
                'help': toolkit._('Number of downloads waiting in the queue'),
                'state': 'good'
                if queued_downloads < 2
                else ('ok' if queued_downloads < 4 else 'bad'),
            }
        )

        queued_imports = get_queue_length('importing')

        status_reports.append(
            {
                'label': toolkit._('Imports'),
                'value': queued_imports,
                'group': toolkit._('Queues'),
                'help': toolkit._('Number of import jobs waiting in the queue'),
                'state': 'good'
                if queued_imports < 2
                else ('ok' if queued_imports < 4 else 'bad'),
            }
        )

        es_health = get_es_health()
        server_status_text = (
            toolkit._('available') if es_health['ping'] else toolkit._('unavailable')
        )

        status_reports.append(
            {
                'label': toolkit._('Search'),
                'value': server_status_text,
                'help': toolkit._(
                    'Multisearch functionality is provided by an Elasticsearch cluster'
                ),
                'state': 'good' if es_health['ping'] else 'bad',
            }
        )

        return status_reports

is_sg_configured()

Returns whether Splitgill is configured and ready for use. This checks if the Mongo client, Elasticsearch client, and Splitgill client have all been created and stored against this plugin instance.

Returns:

Type	Description
bool	True if it's ready, False if not

Source code in ckanext/versioned_datastore/plugin.py

def is_sg_configured(self) -> bool:
    """
    Returns whether Splitgill is configured and ready for use. This checks if the
    Mongo client, Elasticsearch client, and Splitgill client have all been created
    and stored against this plugin instance.

    :returns: True if it's ready, False if not
    """
    return (
        self.mongo_client is not None
        and self.elasticsearch_client is not None
        and self.sg_client is not None
    )