50962731-9dd3-4918-891e-919091c5ac26

Committed 13 Jan 2025 11:06AM UTC coverage: 85.94% (+0.01%) from 85.929%

Build # 50962731-9dd3-4918-891e-919091c5ac26

Build Type

push

circleci

Committed by

JoshBone

Commit Message

Correcting swagger field display

Run Details

10 of 10 new or added lines in 2 files covered. (100.0%)

8 existing lines in 2 files now uncovered.

9847 of 11458 relevant lines covered (85.94%)

0.86 hits per line

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

98.86

/submissionapi/csv_functions/csv_handler.py

import csv

import itertools
import re

from submissionapi.csv_functions.csv_insensitive_dict_reader import DictReaderInsensitive


class CSVHandler:
    """
        Class to handle CSV upload, transform it to a submission request object
    """
    FIELDS = {
        'reports': [
            r'agency',
            r'contributing_agencies\[\d+\]',
            r'report_id',
            r'local_identifier',
            r'status',
            r'decision',
            r'summary',
            r'valid_from',
            r'valid_to',
            r'date_format',
            r'other_comment'
        ],
        'activities': [
            r'activities\[\d+\]\.activity',
            r'activities\[\d+\]\.local_identifier',
            r'activities\[\d+\]\.agency',
        ],
        'report_links': [
            r'link\[\d+\]',
            r'link_display_name\[\d+\]'
        ],
        'report_files': [
            r'file\[\d+\]\.original_location',
            r'file\[\d+\]\.display_name',
        ],
        'report_files__report_language': [
            r'file\[\d+\]\.report_language\[\d+\]',
        ],
        'institutions': [
            r'institution\[\d+\]\.deqar_id',
            r'institution\[\d+\]\.eter_id',
            r'institution\[\d+\]\.identifier',
            r'institution\[\d+\]\.resource'
        ],
        'institutions__alternative_names': [
            r'institution\[\d+\]\.name_alternative\[\d+\]',
            r'institution\[\d+\]\.name_alternative_transliterated\[\d+\]',
        ],
        'institutions__locations': [
            r'institution\[\d+\]\.country\[\d+\]',
            r'institution\[\d+\]\.city\[\d+\]',
            r'institution\[\d+\]\.latitude\[\d+\]',
            r'institution\[\d+\]\.longitude\[\d+\]',
        ],
        'institutions__qf_ehea_levels': [
            r'institution\[\d+\]\.qf_ehea_level\[\d+\]',
        ],
        'programmes': [
            r'programme\[\d+\]\.name_primary',
            r'programme\[\d+\]\.qualification_primary',
            r'programme\[\d+\]\.nqf_level',
            r'programme\[\d+\]\.qf_ehea_level',
            r'programme\[\d+\]\.degree_outcome',
            r'programme\[\d+\]\.workload_ects',
            r'programme\[\d+\]\.learning_outcome_description',
            r'programme\[\d+\]\.field_study',
            r'programme\[\d+\]\.assessment_certification',
        ],
        'programmes__identifiers': [
            r'programme\[\d+\]\.identifier\[\d+\]',
            r'programme\[\d+\]\.resource\[\d+\]',
        ],
        'programmes__alternative_names': [
            r'programme\[\d+\]\.name_alternative\[\d+\]',
            r'programme\[\d+\]\.qualification_alternative\[\d+\]',
        ],
        'programmes__countries': [
            r'programme\[\d+\]\.country\[\d+\]',
        ],
        'programmes__learning_outcomes': [
            r'programme\[\d+\]\.learning_outcome\[\d+\]',
        ]
    }

    def __init__(self, csvfile):
        self.csvfile = csvfile
        self.submission_data = []
        self.report_record = {}
        self.error = False
        self.error_message = ""
        self.dialect = None
        self.reader = None

    def handle(self):
        if self._csv_is_valid():
            self._read_csv()
            for row in self.reader:
                self._create_report(row)
                self._create_activities(row)
                self._create_report_links(row)
                self._create_report_files(row)
                self._create_institutions(row)
                self._create_institutions_alternative_names(row)
                self._create_institutions_locations(row)
                self._create_institutions_qf_ehea_levels(row)
                self._create_programmes(row)
                self._create_programmes_alternative_names(row)
                self._create_programmes_identifiers(row)
                self._create_programmes_countries(row)
                self._create_learning_outcomes(row)
                self._clear_submission_data()
        else:
            self.error = True
            self.error_message = 'The CSV file appears to be invalid.'

    def _csv_is_valid(self):
        try:
            self.csvfile.seek(0)
            self.dialect = csv.Sniffer().sniff(self.csvfile.read(), delimiters=['\t', ',', ';'])
            return True
        except csv.Error:
            return False

    def _read_csv(self):
        self.csvfile.seek(0)
        self.reader = DictReaderInsensitive(self.csvfile)

    def _create_report(self, row):
        csv_fields = self.reader.fieldnames
        for field in self.FIELDS['reports']:
            r = re.compile(field)
            rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)

            if len(rematch) > 0:
                if 'contributing_agencies' in field:
                    self.report_record['contributing_agencies'] = []
                    for column in rematch:
                        self.report_record['contributing_agencies'].append(row[column])
                else:
                    self.report_record[rematch[0]] = row[rematch[0]]

    def _create_activities(self, row):
        self._create_first_level_placeholder(['activities'])
        self._create_first_level_values('activities', row, dotted=True)

    def _create_institutions(self, row):
        self._create_first_level_placeholder(['institutions',
                                              'institutions__alternative_names',
                                              'institutions__locations',
                                              'institutions__qf_ehea_levels'])
        self._create_first_level_values('institutions', row, dotted=True)

    def _create_institutions_alternative_names(self, row):
        self._create_second_level_placeholder('institutions__alternative_names', dictkey=True)
        self._create_second_level_values('institutions__alternative_names', row, dictkey=True)

    def _create_institutions_locations(self, row):
        self._create_second_level_placeholder('institutions__locations', dictkey=True)
        self._create_second_level_values('institutions__locations', row, dictkey=True)

    def _create_institutions_qf_ehea_levels(self, row):
        self._create_second_level_placeholder('institutions__qf_ehea_levels')
        self._create_second_level_values('institutions__qf_ehea_levels', row)

    def _create_programmes(self, row):
        self._create_first_level_placeholder(['programmes',
                                              'programmes__identifiers',
                                              'programmes__alternative_names',
                                              'programmes__countries',
                                              'programmes__learning_outcomes'])
        self._create_first_level_values('programmes', row, dotted=True)

    def _create_programmes_identifiers(self, row):
        self._create_second_level_placeholder('programmes__identifiers', dictkey=True)
        self._create_second_level_values('programmes__identifiers', row, dictkey=True)

    def _create_programmes_alternative_names(self, row):
        self._create_second_level_placeholder('programmes__alternative_names', dictkey=True)
        self._create_second_level_values('programmes__alternative_names', row, dictkey=True)

    def _create_programmes_countries(self, row):
        self._create_second_level_placeholder('programmes__countries')
        self._create_second_level_values('programmes__countries', row)

    def _create_learning_outcomes(self, row):
        self._create_second_level_placeholder('programmes__learning_outcomes')
        self._create_second_level_values('programmes__learning_outcomes', row)

    def _create_report_links(self, row):
        self._create_first_level_placeholder(['report_links'])
        self._create_first_level_values('report_links', row)

    def _create_report_files(self, row):
        self._create_first_level_placeholder(['report_files', 'report_files__report_language'])
        self._create_first_level_values('report_files', row, dotted=True)

        self._create_second_level_placeholder('report_files__report_language')
        self._create_second_level_values('report_files__report_language', row)

    def _create_first_level_placeholder(self, field_key_array):
        fields = []
        wrapper = field_key_array[0].split('__')[0]
        csv_fields = self.reader.fieldnames

        for fk in field_key_array:
            fields += self.FIELDS[fk]

        # Create wrapper
        self.report_record[wrapper] = []

        for field in fields:
            r = re.compile(field)

            max_index = 0
            for csv_field in csv_fields:
                match = r.match(csv_field)
                if match:
                    groups = re.search(r"\d+", csv_field)
                    index = int(groups.group(0))
                    if index > max_index:
                        max_index = index

            # Create plaholder if it doesn't exists yet
            if max_index > 0:
                for i in range(0, max_index):
                    if len(self.report_record[wrapper]) < i+1:
                        self.report_record[wrapper].append({})

    def _create_first_level_values(self, wrapper, row, dotted=False):
        csv_fields = self.reader.fieldnames
        for field in self.FIELDS[wrapper]:
            r = re.compile(field)
            rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)

            if len(rematch) > 0:
                for fld in rematch:
                    if row[fld] != '-':
                        index = re.search(r"\[\d+\]", fld).group()
                        field = fld.replace(index, "")
                        index = int(re.search(r"\d+", index).group())-1
                        if dotted:
                            field = field.split('.')[1]
                        self.report_record[wrapper][index][field] = row[fld]

    def _create_second_level_placeholder(self, field_key, dictkey=None):
        csv_fields = self.reader.fieldnames
        first_level_wrapper_name, wrapper = field_key.split('__')

        # Create second level wrapper
        first_level_wrapper = self.report_record[first_level_wrapper_name]

        for first_level_wrapper_item in first_level_wrapper:
            first_level_wrapper_item[wrapper] = []

            if dictkey:
                for field in self.FIELDS[field_key]:
                    r = re.compile(field)

                    max_index = 0
                    for csv_field in csv_fields:
                        match = r.match(csv_field)
                        if match:
                            groups = re.search(r"\d+", csv_field)
                            index = int(groups.group(0))
                            if index > max_index:
                                max_index = index

                    # Create plaholder if it doesn't exists yet
                    if max_index > 0:
                        for i in range(0, max_index):
                            if len(first_level_wrapper_item[wrapper]) < i+1:
                                first_level_wrapper_item[wrapper].append({})

    def _create_second_level_values(self, field_key, row, dictkey=None):
        csv_fields = self.reader.fieldnames
        first_level_wrapper_name, wrapper = field_key.split('__')
        first_level_wrapper = self.report_record[first_level_wrapper_name]

        for field in self.FIELDS[field_key]:
            r = re.compile(field)
            rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)

            if len(rematch) > 0:
                for fld in rematch:
                    if row[fld] != '-':
                        [field01, field02] = fld.split('.')
                        index01 = int(re.search(r"\d+", field01).group())

                        index02 = re.search(r"\[\d+\]", field02).group()
                        field02 = field02.replace(index02, "")
                        index02 = int(re.search(r"\d+", index02).group())
                        if dictkey:
                            first_level_wrapper[index01-1][wrapper][index02-1][field02] = row[fld]
                        else:
                            first_level_wrapper[index01-1][wrapper].append(row[fld])

    def _clear_submission_data(self):
        self.submission_data.append(self.clean_empty(self.report_record))

    def clean_empty(self, d):
        if not isinstance(d, (dict, list)):
            return d
        if isinstance(d, list):
            return [v for v in (self.clean_empty(v) for v in d) if v]
        return {k: v for k, v in ((k, self.clean_empty(v)) for k, v in d.items()) if v}

1	import csv	1✔
2
3	import itertools	1✔
4	import re	1✔
5
6	from submissionapi.csv_functions.csv_insensitive_dict_reader import DictReaderInsensitive	1✔
7
8
9	class CSVHandler:	1✔
10	"""
11	Class to handle CSV upload, transform it to a submission request object
12	"""
13	FIELDS = {	1✔
14	'reports': [
15	r'agency',
16	r'contributing_agencies\[\d+\]',
17	r'report_id',
18	r'local_identifier',
19	r'status',
20	r'decision',
21	r'summary',
22	r'valid_from',
23	r'valid_to',
24	r'date_format',
25	r'other_comment'
26	],
27	'activities': [
28	r'activities\[\d+\]\.activity',
29	r'activities\[\d+\]\.local_identifier',
30	r'activities\[\d+\]\.agency',
31	],
32	'report_links': [
33	r'link\[\d+\]',
34	r'link_display_name\[\d+\]'
35	],
36	'report_files': [
37	r'file\[\d+\]\.original_location',
38	r'file\[\d+\]\.display_name',
39	],
40	'report_files__report_language': [
41	r'file\[\d+\]\.report_language\[\d+\]',
42	],
43	'institutions': [
44	r'institution\[\d+\]\.deqar_id',
45	r'institution\[\d+\]\.eter_id',
46	r'institution\[\d+\]\.identifier',
47	r'institution\[\d+\]\.resource'
48	],
49	'institutions__alternative_names': [
50	r'institution\[\d+\]\.name_alternative\[\d+\]',
51	r'institution\[\d+\]\.name_alternative_transliterated\[\d+\]',
52	],
53	'institutions__locations': [
54	r'institution\[\d+\]\.country\[\d+\]',
55	r'institution\[\d+\]\.city\[\d+\]',
56	r'institution\[\d+\]\.latitude\[\d+\]',
57	r'institution\[\d+\]\.longitude\[\d+\]',
58	],
59	'institutions__qf_ehea_levels': [
60	r'institution\[\d+\]\.qf_ehea_level\[\d+\]',
61	],
62	'programmes': [
63	r'programme\[\d+\]\.name_primary',
64	r'programme\[\d+\]\.qualification_primary',
65	r'programme\[\d+\]\.nqf_level',
66	r'programme\[\d+\]\.qf_ehea_level',
67	r'programme\[\d+\]\.degree_outcome',
68	r'programme\[\d+\]\.workload_ects',
69	r'programme\[\d+\]\.learning_outcome_description',
70	r'programme\[\d+\]\.field_study',
71	r'programme\[\d+\]\.assessment_certification',
72	],
73	'programmes__identifiers': [
74	r'programme\[\d+\]\.identifier\[\d+\]',
75	r'programme\[\d+\]\.resource\[\d+\]',
76	],
77	'programmes__alternative_names': [
78	r'programme\[\d+\]\.name_alternative\[\d+\]',
79	r'programme\[\d+\]\.qualification_alternative\[\d+\]',
80	],
81	'programmes__countries': [
82	r'programme\[\d+\]\.country\[\d+\]',
83	],
84	'programmes__learning_outcomes': [
85	r'programme\[\d+\]\.learning_outcome\[\d+\]',
86	]
87	}
88
89	def __init__(self, csvfile):	1✔
90	self.csvfile = csvfile	1✔
91	self.submission_data = []	1✔
92	self.report_record = {}	1✔
93	self.error = False	1✔
94	self.error_message = ""	1✔
95	self.dialect = None	1✔
96	self.reader = None	1✔
97
98	def handle(self):	1✔
99	if self._csv_is_valid():	1✔
100	self._read_csv()	1✔
101	for row in self.reader:	1✔
102	self._create_report(row)	1✔
103	self._create_activities(row)	1✔
104	self._create_report_links(row)	1✔
105	self._create_report_files(row)	1✔
106	self._create_institutions(row)	1✔
107	self._create_institutions_alternative_names(row)	1✔
108	self._create_institutions_locations(row)	1✔
109	self._create_institutions_qf_ehea_levels(row)	1✔
110	self._create_programmes(row)	1✔
111	self._create_programmes_alternative_names(row)	1✔
112	self._create_programmes_identifiers(row)	1✔
113	self._create_programmes_countries(row)	1✔
114	self._create_learning_outcomes(row)	1✔
115	self._clear_submission_data()	1✔
116	else:
UNCOV 117	self.error = True	×
UNCOV 118	self.error_message = 'The CSV file appears to be invalid.'	×
119
120	def _csv_is_valid(self):	1✔
121	try:	1✔
122	self.csvfile.seek(0)	1✔
123	self.dialect = csv.Sniffer().sniff(self.csvfile.read(), delimiters=['\t', ',', ';'])	1✔
124	return True	1✔
125	except csv.Error:	1✔
126	return False	1✔
127
128	def _read_csv(self):	1✔
129	self.csvfile.seek(0)	1✔
130	self.reader = DictReaderInsensitive(self.csvfile)	1✔
131
132	def _create_report(self, row):	1✔
133	csv_fields = self.reader.fieldnames	1✔
134	for field in self.FIELDS['reports']:	1✔
135	r = re.compile(field)	1✔
136	rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)	1✔
137
138	if len(rematch) > 0:	1✔
139	if 'contributing_agencies' in field:	1✔
140	self.report_record['contributing_agencies'] = []	1✔
141	for column in rematch:	1✔
142	self.report_record['contributing_agencies'].append(row[column])	1✔
143	else:
144	self.report_record[rematch[0]] = row[rematch[0]]	1✔
145
146	def _create_activities(self, row):	1✔
147	self._create_first_level_placeholder(['activities'])	1✔
148	self._create_first_level_values('activities', row, dotted=True)	1✔
149
150	def _create_institutions(self, row):	1✔
151	self._create_first_level_placeholder(['institutions',	1✔
152	'institutions__alternative_names',
153	'institutions__locations',
154	'institutions__qf_ehea_levels'])
155	self._create_first_level_values('institutions', row, dotted=True)	1✔
156
157	def _create_institutions_alternative_names(self, row):	1✔
158	self._create_second_level_placeholder('institutions__alternative_names', dictkey=True)	1✔
159	self._create_second_level_values('institutions__alternative_names', row, dictkey=True)	1✔
160
161	def _create_institutions_locations(self, row):	1✔
162	self._create_second_level_placeholder('institutions__locations', dictkey=True)	1✔
163	self._create_second_level_values('institutions__locations', row, dictkey=True)	1✔
164
165	def _create_institutions_qf_ehea_levels(self, row):	1✔
166	self._create_second_level_placeholder('institutions__qf_ehea_levels')	1✔
167	self._create_second_level_values('institutions__qf_ehea_levels', row)	1✔
168
169	def _create_programmes(self, row):	1✔
170	self._create_first_level_placeholder(['programmes',	1✔
171	'programmes__identifiers',
172	'programmes__alternative_names',
173	'programmes__countries',
174	'programmes__learning_outcomes'])
175	self._create_first_level_values('programmes', row, dotted=True)	1✔
176
177	def _create_programmes_identifiers(self, row):	1✔
178	self._create_second_level_placeholder('programmes__identifiers', dictkey=True)	1✔
179	self._create_second_level_values('programmes__identifiers', row, dictkey=True)	1✔
180
181	def _create_programmes_alternative_names(self, row):	1✔
182	self._create_second_level_placeholder('programmes__alternative_names', dictkey=True)	1✔
183	self._create_second_level_values('programmes__alternative_names', row, dictkey=True)	1✔
184
185	def _create_programmes_countries(self, row):	1✔
186	self._create_second_level_placeholder('programmes__countries')	1✔
187	self._create_second_level_values('programmes__countries', row)	1✔
188
189	def _create_learning_outcomes(self, row):	1✔
190	self._create_second_level_placeholder('programmes__learning_outcomes')	1✔
191	self._create_second_level_values('programmes__learning_outcomes', row)	1✔
192
193	def _create_report_links(self, row):	1✔
194	self._create_first_level_placeholder(['report_links'])	1✔
195	self._create_first_level_values('report_links', row)	1✔
196
197	def _create_report_files(self, row):	1✔
198	self._create_first_level_placeholder(['report_files', 'report_files__report_language'])	1✔
199	self._create_first_level_values('report_files', row, dotted=True)	1✔
200
201	self._create_second_level_placeholder('report_files__report_language')	1✔
202	self._create_second_level_values('report_files__report_language', row)	1✔
203
204	def _create_first_level_placeholder(self, field_key_array):	1✔
205	fields = []	1✔
206	wrapper = field_key_array[0].split('__')[0]	1✔
207	csv_fields = self.reader.fieldnames	1✔
208
209	for fk in field_key_array:	1✔
210	fields += self.FIELDS[fk]	1✔
211
212	# Create wrapper
213	self.report_record[wrapper] = []	1✔
214
215	for field in fields:	1✔
216	r = re.compile(field)	1✔
217
218	max_index = 0	1✔
219	for csv_field in csv_fields:	1✔
220	match = r.match(csv_field)	1✔
221	if match:	1✔
222	groups = re.search(r"\d+", csv_field)	1✔
223	index = int(groups.group(0))	1✔
224	if index > max_index:	1✔
225	max_index = index	1✔
226
227	# Create plaholder if it doesn't exists yet
228	if max_index > 0:	1✔
229	for i in range(0, max_index):	1✔
230	if len(self.report_record[wrapper]) < i+1:	1✔
231	self.report_record[wrapper].append({})	1✔
232
233	def _create_first_level_values(self, wrapper, row, dotted=False):	1✔
234	csv_fields = self.reader.fieldnames	1✔
235	for field in self.FIELDS[wrapper]:	1✔
236	r = re.compile(field)	1✔
237	rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)	1✔
238
239	if len(rematch) > 0:	1✔
240	for fld in rematch:	1✔
241	if row[fld] != '-':	1✔
242	index = re.search(r"\[\d+\]", fld).group()	1✔
243	field = fld.replace(index, "")	1✔
244	index = int(re.search(r"\d+", index).group())-1	1✔
245	if dotted:	1✔
246	field = field.split('.')[1]	1✔
247	self.report_record[wrapper][index][field] = row[fld]	1✔
248
249	def _create_second_level_placeholder(self, field_key, dictkey=None):	1✔
250	csv_fields = self.reader.fieldnames	1✔
251	first_level_wrapper_name, wrapper = field_key.split('__')	1✔
252
253	# Create second level wrapper
254	first_level_wrapper = self.report_record[first_level_wrapper_name]	1✔
255
256	for first_level_wrapper_item in first_level_wrapper:	1✔
257	first_level_wrapper_item[wrapper] = []	1✔
258
259	if dictkey:	1✔
260	for field in self.FIELDS[field_key]:	1✔
261	r = re.compile(field)	1✔
262
263	max_index = 0	1✔
264	for csv_field in csv_fields:	1✔
265	match = r.match(csv_field)	1✔
266	if match:	1✔
267	groups = re.search(r"\d+", csv_field)	1✔
268	index = int(groups.group(0))	1✔
269	if index > max_index:	1✔
270	max_index = index	1✔
271
272	# Create plaholder if it doesn't exists yet
273	if max_index > 0:	1✔
274	for i in range(0, max_index):	1✔
275	if len(first_level_wrapper_item[wrapper]) < i+1:	1✔
276	first_level_wrapper_item[wrapper].append({})	1✔
277
278	def _create_second_level_values(self, field_key, row, dictkey=None):	1✔
279	csv_fields = self.reader.fieldnames	1✔
280	first_level_wrapper_name, wrapper = field_key.split('__')	1✔
281	first_level_wrapper = self.report_record[first_level_wrapper_name]	1✔
282
283	for field in self.FIELDS[field_key]:	1✔
284	r = re.compile(field)	1✔
285	rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)	1✔
286
287	if len(rematch) > 0:	1✔
288	for fld in rematch:	1✔
289	if row[fld] != '-':	1✔
290	[field01, field02] = fld.split('.')	1✔
291	index01 = int(re.search(r"\d+", field01).group())	1✔
292
293	index02 = re.search(r"\[\d+\]", field02).group()	1✔
294	field02 = field02.replace(index02, "")	1✔
295	index02 = int(re.search(r"\d+", index02).group())	1✔
296	if dictkey:	1✔
297	first_level_wrapper[index01-1][wrapper][index02-1][field02] = row[fld]	1✔
298	else:
299	first_level_wrapper[index01-1][wrapper].append(row[fld])	1✔
300
301	def _clear_submission_data(self):	1✔
302	self.submission_data.append(self.clean_empty(self.report_record))	1✔
303
304	def clean_empty(self, d):	1✔
305	if not isinstance(d, (dict, list)):	1✔
306	return d	1✔
307	if isinstance(d, list):	1✔
308	return [v for v in (self.clean_empty(v) for v in d) if v]	1✔
309	return {k: v for k, v in ((k, self.clean_empty(v)) for k, v in d.items()) if v}	1✔

EQAR / eqar_backend / 50962731-9dd3-4918-891e-919091c5ac26

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