-
Notifications
You must be signed in to change notification settings - Fork 12
/
research.py
698 lines (529 loc) · 27.7 KB
/
research.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
"""Functions for the research space."""
__copyright__ = 'Copyright (c) 2019-2024, Utrecht University'
__license__ = 'GPLv3, see LICENSE'
from typing import Tuple
import genquery
from pathvalidate import validate_filename, validate_filepath, ValidationError
import folder
import groups
import meta_form
from util import *
__all__ = ['api_research_folder_add',
'api_research_folder_copy',
'api_research_folder_move',
'api_research_folder_delete',
'api_research_folder_rename',
'api_research_file_copy',
'api_research_file_rename',
'api_research_file_move',
'api_research_file_delete',
'api_research_system_metadata',
'api_research_collection_details',
'api_research_list_temporary_files',
'api_research_manifest']
def folder_new_name_check(folder_name: str) -> Tuple[bool, str]:
if len(folder_name) == 0:
return False, api.Error('missing_foldername', 'Missing folder name. Please add a folder name')
# TODO remove when upgrade to GenQuery 2
# This check should only be done on new folders, since may have old folders with apostrophes
if '\'' in folder_name:
return False, api.Error('invalid_foldername', 'It is not allowed to use apostrophes in a folder name')
# Name should not contain '\\' or '/'
if '/' in folder_name or '\\' in folder_name:
return False, api.Error('invalid_foldername', 'It is not allowed to use slashes in the new folder name')
# name should not be '.' or '..'
if folder_name in ('.', '..'):
return False, api.Error('invalid_foldername', 'it is not allowed to name the folder {}'.format(folder_name))
return True, ""
@api.make()
def api_research_folder_add(ctx: rule.Context, coll: str, new_folder_name: str) -> api.Result:
"""Add a new folder to a research folder.
:param ctx: Combined type of a callback and rei struct
:param coll: Collection to create new folder in
:param new_folder_name: Name of the new folder
:returns: API status result
"""
coll_target = coll + '/' + new_folder_name
valid_folder_name, error_response = folder_new_name_check(new_folder_name)
if not valid_folder_name:
return error_response
try:
validate_filepath(coll_target)
except ValidationError:
return api.Error('invalid_foldername', 'This is not a valid folder name. Please choose another name for your folder')
# not in home - a groupname must be present ie at least 2!?
if not len(coll.split('/')) > 2:
return api.Error('invalid_destination', 'It is not possible to add folder ' + new_folder_name + ' at this location')
# in vault?
target_group_name = coll_target.split('/')[3]
if target_group_name.startswith('vault-'):
return api.Error('not_allowed', 'It is not possible to add folders in the vault')
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return api.Error('not_allowed', 'You do not have sufficient permissions to add new folders')
# Collection exists?
if not collection.exists(ctx, coll):
return api.Error('invalid_foldername', 'The selected folder to add a new folder to does not exist')
# Folder not locked?
if folder.is_locked(ctx, coll):
return api.Error('not_allowed', 'The indicated folder is locked so no new folders can be added to it')
# new collection exists?
if collection.exists(ctx, coll_target):
return api.Error('invalid_foldername', 'The folder already exists. Please choose another name')
# All requirements OK
try:
collection.create(ctx, coll_target)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
def folder_copy_check(ctx: rule.Context, folder_path: str, new_folder_path: str, overwrite: bool, copy: bool = True) -> Tuple[bool, str]:
"""Check whether can copy (or move) folder to new folder location.
:param ctx: Combined type of a callback and rei struct
:param folder_path: Path to the folder to copy
:param new_folder_path: Path to the new copy of the folder
:param overwrite: Overwrite folder if it already exists
:param copy: Whether a copy operation (True) or move (False) (just for logging purposes)
:returns: 2-Tuple containing whether can copy/move, and the error if cannot
"""
# Whether copy or move
verb = 'copy' if copy else 'move'
verb_past = 'copied' if copy else 'moved'
if len(new_folder_path) == 0:
return False, api.Error('missing_folder_path', 'Missing folder path. Please add a folder path')
# TODO remove when upgrade to GenQuery 2
if '\'' in new_folder_path:
return False, api.Error('invalid_foldername', 'It is not allowed to use apostrophes in a folder name')
try:
validate_filepath(new_folder_path)
except ValidationError:
return False, api.Error('invalid_foldername', 'This is not a valid folder name. Please choose another name for your folder')
# Same folder path makes no sense.
if folder_path == new_folder_path:
return False, api.Error('invalid_folder_path', 'Origin and {} folder paths are equal. Please choose another destination'.format(verb))
# Inside the same path makes no sense.
if "{}/".format(folder_path) in new_folder_path:
return False, api.Error('invalid_folder_path', 'Cannot {} folder inside itself. Please choose another destination'.format(verb))
# not in home - a groupname must be present ie at least 2!?
if not len(new_folder_path.split('/')) > 2:
return False, api.Error('invalid_destination', 'It is not possible to {} folder at this location'.format(verb))
# in vault?
target_group_name = new_folder_path.split('/')[3]
if target_group_name.startswith('vault-'):
return False, api.Error('invalid_destination', 'It is not possible to {} folder to the vault'.format(verb))
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return False, api.Error('not_allowed', 'You do not have sufficient permissions to {} the selected folder'.format(verb))
# Folder not locked?
if folder.is_locked(ctx, new_folder_path):
return False, api.Error('not_allowed', 'The indicated folder is locked and therefore the folder can not be {}'.format(verb_past))
# Does original folder exist?
if not collection.exists(ctx, folder_path):
return False, api.Error('invalid_source', 'The original folder ' + folder_path + ' can not be found')
# Collection exists in destination?
if not overwrite and collection.exists(ctx, new_folder_path):
return False, api.Error('invalid_destination', 'Folder with this name already exists in destination')
return True, ""
@api.make()
def api_research_folder_copy(ctx: rule.Context, folder_path: str, new_folder_path: str, overwrite: bool = False) -> api.Result:
"""Copy a folder in a research folder.
:param ctx: Combined type of a callback and rei struct
:param folder_path: Path to the folder to copy
:param new_folder_path: Path to the new copy of the folder
:param overwrite: Overwrite folder if it already exists
:returns: API status result
"""
valid, errorResponse = folder_copy_check(ctx, folder_path, new_folder_path, overwrite, True)
if not valid:
return errorResponse
# All requirements OK
try:
collection.copy(ctx, folder_path, new_folder_path, force=overwrite)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_folder_move(ctx: rule.Context, folder_path: str, new_folder_path: str, overwrite: bool = False) -> api.Result:
"""Move a folder in a research folder.
:param ctx: Combined type of a callback and rei struct
:param folder_path: Path to the folder to move
:param new_folder_path: Path to the new folder
:param overwrite: Overwrite folder if it already exists
:returns: API status result
"""
valid, errorResponse = folder_copy_check(ctx, folder_path, new_folder_path, overwrite, False)
if not valid:
return errorResponse
# All requirements OK
try:
collection.move(ctx, folder_path, new_folder_path, force=overwrite)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_folder_rename(ctx: rule.Context, new_folder_name: str, coll: str, org_folder_name: str) -> api.Result:
"""Rename an existing research folder.
:param ctx: Combined type of a callback and rei struct
:param new_folder_name: New folder name
:param coll: Parent collection of folder
:param org_folder_name: Current name of the folder
:returns: API status result
"""
coll_target = coll + '/' + new_folder_name
valid_folder_name, error_response = folder_new_name_check(new_folder_name)
if not valid_folder_name:
return error_response
try:
validate_filepath(coll_target)
except ValidationError:
return api.Error('invalid_foldername', 'This is not a valid folder name. Please choose another name for your folder')
# Same name makes no sense
if new_folder_name == org_folder_name:
return api.Error('invalid_foldername', 'Origin and target folder names are equal. Please choose another name')
# not in home - a groupname must be present ie at least 2!?
if not len(coll.split('/')) > 2:
return api.Error('invalid_destination', 'It is not possible to add folder ' + org_folder_name + ' at this location')
# in vault?
target_group_name = coll_target.split('/')[3]
if target_group_name.startswith('vault-'):
return api.Error('not_allowed', 'It is not possible to rename folders in the vault')
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return api.Error('not_allowed', 'You do not have sufficient permissions to rename the selected folder')
# Collection exists?
if not collection.exists(ctx, coll):
return api.Error('invalid_foldername', 'The selected folder does not exist')
# Folder not locked?
if folder.is_locked(ctx, coll):
return api.Error('not_allowed', 'The indicated folder is locked and therefore can not be renamed')
# new collection exists?
if collection.exists(ctx, coll_target):
return api.Error('invalid_foldername', 'The folder already exists. Please choose another name')
# All requirements OK
try:
collection.rename(ctx, coll + '/' + org_folder_name, coll_target)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_folder_delete(ctx: rule.Context, coll: str, folder_name: str) -> api.Result:
"""Delete a research folder.
:param ctx: Combined type of a callback and rei struct
:param coll: Parent collection of folder to delete
:param folder_name: Name of folder to delete
:returns: API status result
"""
coll_target = coll + '/' + folder_name
# Not in home - a groupname must be present ie at least 2!?
if not len(coll.split('/')) > 2:
return api.Error('invalid_target', 'It is not possible to delete folder ' + folder_name + ' at this location')
# Name should not contain '\\' or '/'.
if '/' in folder_name or '\\' in folder_name:
return api.Error('invalid_foldername', 'It is not allowed to use slashes in folder name that will be deleted')
# in vault?
target_group_name = coll_target.split('/')[3]
if target_group_name.startswith('vault-'):
return api.Error('not_allowed', 'It is not possible to delete folders from the vault')
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return api.Error('not_allowed', 'You do not have sufficient permissions to delete the selected folder')
# Folder not locked?
if folder.is_locked(ctx, coll):
return api.Error('not_allowed', 'The indicated folder is locked and therefore can not be deleted')
# Collection exists?
if not collection.exists(ctx, coll_target):
return api.Error('invalid_target', 'The folder to delete does not exist')
# All requirements OK
try:
collection.remove(ctx, coll_target)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_list_temporary_files(ctx: rule.Context, coll: str) -> api.Result:
"""Get list of temporary files to be cleaned up.
:param ctx: Combined type of a callback and rei struct
:param coll: Parent collection of folder to delete
:returns: List of files that possibly require cleaning up
"""
list_cleanup_files = []
for uw_file in config.temporary_files:
if "?" in uw_file or "*" in uw_file:
wildcard_file = uw_file.replace('%', '\\%').replace('_', '\\_').replace('?', '_').replace('*', '%')
iter = genquery.row_iterator(
"DATA_NAME, COLL_NAME",
"COLL_NAME like '" + coll + "%' AND DATA_NAME LIKE '" + wildcard_file + "'",
genquery.AS_LIST, ctx
)
else:
iter = genquery.row_iterator(
"DATA_NAME, COLL_NAME",
"COLL_NAME like '" + coll + "%' AND DATA_NAME = '" + uw_file + "'",
genquery.AS_LIST, ctx
)
for row in iter:
list_cleanup_files.append([row[1], row[0]])
return list_cleanup_files
@api.make()
def api_research_file_copy(ctx: rule.Context, filepath: str, new_filepath: str, overwrite: bool = False) -> api.Result:
"""Copy a file in a research folder.
:param ctx: Combined type of a callback and rei struct
:param filepath: Path to the file to copy
:param new_filepath: Path to the new copy of the file
:param overwrite: Overwrite file if it already exists
:returns: API status result
"""
if len(new_filepath) == 0:
return api.Error('missing_filepath', 'Missing file path. Please add a file path')
# Same filepath makes no sense.
if filepath == new_filepath:
return api.Error('invalid_filepath', 'Origin and copy file paths are equal. Please choose another destination')
_, org_data_name = pathutil.chop(filepath)
# These are of the NEW filepath
coll, data_name = pathutil.chop(new_filepath)
try:
validate_filename(data_name)
except Exception:
return api.Error('invalid_filename', 'This is not a valid file name. Please choose another name')
# TODO remove when upgrade to GenQuery 2
# This check should only be done on new folders, since may have old folders with apostrophes
if '\'' in coll:
return api.Error('invalid_filepath', 'It is not allowed to copy a file to a folder with an apostrophe in the name')
# not in home - a groupname must be present ie at least 2!?
if not len(coll.split('/')) > 2:
return api.Error('invalid_destination', 'It is not possible to copy files at this location')
# Name should not contain '\\' or '/'
if '/' in data_name or '\\' in data_name:
return api.Error('invalid_filename', 'It is not allowed to use slashes in the new name of a file')
# in vault?
target_group_name = new_filepath.split('/')[3]
if target_group_name.startswith('vault-'):
return api.Error('invalid_destination', 'It is not possible to copy files in the vault')
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return api.Error('not_allowed', 'You do not have sufficient permissions to copy the selected file')
# Folder not locked?
if folder.is_locked(ctx, coll):
return api.Error('not_allowed', 'The indicated folder is locked and therefore the indicated file can not be copied')
# Does org file exist?
if not data_object.exists(ctx, filepath):
return api.Error('invalid_source', 'The original file ' + org_data_name + ' can not be found')
# new filename already exists?
if not overwrite and data_object.exists(ctx, new_filepath):
return api.Error('invalid_destination', 'The file ' + data_name + ' already exists')
# All requirements OK
try:
data_object.copy(ctx, filepath, new_filepath, force=overwrite)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_file_rename(ctx: rule.Context, new_file_name: str, coll: str, org_file_name: str) -> api.Result:
"""Rename a file in a research folder.
:param ctx: Combined type of a callback and rei struct
:param new_file_name: New file name
:param coll: Parent collection of file
:param org_file_name: Current name of the file
:returns: API status result
"""
if len(new_file_name) == 0:
return api.Error('missing_filename', 'Missing filename. Please add a file name')
try:
validate_filename(new_file_name)
except Exception:
return api.Error('invalid_filename', 'This is not a valid file name. Please choose another name')
# Same name makes no sense
if new_file_name == org_file_name:
return api.Error('invalid_filename', 'Origin and target file names are equal. Please choose another name')
path_target = coll + '/' + new_file_name
# not in home - a groupname must be present ie at least 2!?
if not len(coll.split('/')) > 2:
return api.Error('invalid_destination', 'It is not possible to rename files at this location')
# Name should not contain '\\' or '/'
if '/' in new_file_name or '\\' in new_file_name:
return api.Error('invalid_filename', 'It is not allowed to use slashes in the new name of a file')
# in vault?
target_group_name = path_target.split('/')[3]
if target_group_name.startswith('vault-'):
return api.Error('invalid_destination', 'It is not possible to rename files in the vault')
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return api.Error('not_allowed', 'You do not have sufficient permissions to rename the selected file')
# Folder not locked?
if folder.is_locked(ctx, coll):
return api.Error('not_allowed', 'The indicated folder is locked and therefore the indicated file can not be renamed')
# Does org file exist?
if not data_object.exists(ctx, coll + '/' + org_file_name):
return api.Error('invalid_source', 'The original file ' + org_file_name + ' can not be found')
# new filename already exists?
if data_object.exists(ctx, path_target):
return api.Error('invalid_destination', 'The selected filename ' + new_file_name + ' already exists')
# Does data object have replica in intermediate or locked state? (replica is not actively being written to or opened for read)
if data_object.has_replica_with_status(ctx, coll + '/' + org_file_name, [constants.replica_status.INTERMEDIATE_REPLICA,
constants.replica_status.READ_LOCKED,
constants.replica_status.WRITE_LOCKED]):
return api.Error('not_allowed', 'Not allowed, this file is actively being written to')
# All requirements OK
try:
data_object.rename(ctx, coll + '/' + org_file_name, path_target)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_file_move(ctx: rule.Context, filepath: str, new_filepath: str, overwrite: bool = False) -> api.Result:
"""Move a file in a research folder.
:param ctx: Combined type of a callback and rei struct
:param filepath: Path to the file to move
:param new_filepath: Path to the new location of the file
:param overwrite: Overwrite file if it already exists
:returns: API status result
"""
if len(new_filepath) == 0:
return api.Error('missing_filepath', 'Missing file path. Please add a file path')
# Same filepath makes no sense.
if filepath == new_filepath:
return api.Error('invalid_filepath', 'Origin and move file paths are equal. Please choose another destination')
# These are of the NEW filepath
coll, data_name = pathutil.chop(new_filepath)
try:
validate_filename(data_name)
except Exception:
return api.Error('invalid_filename', 'This is not a valid file name. Please choose another name')
# TODO remove when upgrade to GenQuery 2
# This check should only be done on new folders, since may have old folders with apostrophes
if '\'' in coll:
return api.Error('invalid_filepath', 'It is not allowed to move a file to a folder with an apostrophe in the name')
# not in home - a groupname must be present ie at least 2!?
if not len(coll.split('/')) > 2:
return api.Error('invalid_destination', 'It is not possible to move files to this location')
# Name should not contain '\\' or '/'
if '/' in data_name or '\\' in data_name:
return api.Error('invalid_filename', 'It is not allowed to use slashes in the new name of a file')
# in vault?
target_group_name = new_filepath.split('/')[3]
if target_group_name.startswith('vault-'):
return api.Error('invalid_destination', 'It is not possible to move files in the vault')
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return api.Error('not_allowed', 'You do not have sufficient permissions to move the selected file')
# Folder not locked?
if folder.is_locked(ctx, coll):
return api.Error('not_allowed', 'The indicated folder is locked and therefore the indicated file can not be moved')
# Does org file exist?
if not data_object.exists(ctx, filepath):
return api.Error('invalid_source', 'The original file ' + data_name + ' can not be found')
# new filename already exists?
if not overwrite and data_object.exists(ctx, new_filepath):
return api.Error('invalid_destination', 'The file ' + data_name + ' already exists')
# All requirements OK
try:
if overwrite:
data_object.copy(ctx, filepath, new_filepath, force=overwrite)
data_object.remove(ctx, filepath)
else:
data_object.rename(ctx, filepath, new_filepath)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_file_delete(ctx: rule.Context, coll: str, file_name: str) -> api.Result:
"""Delete a file in a research folder.
:param ctx: Combined type of a callback and rei struct
:param coll: Parent collection of file to delete
:param file_name: Name of file to delete
:returns: API status result
"""
path_target = coll + '/' + file_name
# not in home - a groupname must be present ie at least 2!?
if not len(coll.split('/')) > 2:
return api.Error('invalid_target', 'It is not possible to delete files at this location')
# in vault?
target_group_name = path_target.split('/')[3]
if target_group_name.startswith('vault-'):
return api.Error('not_allowed', 'It is not possible to delete files from the vault')
# permissions ok for group?
user_full_name = user.full_name(ctx)
if groups.user_role(ctx, user_full_name, target_group_name) in ['none', 'reader']:
return api.Error('not_allowed', 'You do not have sufficient permissions to delete the selected file')
# Folder not locked?
if folder.is_locked(ctx, coll):
return api.Error('not_allowed', 'The indicated folder is locked and therefore the indicated file can not be deleted')
# Data object exists?
if not data_object.exists(ctx, path_target):
return api.Error('invalid_target', 'The data object to delete does not exist')
# All requirements OK
try:
data_object.remove(ctx, path_target)
except msi.Error:
return api.Error('internal', 'Something went wrong. Please try again')
return api.Result.ok()
@api.make()
def api_research_system_metadata(ctx: rule.Context, coll: str) -> api.Result:
"""Return collection statistics as JSON.
:param ctx: Combined type of a callback and rei struct
:param coll: Research collection
:returns: API status result
"""
data_count = collection.data_count(ctx, coll)
collection_count = collection.collection_count(ctx, coll)
size = collection.size(ctx, coll)
size_readable = misc.human_readable_size(size)
result = "{} files, {} folders, total of {}".format(data_count, collection_count, size_readable)
return {"Package size": result}
@api.make()
def api_research_collection_details(ctx: rule.Context, path: str) -> api.Result:
"""Return details of a research collection.
:param ctx: Combined type of a callback and rei struct
:param path: Path to research collection
:returns: API status result
"""
if not collection.exists(ctx, path):
return api.Error('nonexistent', 'The given path does not exist')
# Check if collection is a research group.
space, _, group, _ = pathutil.info(path)
if space != pathutil.Space.RESEARCH:
return {}
basename = pathutil.chop(path)[1]
# Retrieve user type.
member_type = groups.user_role(ctx, user.full_name(ctx), group)
# Retrieve research folder status.
status = folder.get_status(ctx, path)
# Check if user is datamanager.
category = groups.group_category(ctx, group)
is_datamanager = groups.user_is_datamanager(ctx, category, user.full_name(ctx))
# Retrieve lock count.
lock_count = meta_form.get_coll_lock_count(ctx, path)
is_locked = folder.is_locked(ctx, path)
return {"basename": basename,
"status": status.value,
"member_type": member_type,
"is_datamanager": is_datamanager,
"lock_count": lock_count,
"is_locked": is_locked}
@api.make()
def api_research_manifest(ctx: rule.Context, coll: str) -> api.Result:
"""Produce a manifest of data objects in a collection
:param ctx: Combined type of a callback and rei struct
:param coll: Parent collection of data objects to include
:returns: List of json objects with name and checksum
"""
iter = genquery.row_iterator(
"ORDER(DATA_NAME), DATA_SIZE, DATA_CHECKSUM",
"COLL_NAME = '{}'".format(coll),
genquery.AS_LIST, ctx
)
checksums = [{"name": row[0], "size": misc.human_readable_size(int(row[1])), "checksum": data_object.decode_checksum(row[2])} for row in iter]
iter_sub = genquery.row_iterator(
"ORDER(COLL_NAME), ORDER(DATA_NAME), DATA_SIZE, DATA_CHECKSUM",
"COLL_PARENT_NAME like '{}%'".format(coll),
genquery.AS_LIST, ctx
)
length = len(coll) + 1
checksums_sub = [{"name": (row[0] + "/")[length:] + row[1], "size": misc.human_readable_size(int(row[2])), "checksum": data_object.decode_checksum(row[3])} for row in iter_sub]
return checksums + checksums_sub