Skip to content

Helpers

get_available_formats()

Simply returns all the formats that we can ingest.

Returns:

Type Description

a list of formats
Source code in ckanext/versioned_datastore/helpers.py 
110
111
112
113
114
115
116
def get_available_formats():
    """
    Simply returns all the formats that we can ingest.

    :returns: a list of formats
    """
    return ALL_FORMATS

get_human_duration(stat)

Get the duration on the passed ImportStats object in a sensible human readable format. The duration on stats objects is in seconds and therefore is great for small values but horrendous if it took 20 minutes. The output from this function is a string with either the number of seconds (to 2 decimal places), the number of minutes (to 0 decimal places) or the number of hours (to 0 decimal places).

Parameters:

Name Type Description Default
stat

the ImportStats object

 required

Returns:

Type Description

a nicely formatted duration string
Source code in ckanext/versioned_datastore/helpers.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def get_human_duration(stat):
    """
    Get the duration on the passed ImportStats object in a sensible human readable
    format. The duration on stats objects is in seconds and therefore is great for small
    values but horrendous if it took 20 minutes. The output from this function is a
    string with either the number of seconds (to 2 decimal places), the number of
    minutes (to 0 decimal places) or the number of hours (to 0 decimal places).

    :param stat: the ImportStats object
    :returns: a nicely formatted duration string
    """
    if stat.duration < 60:
        return toolkit._(f'{stat.duration:.2f} seconds')
    elif stat.duration < 60 * 60:
        return toolkit._(f'{stat.duration / 60:.0f} minutes')
    else:
        return toolkit._(f'{stat.duration / (60 * 60):.0f} hours')

get_stat_activity_class(stat)

Returns the activity css class to use for the given ImportStats object. The return value is a string css class which either matches one of the activity item classes from core ckan or matches one of the custom ones in this extension's css.

Parameters:

Name Type Description Default
stat

the ImportStats object

 required

Returns:

Type Description

a string
Source code in ckanext/versioned_datastore/helpers.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_stat_activity_class(stat):
    """
    Returns the activity css class to use for the given ImportStats object. The return
    value is a string css class which either matches one of the activity item classes
    from core ckan or matches one of the custom ones in this extension's css.

    :param stat: the ImportStats object
    :returns: a string
    """
    if stat.in_progress:
        return 'in_progress'
    if stat.error:
        if is_duplicate_ingestion(stat):
            return 'duplicate'
        return 'failure'
    return stat.type

get_stat_icon(stat)

Returns the fontawesome icon class(-es) to be used for the given ImportStats object. The return value is based on the type of stat and other factors like whether the operation the stat represents is still in progress.

Parameters:

Name Type Description Default
stat

the ImportStats object

 required

Returns:

Type Description

the fontawesome icon classes to use, as a string
Source code in ckanext/versioned_datastore/helpers.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_stat_icon(stat):
    """
    Returns the fontawesome icon class(-es) to be used for the given ImportStats object.
    The return value is based on the type of stat and other factors like whether the
    operation the stat represents is still in progress.

    :param stat: the ImportStats object
    :returns: the fontawesome icon classes to use, as a string
    """
    if stat.in_progress:
        # a spinner, that spins
        return 'fa-spinner fa-pulse'
    if stat.error:
        if is_duplicate_ingestion(stat):
            # we don't want this to look like an error
            return 'fa-copy'
        return 'fa-exclamation'

    if stat.type == stats.INGEST:
        return 'fa-tasks'
    if stat.type == stats.INDEX:
        return 'fa-search'
    if stat.type == stats.PREP:
        return 'fa-cogs'
    # shouldn't get here, just use some default tick thing
    return 'fa-check'

get_stat_title(stat)

Returns the title to use for the activity item created for the given ImportStats object. This is based on the stat's type.

Parameters:

Name Type Description Default
stat

the ImportStats object

 required

Returns:

Type Description

