NaturalHistoryMuseum / data-importer / #15

Committed 11 Aug 2025 01:38PM UTC coverage: 79.133% (-0.1%) from 79.237%

Build # #15

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

merge: #59 from dev

Run Details

196 of 294 new or added lines in 26 files covered. (66.67%)

5 existing lines in 2 files now uncovered.

1259 of 1591 relevant lines covered (79.13%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/dataimporter/cli/shell.py

from functools import partial
from typing import Union

from dataimporter.cli.utils import console
from dataimporter.importer import DataImporter


def setup_env(importer: DataImporter) -> dict:
    """
    Returns a dict of variables to made available in the maintenance shell.

    :param importer: a DataImporter instance
    :return: a dict
    """
    return {
        'importer': importer,
        'console': console,
        # some convenience functions for printing data
        'pcd': partial(print_record_data, importer, 'ecatalogue'),
        'ptd': partial(print_record_data, importer, 'etaxonomy'),
        'pmd': partial(print_record_data, importer, 'emultimedia'),
        'pgd': partial(print_record_data, importer, 'gbif'),
        'cm': partial(check_membership, importer),
    }


def print_record_data(importer: DataImporter, name: str, record_id: Union[str, int]):
    """
    A convenience function for printing record data from the given data store.

    :param importer: a DataImporter instance
    :param name: the store to look up the record ID in
    :param record_id: a record ID (can be int or str, we deal with it)
    """
    store = importer.get_store(name)
    if store is None:
        console.print('store not found', style='red')
        return
    record = store.get_record(str(record_id))
    if record is None:
        console.print('record not found', style='red')
        return
    console.print(record.data)


def check_membership(importer: DataImporter, name: str, record_id: Union[str, int]):
    """
    A convenience function which checks if the given record ID is a member of the given
    view. If it isn't, the reason why is printed.

    :param importer: a DataImporter instance
    :param name: name of the view
    :param record_id: record ID (can be int or str, we deal with it)
    """
    view = importer.get_view(name)
    if view is None:
        console.print('view not found', style='red')
        return
    record = view.store.get_record(str(record_id))
    if record is None:
        console.print('record not found', style='red')
        return
    result = view.is_member(record)
    if result:
        print(f'{record_id} is a member of {name}')
    else:
        print(f"{record_id} is not a member of {name} due to '{result.reason}'")

1	from functools import partial	×
2	from typing import Union	×
3
4	from dataimporter.cli.utils import console	×
5	from dataimporter.importer import DataImporter	×
6
7
8	def setup_env(importer: DataImporter) -> dict:	×
9	"""
10	Returns a dict of variables to made available in the maintenance shell.
11
12	:param importer: a DataImporter instance
13	:return: a dict
14	"""
15	return {	×
16	'importer': importer,
17	'console': console,
18	# some convenience functions for printing data
19	'pcd': partial(print_record_data, importer, 'ecatalogue'),
20	'ptd': partial(print_record_data, importer, 'etaxonomy'),
21	'pmd': partial(print_record_data, importer, 'emultimedia'),
22	'pgd': partial(print_record_data, importer, 'gbif'),
23	'cm': partial(check_membership, importer),
24	}
25
26
27	def print_record_data(importer: DataImporter, name: str, record_id: Union[str, int]):	×
28	"""
29	A convenience function for printing record data from the given data store.
30
31	:param importer: a DataImporter instance
32	:param name: the store to look up the record ID in
33	:param record_id: a record ID (can be int or str, we deal with it)
34	"""
35	store = importer.get_store(name)	×
36	if store is None:	×
NEW 37	console.print('store not found', style='red')	×
38	return	×
39	record = store.get_record(str(record_id))	×
40	if record is None:	×
NEW 41	console.print('record not found', style='red')	×
42	return	×
43	console.print(record.data)	×
44
45
46	def check_membership(importer: DataImporter, name: str, record_id: Union[str, int]):	×
47	"""
48	A convenience function which checks if the given record ID is a member of the given
49	view. If it isn't, the reason why is printed.
50
51	:param importer: a DataImporter instance
52	:param name: name of the view
53	:param record_id: record ID (can be int or str, we deal with it)
54	"""
55	view = importer.get_view(name)	×
56	if view is None:	×
NEW 57	console.print('view not found', style='red')	×
58	return	×
59	record = view.store.get_record(str(record_id))	×
60	if record is None:	×
NEW 61	console.print('record not found', style='red')	×
62	return	×
63	result = view.is_member(record)	×
64	if result:	×
NEW 65	print(f'{record_id} is a member of {name}')	×
66	else:
67	print(f"{record_id} is not a member of {name} due to '{result.reason}'")	×

NaturalHistoryMuseum / data-importer / #15

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous