dspace_api_tools/src/delete_via_api.py at main · ualbertalib/dspace_api_tools · GitHub

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
"""
##############################################################################################
# desc: delete_via_api.py:
#     Test https://pypi.org/project/dspace-rest-client
#     to delete communities, collections, items from a DSpace instance
#     and write the output to a file.
#     https://github.com/the-library-code/dspace-rest-python/blob/main/dspace_rest_client/client.py
# license: CC0 1.0 Universal (CC0 1.0) Public Domain Dedication
# date: March 24, 2025
##############################################################################################
"""

import argparse
import csv
import logging
import sys

from dspace_rest_client.client import DSpaceClient

from utils import utilities as utils


def parse_args():
    """
    Parse command line arguments
    """

    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--input",
        required=True,
        help="Location of a CSV with a column named 'id' with cells of the form 'uuid'.",
    )
    parser.add_argument(
        "--logging_level", required=False, help="Logging level.", default="INFO"
    )

    return parser.parse_args()


#
def process(dspace_client, csv_reader):
    """
    Process CSV rows and call DSpace delete
    """
    for row in csv_reader:
        print(row)
        collections = dspace_client.get_collections(uuid=row["id"])
        for collection in collections:
            logging.info("%s (%s)", collection.name, collection.uuid)
            dspace_client.delete_dso(collection)


#
def main():
    """
    Main entry point
    """

    args = parse_args()

    dspace_client = DSpaceClient(fake_user_agent=True)

    utils.configure_logging(args.logging_level)

    utils.check_required_env_vars()

    try:
        dspace_client.authenticate()
    except TypeError as e:
        logging.error("Check credentials and VPN (if applicable) [%s]", e)
        sys.exit(1)

    with open(args.input, "r", encoding="utf-8", newline="") as input_file:
        csv_reader = csv.DictReader(
            input_file,
        )
        process(dspace_client, csv_reader)


#
if __name__ == "__main__":
    main()