the title for the activity item as a unicode string
Source code in ckanext/versioned_datastore/helpers.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def get_stat_title(stat):
    """
    Returns the title to use for the activity item created for the given ImportStats
    object. This is based on the stat's type.

    :param stat: the ImportStats object
    :returns: the title for the activity item as a unicode string
    """
    if stat.type == stats.INGEST:
        return toolkit._('Ingested new resource data')
    if stat.type == stats.INDEX:
        return toolkit._('Updated search index with resource data')
    if stat.type == stats.PREP:
        return toolkit._('Validated and prepared the data for ingestion')
    return stat.type

get_version_date(version)

Returns a date object from a version number.

Parameters:

Name Type Description Default
version

a resource/record version number (i.e. a timestamp in ms)

 required

Returns:

Type Description

a date object
Source code in ckanext/versioned_datastore/helpers.py 
129
130
131
132
133
134
135
136
def get_version_date(version):
    """
    Returns a date object from a version number.

    :param version: a resource/record version number (i.e. a timestamp in ms)
    :returns: a date object
    """
    return date.fromtimestamp(int(version) / 1000)

is_duplicate_ingestion(stat)

Detects whether the error message on this ImportStats object is a DuplicateDataSource error message or not. This is based on the error containing the phrase "this file has been ingested before" and is therefore a quite fragile, but it's only for UI purposes so it's not he end of the world if it produces a few false positives/negatives.

Parameters:

Name Type Description Default
stat

the ImportStats object

 required

Returns:

Type Description

True if the error on this stat is a duplicate ingestion error, False if not
Source code in ckanext/versioned_datastore/helpers.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
def is_duplicate_ingestion(stat):
    """
    Detects whether the error message on this ImportStats object is a
    DuplicateDataSource error message or not. This is based on the error containing the
    phrase "this file has been ingested before" and is therefore a quite fragile, but
    it's only for UI purposes so it's not he end of the world if it produces a few false
    positives/negatives.

    :param stat: the ImportStats object
    :returns: True if the error on this stat is a duplicate ingestion error, False if
        not
    """
    return stat.error and 'this file has been ingested before' in stat.error.lower()

latest_item_version(resource_id, record_id=None)

Returns the most recent version for the given resource or record.

Parameters:

Name Type Description Default
resource_id

the id of the resource to search in

 required
record_id

optional; a record id to search for instead

 None

Returns:

Type Description

if record id is provided, the latest record version, else the latest resource version
Source code in ckanext/versioned_datastore/helpers.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def latest_item_version(resource_id, record_id=None):
    """
    Returns the most recent version for the given resource or record.

    :param resource_id: the id of the resource to search in
    :param record_id: optional; a record id to search for instead
    :returns: if record id is provided, the latest record version, else the latest
        resource version
    """
    if record_id:
        action = 'vds_version_record'
        data_dict = {'resource_id': resource_id, 'record_id': record_id}
    else:
        action = 'vds_version_resource'
        data_dict = {'resource_id': resource_id}

    result = toolkit.get_action(action)({}, data_dict)

    if record_id:
        # we get back a list of versions in ascending order
        return result[-1]
    else:
        # we get back a dict of versions and counts
        return max(result.keys())

multisearch(query=None)

A helper proxy for convert_to_multisearch.

Source code in ckanext/versioned_datastore/helpers.py 
173
174
175
176
177
178
def multisearch(query=None):
    """
    A helper proxy for convert_to_multisearch.
    """
    multisearch = convert_to_multisearch(query)
    return multisearch

nav_slug(query=None, version=None, resource_ids=None)

Just a helper proxy for create_nav_slug.

Source code in ckanext/versioned_datastore/helpers.py 
165
166
167
168
169
170
def nav_slug(query=None, version=None, resource_ids=None):
    """
    Just a helper proxy for create_nav_slug.
    """
    is_new, slug = create_nav_slug(SchemaQuery(resource_ids, version, query))
    return slug.get_slug_string()

pretty_print_json(json_string)

Does what you'd expect really.

Parameters:

Name Type Description Default
json_string

a json string

 required

Returns:

Type Description

a string of pretty json
Source code in ckanext/versioned_datastore/helpers.py 
119
120
121
122
123
124
125
126
def pretty_print_json(json_string):
    """
    Does what you'd expect really.

    :param json_string: a json string
    :returns: a string of pretty json
    """
    return json.dumps(json_string, sort_keys=True, indent=2)