Action

datastore_search(original_action, context, data_dict)

This chains the datastore_search action provided by the vds extension so that we can provide an "include_urls" parameter to users which, if included on a search of the specimen collection, adds a "permanentUrl" key and value to each record dict in the result.

Source code in ckanext/nhm/logic/action.py

@toolkit.chained_action
@toolkit.side_effect_free
def datastore_search(original_action, context, data_dict):
    """
    This chains the datastore_search action provided by the vds extension so that we can
    provide an "include_urls" parameter to users which, if included on a search of the
    specimen collection, adds a "permanentUrl" key and value to each record dict in the
    result.
    """
    # check for include_urls and the resource_id before calling the original action just
    # to make sure we get a view of them before the vds action potentially changes them
    should_add_urls = (
        'include_urls' in data_dict
        and helpers.get_specimen_resource_id() == data_dict.get('resource_id')
    )

    # call vds action
    result = original_action(context, data_dict)

    if should_add_urls:
        for record in result['records']:
            if 'occurrenceID' in record:
                record['permanentUrl'] = toolkit.url_for(
                    'object_view', uuid=record['occurrenceID']
                )

    return result

get_permanent_url(context, data_dict)

Retrieve the permanent URL of a specimen from the specimen collection using the field and value to filter the results (i.e. field must equal value for the record to match). A URL is returned only if exactly one record is matched by the field and value combination. If more than 1 record is matched or if 0 records are matched then an error is returned.

Params:

Parameters:

Name	Type	Description	Default
field	string	the name of the field you would like to filter the records on	required
value	string	the value of the field to filter by	required
include_version	boolean **Results:**	whether to include the version in the permanent URL (default: false)	required

Returns:

Type	Description
string	the full URL of the specimen

Source code in ckanext/nhm/logic/action.py

@toolkit.side_effect_free
def get_permanent_url(context, data_dict):
    """
    Retrieve the permanent URL of a specimen from the specimen collection using the
    field and value to filter the results (i.e. field must equal value for the record to
    match). A URL is returned only if exactly one record is matched by the field and
    value combination. If more than 1 record is matched or if 0 records are matched then
    an error is returned.

    **Params:**

    :param field: the name of the field you would like to filter the records on
    :type field: string
    :param value: the value of the field to filter by
    :type value: string
    :param include_version: whether to include the version in the permanent URL
        (default: false)
    :type include_version: boolean  **Results:**
    :returns: the full URL of the specimen
    :rtype: string
    """
    schema = context.get('schema', nhm_schema.get_permanent_url_schema())
    data_dict, errors = toolkit.navl_validate(data_dict, schema, context)

    # extract the request parameters
    field = data_dict['field']
    value = data_dict['value']
    include_version = data_dict.get('include_version', False)

    # create a search dict to use with the datastore_search action
    search_dict = {
        'resource_id': helpers.get_specimen_resource_id(),
        'filters': {field: value},
        'limit': 1,
    }
    result = toolkit.get_action('datastore_search')(context, search_dict)
    records = result['records']
    total = result['total']
    if total == 0:
        raise toolkit.ValidationError(
            {
                'message': 'No records found matching the given criteria',
                'total': total,
            }
        )
    elif total > 1:
        raise toolkit.ValidationError(
            {
                'message': 'More than 1 record found matching the given criteria',
                'total': total,
            }
        )
    else:
        uuid = records[0]['occurrenceID']
        if include_version:
            # figure out the latest rounded version of the specimen resource data
            version = toolkit.get_action('vds_version_round')(
                context, {'resource_id': helpers.get_specimen_resource_id()}
            )
            # create a path with the version included
            path = toolkit.url_for('object_view_versioned', uuid=uuid, version=version)
        else:
            path = toolkit.url_for('object_view', uuid=uuid)

        # concatenate the path with the site url and return
        return f'{toolkit.config.get("ckan.site_url")}{path}'

object_rdf(context, data_dict)

Get record RDF.

Parameters:

Name	Type	Description	Default
context			required
data_dict			required

Source code in ckanext/nhm/logic/action.py

def object_rdf(context, data_dict):
    """
    Get record RDF.

    :param context:
    :param data_dict:
    """

    # Validate the data
    context = {'user': toolkit.c.user or toolkit.c.author}
    schema = context.get('schema', nhm_schema.object_rdf_schema())
    data_dict, errors = toolkit.navl_validate(data_dict, schema, context)
    # Raise any validation errors
    if errors:
        raise toolkit.ValidationError(errors)

    # get the record
    version = data_dict.get('version', None)
    record = get_record_by_uuid(data_dict['uuid'], version)
    if record:
        serializer = ObjectSerializer()
        output = serializer.serialize_record(record, data_dict.get('format'), version)
        return output
    raise toolkit.ObjectNotFound

show_extension_versions(context, data_dict)

Find all the installed extension packages and return their names and versions.

Returns:

Type	Description
Dict[str, str]	a dict of extension package name -> version

Source code in ckanext/nhm/logic/action.py

@toolkit.side_effect_free
def show_extension_versions(context, data_dict) -> Dict[str, str]:
    """
    Find all the installed extension packages and return their names and versions.

    :returns: a dict of extension package name -> version
    """
    dists = []

    for distribution in distributions():
        # ckan will show up in this list and will pass the lower tests, ignore it
        if distribution.name == 'ckan':
            continue
        # loop through the entry points in the package and if they have a ckan plugin
        # entry, and it's loaded, add it to the extensions dict
        for entry_point in distribution.entry_points:
            if entry_point.group == 'ckan.plugins' and plugin_loaded(entry_point.name):
                dists.append(distribution)
                # break on the first matching entry point as that's enough to confirm
                # that the package as a whole should be in the list
                break

    # sort the result alphabetically
    return {dist.name: dist.version for dist in sorted(dists, key=lambda d: d.name)}