CLI: add option --clean-workdir to verdi node delete

src/aiida/cmdline/commands/cmd_node.py

@@ -328,9 +328,14 @@ def extras(nodes, keys, fmt, identifier, raw):
 @click.argument('identifier', nargs=-1, metavar='NODES')
 @options.DRY_RUN()
 @options.FORCE()
+@click.option(
+    '--clean-workdir',
+    is_flag=True,
+    help='Also clean the remote work directory. (only applies to workflows and CalcJobNodes)',
+)
 @options.graph_traversal_rules(GraphTraversalRules.DELETE.value)
 @with_dbenv()
-def node_delete(identifier, dry_run, force, **traversal_rules):
+def node_delete(identifier, dry_run, force, clean_workdir, **traversal_rules):
     """Delete nodes from the provenance graph.
 
     This will not only delete the nodes explicitly provided via the command line, but will also include
@@ -349,6 +354,65 @@ def node_delete(identifier, dry_run, force, **traversal_rules):
         except ValueError:
             pks.append(NodeEntityLoader.load_entity(obj).pk)
 
+    if clean_workdir:
+        from aiida.manage import get_manager
+        from aiida.orm import AuthInfo, CalcJobNode, QueryBuilder, User, load_computer
+        from aiida.orm.utils.remote import get_calcjob_remote_paths
+        from aiida.tools.graph.graph_traversers import get_nodes_delete
+
+        backend = get_manager().get_backend()
+        # For here we ignore missing nodes will be raised via func:``delete_nodes`` in the next block
+        pks_set_to_delete = get_nodes_delete(
+            pks, get_links=False, missing_callback=lambda missing_pks: None, backend=backend, **traversal_rules
+        )['nodes']
+
+        qb = QueryBuilder()
+        qb.append(CalcJobNode, filters={'id': {'in': pks_set_to_delete}}, project='id')
+        calcjobs_pks = [result[0] for result in qb.all()]
+
+        if not calcjobs_pks:
+            echo.echo_report('--clean-workdir ignored. No CalcJobNode associated with the given node, found.')
+        else:
+            path_mapping = get_calcjob_remote_paths(
+                calcjobs_pks,
+                only_not_cleaned=True,
+            )
+
+            if not path_mapping:
+                echo.echo_report('--clean-workdir ignored. CalcJobNode work directories are already cleaned.')
+            else:
+                descendant_pks = [remote_folder.pk for paths in path_mapping.values() for remote_folder in paths]
+
+                if not force and not dry_run:
+                    echo.echo_warning(
+                        f'YOU ARE ABOUT TO CLEAN {len(descendant_pks)} REMOTE DIRECTORIES! ' 'THIS CANNOT BE UNDONE!'
+                    )
+                    echo.echo_info(
+                        'Remote directories of nodes with the following pks would be cleaned: '
+                        + ' '.join(map(str, descendant_pks))
+                    )
+                    click.confirm('Shall I continue?', abort=True)
+
+                if dry_run:
+                    echo.echo_report(
+                        'Remote folders of these node are marked for deletion: ' + ' '.join(map(str, descendant_pks))
+                    )
+                else:
+                    user = User.collection.get_default()
+                    counter = 0
+                    for computer_uuid, paths in path_mapping.items():
+                        computer = load_computer(uuid=computer_uuid)
+                        transport = AuthInfo.collection.get(
+                            dbcomputer_id=computer.pk, aiidauser_id=user.pk
+                        ).get_transport()
+
+                        with transport:
+                            for remote_folder in paths:
+                                remote_folder._clean(transport=transport)
+                                counter += 1
+
+                    echo.echo_report(f'{counter} remote folders cleaned on {computer.label}')
+
     def _dry_run_callback(pks):
         if not pks or force:
             return False