DD-1493: Add command for dataverse role assignment

pyproject.toml

@@ -52,3 +52,4 @@ ingest-flow = "datastation.ingest_flow:main"
 dv-dataverse-root-collect-storage-usage = "datastation.dv_dataverse_root_collect_storage_usage:main"
 dv-dataverse-root-collect-permission-overview = "datastation.dv_dataverse_root_collect_permission_overview:main"
 datastation-get-component-versions = "datastation.datastation_get_component_versions:main"
+dv-dataverse-role-assignment = "datastation.dv_dataverse_role_assignment:main"

src/datastation/dataverse/dataverse_api.py

@@ -1,13 +1,18 @@
 import requests
+import json
 
 
 from datastation.common.utils import print_dry_run_message
 
 
 class DataverseApi:
-    def __init__(self, server_url, api_token):
+    def __init__(self, server_url, api_token, alias=None):
         self.server_url = server_url
         self.api_token = api_token
+        self.alias = alias  # Methods should use this one if specified
+
+    def get_alias(self):
+        return self.alias
 
     # get json data for a specific dataverses API endpoint using an API token
     def get_resource_data(self, resource, alias="root", dry_run=False):
@@ -30,7 +35,9 @@ def get_contents(self, alias="root", dry_run=False):
     def get_roles(self, alias="root", dry_run=False):
         return self.get_resource_data("roles", alias, dry_run)
 
-    def get_assignments(self, alias="root", dry_run=False):
+    def get_role_assignments(self, alias="root", dry_run=False):
+        if self.alias is not None:
+            alias = self.alias
         return self.get_resource_data("assignments", alias, dry_run)
 
     def get_groups(self, alias="root", dry_run=False):
@@ -47,3 +54,31 @@ def get_storage_size(self, alias="root", dry_run=False):
             r = requests.get(url, headers=headers)
         r.raise_for_status()
         return r.json()['data']['message']
+
+    def add_role_assignment(self, assignee, role, alias=None, dry_run=False):
+        if self.alias is not None:
+            alias = self.alias
+        url = f'{self.server_url}/api/dataverses/{alias}/assignments'
+        headers = {'X-Dataverse-key': self.api_token, 'Content-type': 'application/json'}
+        role_assignment = {"assignee": assignee, "role": role}
+        if dry_run:
+            print_dry_run_message(method='POST', url=url, headers=headers,
+                                  data=json.dumps(role_assignment))
+            return None
+        else:
+            r = requests.post(url, headers=headers, json=role_assignment)
+            r.raise_for_status()
+            return r
+
+    def remove_role_assignment(self, assignment_id, alias=None, dry_run=False):
+        if self.alias is not None:
+            alias = self.alias
+        url = f'{self.server_url}/api/dataverses/{alias}/assignments/{assignment_id}'
+        headers = {'X-Dataverse-key': self.api_token, 'Content-type': 'application/json'}
+        if dry_run:
+            print_dry_run_message(method='DELETE', url=url, headers=headers)
+            return None
+        else:
+            r = requests.delete(url, headers=headers)
+        r.raise_for_status()
+        return r

src/datastation/dataverse/dataverse_client.py

@@ -27,8 +27,8 @@ def search_api(self):
     def dataset(self, pid):
         return DatasetApi(pid, self.server_url, self.api_token, self.unblock_key, self.safety_latch)
 
-    def dataverse(self):
-        return DataverseApi(self.server_url, self.api_token)
+    def dataverse(self, alias=None):
+        return DataverseApi(self.server_url, self.api_token, alias)
 
     def file(self, file_id):
         return FileApi(file_id, self.server_url, self.api_token, self.unblock_key, self.safety_latch)

src/datastation/dataverse/permissions_collect.py

@@ -61,7 +61,7 @@ def get_role_info(self, alias):
         return ', '.join(result_list)
 
     def get_assignment_info(self, alias):
-        resp_data = self.dataverse_client.dataverse().get_assignments(alias)
+        resp_data = self.dataverse_client.dataverse().get_role_assignments(alias)
         # flatten and compact it... no list comprehension though
         result_list = []
         for assignment in resp_data:

src/datastation/dv_dataverse_role_assignment.py

@@ -0,0 +1,127 @@
+import argparse
+from datetime import datetime
+
+import rich
+
+from datastation.common.batch_processing import get_pids, BatchProcessorWithReport
+from datastation.common.config import init
+from datastation.common.utils import add_batch_processor_args, add_dry_run_arg
+from datastation.dataverse.dataset_api import DatasetApi
+from datastation.dataverse.dataverse_api import DataverseApi
+from datastation.dataverse.dataverse_client import DataverseClient
+
+
+def add_role_assignments(args, dataverse_client: DataverseClient):
+    pids = get_pids(args.pid_or_pid_file)
+    batch_processor = BatchProcessorWithReport(wait=args.wait, fail_on_first_error=args.fail_fast,
+                                               report_file=args.report_file,
+                                               headers=['DOI', 'Modified', 'Assignee', 'Role', 'Change'])
+    batch_processor.process_pids(pids,
+                                 lambda pid, csv_report: add_role_assignment(args.role_assignment,
+                                                                             dataverse_api=dataverse_client.dataverse(pid),
+                                                                             csv_report=csv_report))
+
+
+def add_role_assignment(role_assignment, dataverse_api: DataverseApi, csv_report, dry_run: bool = False):
+    assignee = role_assignment.split('=')[0]
+    role = role_assignment.split('=')[1]
+    action = "None"
+    if in_current_assignments(assignee, role, dataverse_api):
+        print("{} is already {} for dataset {}".format(assignee, role, dataverse_api.get_alias()))
+    else:
+        print(
+            "Adding {} as {} for dataset {}".format(assignee, role, dataverse_api.get_alias()))
+        dataverse_api.add_role_assignment(assignee, role, dry_run=dry_run)
+        action = "Added"
+    csv_report.write(
+        {'DOI': dataverse_api.get_alias(), 'Modified': datetime.now(), 'Assignee': assignee, 'Role': role,
+         'Change': action})
+
+
+def in_current_assignments(assignee, role, dataverse_api: DataverseApi):
+    current_assignments = dataverse_api.get_role_assignments()
+    found = False
+    for current_assignment in current_assignments:
+        if current_assignment.get('assignee') == assignee and current_assignment.get(
+                '_roleAlias') == role:
+            found = True
+            break
+    return found
+
+
+def list_role_assignments(args, dataverse_client):
+    r = dataverse_client.dataverse(args.pid).get_role_assignments()
+    if r is not None:
+        rich.print_json(data=r)
+
+
+def remove_role_assignments(args, dataverse_client: DataverseClient):
+    pids = get_pids(args.pid_or_pid_file)
+    batch_processor = BatchProcessorWithReport(wait=args.wait, report_file=args.report_file,
+                                               headers=['alias', 'Modified', 'Assignee', 'Role', 'Change'])
+    batch_processor.process_pids(pids,
+                                 lambda pid,
+                                 csv_report: remove_role_assignment(args.role_assignment,
+                                                                    dataverse_api=dataverse_client.dataverse(pid),
+                                                                    csv_report=csv_report))
+
+def remove_role_assignment(role_assignment, dataverse_api: DataverseApi, csv_report, dry_run: bool = False):
+    assignee = role_assignment.split('=')[0]
+    role = role_assignment.split('=')[1]
+    action = "None"
+    if in_current_assignments(assignee, role, dataverse_api):
+        print("Removing {} as {} for dataverse {}".format(assignee, role, dataverse_api.get_alias()))
+        all_assignments = dataverse_api.get_role_assignments()
+        for assignment in all_assignments:
+            if assignment.get('assignee') == assignee and assignment.get('_roleAlias') == role:
+                dataverse_api.remove_role_assignment(assignment.get('id'), dry_run=dry_run)
+                action = "Removed"
+                break
+    else:
+        print("{} is not {} for dataset {}".format(assignee, role, dataverse_api.get_alias()))
+    csv_report.write(
+        {'alias': dataverse_api.get_alias(), 'Modified': datetime.now(), 'Assignee': assignee, 'Role': role,
+         'Change': action})
+
+
+def main():
+    config = init()
+    dataverse_client = DataverseClient(config['dataverse'])
+    batch_processor = BatchProcessorWithReport(headers=['alias', 'Modified', 'Assignee', 'Role', 'Change'])
+
+    # Create main parser and subparsers
+    parser = argparse.ArgumentParser(description='Manage role assignments on one or more datasets.')
+    subparsers = parser.add_subparsers(help='subcommands', dest='subcommand')
+
+    # Add role assignment
+    parser_add = subparsers.add_parser('add', help='add role assignment to specified dataset(s)')
+    parser_add.add_argument('role_assignment',
+                            help='role assignee and alias (example: @dataverseAdmin=contributor) to add')
+    parser_add.add_argument('pid_or_pid_file', help='the dataset pid or the input file with the dataset pids')
+    add_batch_processor_args(parser_add)
+    add_dry_run_arg(parser_add)
+
+    parser_add.set_defaults(func=lambda _: add_role_assignments(_, dataverse_client))
+
+    # Remove role assignment
+    parser_remove = subparsers.add_parser('remove', help='remove role assignment from specified dataset(s)')
+    parser_remove.add_argument('role_assignment',
+                               help='role assignee and alias (example: @dataverseAdmin=contributor)')
+    parser_remove.add_argument('pid_or_pid_file', help='The dataset pid or the input file with the dataset pids')
+    add_batch_processor_args(parser_remove)
+    add_dry_run_arg(parser_remove)
+    parser_remove.set_defaults(func=lambda _: remove_role_assignments(_, dataverse_client))
+
+    # List role assignments
+    parser_list = subparsers.add_parser('list',
+                                        help='list role assignments for specified dataset (only one pid allowed)')
+    parser_list.add_argument('pid', help='the dataset pid')
+    add_dry_run_arg(parser_list)
+    parser_list.set_defaults(func=lambda _: list_role_assignments(_, dataverse_client))
+
+    args = parser.parse_args()
+    args.func(args)
+
+
+if __name__ == '__main__':
+    main()