zenodoapi.py 11.9 KB
Newer Older
1 2
#!/usr/bin/env python

3
# E. Garcia 2020
4
# email: garcia 'at' lapp.in2p3.fr
5 6 7

import json
import requests
8 9 10
from utils_zenodoci import (find_root_directory,
                            parse_codemeta_and_write_zenodo_metadata_file
                            )
11 12 13 14 15


class ZenodoAPI:
    def __init__(self, access_token, sandbox=True):
        """
16
        Manages the communication with the (sandbox.)zenodo REST API through the Python request library.
17
        This class is **EXCLUSIVELY** developed to be used within a CI/CD pipeline and to **EXCLUSIVELY PERFORM**
18
         the following tasks within the (sandbox.)zenodo api environment:
19

20 21 22 23 24 25 26 27 28
          - Fetches a user's published entries,
          - Creates a new deposit,
          - Creates a new version of an existing deposit,
          - Uploads files to a specific Zenodo entry,
          - Erases a non-published entry / new version draft,
          - Erases (old version) files from an entry (when creating a new_version entry and uploading
            new_version files),
          - Uploads information to the entry (Zenodo compulsory deposit information),
          - Publishes an entry
29 30 31 32 33 34 35 36 37


        :param access_token: str
            Personal access token to (sandbox.)zenodo.org/api
        :param sandbox: bool
            Communicates with either zenodo or sandbox.zenodo api
        """

        if sandbox:
38
            zenodo_api_url = "https://sandbox.zenodo.org/api"
39
        else:
40
            zenodo_api_url = "https://zenodo.org/api"
41 42 43

        self.zenodo_api_url = zenodo_api_url
        self.access_token = access_token
44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61
        self.exist_codemeta_file = False
        self.path_codemeta_file = ''
        self.exist_zenodo_metadata_file = False
        self.path_zenodo_metadata_file = ''

    def fetch_user_entries(self):
        """
        Fetch the published entries of an user. Works to tests connection to Zenodo too.

        GET method to {zenodo_api_url}/deposit/depositions

        :return: obj
            request.get answer
        """
        url = f"{self.zenodo_api_url}/deposit/depositions"
        parameters = {'access_token': self.access_token}

        return requests.get(url, params=parameters)
62 63 64 65

    def create_new_entry(self):
        """
        Create a new entry / deposition in (sandbox.)zenodo
66

67 68
        POST method to {zenodo_api_url}/deposit/depositions

69 70
        :return: obj
            requests.put answer
71 72 73 74 75
        """
        url = f"{self.zenodo_api_url}/deposit/depositions"
        headers = {"Content-Type": "application/json"}
        parameters = {'access_token': self.access_token}

76
        return requests.post(url, json={}, headers=headers, params=parameters)
77

78
    def upload_file_entry(self, entry_id, name_file, path_file):
79 80 81
        """
        Upload a file to a Zenodo entry. If first retrieve the entry by a GET method to the
            {zenodo_api_url}/deposit/depositions/{entry_id}.
82

83 84 85 86 87 88 89 90 91
        PUT method to {bucket_url}/{filename}. The full api url is recovered when the entry is firstly retrieved.

        :param entry_id: str
            deposition_id of the Zenodo entry
        :param name_file: str
            File name of the file when uploaded
        :param path_file: str
            Path to the file to be uploaded

92 93
        :return: obj
            json requests.put object
94 95 96 97 98 99 100 101 102 103 104
        """
        # 1 - Retrieve and recover information of an existing deposit
        parameters = {'access_token': self.access_token}
        fetch = requests.get(f"{self.zenodo_api_url}/deposit/depositions/{entry_id}",
                             params=parameters)

        # 2 - Upload the files
        bucket_url = fetch.json()['links']['bucket']  # full url is recovered from fetch (GET) method
        url = f"{bucket_url}/{name_file}"

        with open(path_file, 'rb') as upload_file:
105
            upload = requests.put(url, data=upload_file, params=parameters)
106 107 108

        return upload.json()

109
    def update_metadata_entry(self, entry_id, data):
110 111 112
        """
        Update an entry resource. Data should be the entry information that will be shown when a deposition is visited
        at the Zenodo site.
113

114 115 116 117 118 119 120
        PUT method to {zenodo_api_url}/deposit/depositions/{entry_id}. `data` MUST be included as json.dump(data)

        :param entry_id: str
            deposition_id of the Zenodo entry
        :param data: object
            json object containing the metadata (compulsory fields) that are enclosed when a new entry is created.

121 122
        :return: obj
            requests.put answer
123 124 125 126 127
        """
        url = f"{self.zenodo_api_url}/deposit/depositions/{entry_id}"
        headers = {"Content-Type": "application/json"}
        parameters = {'access_token': self.access_token}

128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
        return requests.put(url, data=json.dumps(data),
                            headers=headers, params=parameters)

    def erase_entry(self, entry_id):
        """
        Erase an entry/new version of an entry that HAS NOT BEEN published yet.
        Any new upload/version will be first saved as 'draft' and not published until confirmation (i.e, requests.post)

        DELETE method to {zenodo_api_url}/deposit/depositions/{entry_id}.

        :param entry_id: str
            deposition_id of the Zenodo entry to be erased

        :return: obj
            requests.delete json answer
        """
        url = f"{self.zenodo_api_url}/deposit/depositions/{entry_id}"
        parameters = {'access_token': self.access_token}

        return requests.delete(url, params=parameters)
148

149 150 151 152 153 154 155 156 157 158 159 160
    def erase_file_entry(self, entry_id, file_id):
        """
        Erase a file from an entry resource.
        This method is intended to be used for substitution of files (deletion) within an entry by their correspondent
         new versions.
        DELETE method to {zenodo_api_url}/deposit/depositions/{entry_id}/files/{file_id}

        :param entry_id: str
            deposition_id of the Zenodo entry
        :param file_id: str
            Id of the files stored in Zenodo

161 162
        :return: obj
            requests.delete answer
163 164 165 166 167 168
        """
        url = f"{self.zenodo_api_url}/deposit/depositions/{entry_id}/files/{file_id}"
        parameters = {'access_token': self.access_token}

        return requests.delete(url, params=parameters)

169 170 171 172 173 174 175 176
    def publish_entry(self, entry_id):
        """
        Publishes an entry in (sandbox.)zenodo
        POST method to {zenodo_api_url}/deposit/depositions/{entry_id}/actions/publish

        :param entry_id: str
            deposition_id of the Zenodo entry

177 178
        :return: obj
            requests.put answer
179 180 181 182 183 184 185 186 187 188 189 190 191 192
        """
        url = f"{self.zenodo_api_url}/deposit/depositions/{entry_id}/actions/publish"
        parameters = {'access_token': self.access_token}

        return requests.post(url, params=parameters)

    def new_version_entry(self, entry_id):
        """
        Creates a new version of AN EXISTING entry resource.
        POST method to {zenodo_api_url}/deposit/depositions/{entry_id}/actions/newversion

        :param entry_id: str
            deposition_id of the Zenodo entry

193 194
        :return: obj
            requests.post answer
195
        """
196
        url = f"{self.zenodo_api_url}/deposit/depositions/{entry_id}/actions/newversion"
197 198 199
        parameters = {'access_token': self.access_token}

        return requests.post(url, params=parameters)
200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222

    def search_codemeta_file(self):
        """Check if a `codemeta.json` files exists in the ROOT directory of the project"""

        root_dir = find_root_directory()
        codemeta_file = root_dir / 'codemeta.json'

        if codemeta_file.exists():
            print("\n * Found codemeta.json file within the project !")
            self.exist_codemeta_file = True
            self.path_codemeta_file = codemeta_file
        else:
            print("\n ! codemeta.json file NOT found in the root directory of the  project !")

    def search_zenodo_json_file(self):
        """Check if a `.zenodo.json` files exists in the ROOT directory of the project"""

        root_dir = find_root_directory()
        zenodo_metadata_file = root_dir / '.zenodo.json'

        if zenodo_metadata_file.exists():
            print("\n * Found .zenodo.json file within the project !")
            self.exist_zenodo_metadata_file = True
Vuillaume's avatar
Vuillaume committed
223
            self.path_zenodo_metadata_file = zenodo_metadata_file
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
        else:
            print("\n ! .zenodo.json file NOT found in the root directory of the  project !")

    def test_upload_to_zenodo(self):
        """
        `Tests` the different stages of the GitLab-Zenodo connection and that the status_code returned by every
        stage is the correct one.

        Tests:
         - If it exists a `codemeta.json` file
            - If it exists a `.zenodo.json` file
               - If not, it creates one, based on the `codemeta.json` file
         - The communication with Zenodo through its API to verify that:
            - You can fetch an user entries
            - You can create a new entry
            - The provided zenodo metadata can be digested, and not errors appear
            - Finally erases the test entry - because IT HAS NOT BEEN PUBLISHED !
        """
        # Search for codemeta.json file within the project
        self.search_codemeta_file()

        if self.exist_codemeta_file:

            # Search for zenodo.json metadata file within the project
            self.search_zenodo_json_file()
            if self.exist_zenodo_metadata_file:
                print("\n * Using the .zenodo.json file to simulate a new upload to Zenodo... \n")
            else:
Vuillaume's avatar
Vuillaume committed
252
                print("\n ! Creating a .zenodo.json file from your codemeta.json file... \n")
253 254 255 256 257 258 259 260
                self.path_zenodo_metadata_file = self.path_codemeta_file.parent / '.zenodo.json'

                parse_codemeta_and_write_zenodo_metadata_file(self.path_codemeta_file,
                                                              self.path_zenodo_metadata_file)

            # 1 - Test connection
            print("Testing communication with Zenodo...")
            test_connection = self.fetch_user_entries()
Vuillaume's avatar
Vuillaume committed
261
            if test_connection.status_code == 200:
Vuillaume's avatar
Vuillaume committed
262
                print("Test connection status OK !")
Vuillaume's avatar
Vuillaume committed
263
            else:
Vuillaume's avatar
Vuillaume committed
264
                print("ERROR while testing connection status\n", test_connection.json())
265 266 267 268

            # 2 - Test new entry
            print("Testing the creation of a dummy entry to (sandbox)Zenodo...")
            new_entry = self.create_new_entry()
Vuillaume's avatar
Vuillaume committed
269
            if new_entry.status_code == 201:
Vuillaume's avatar
Vuillaume committed
270
                print("Test new entry status OK !")
Vuillaume's avatar
Vuillaume committed
271
            else:
Vuillaume's avatar
Vuillaume committed
272
                print("ERROR while testing the creation of new entry\n", new_entry.json())
273 274 275 276 277 278 279 280 281 282 283

            # 3 - Test upload metadata
            print("Testing the ingestion of the Zenodo metadata...")
            test_entry_id = new_entry.json()['id']

            with open(self.path_zenodo_metadata_file) as file:
                metadata_entry = json.load(file)

            update_metadata = self.update_metadata_entry(test_entry_id,
                                                         data=metadata_entry
                                                         )
Vuillaume's avatar
Vuillaume committed
284
            if update_metadata.status_code == 200:
Vuillaume's avatar
Vuillaume committed
285
                print("Update metadata status OK !")
Vuillaume's avatar
Vuillaume committed
286
            else:
Vuillaume's avatar
Vuillaume committed
287
                print("ERROR while testing update of metadata\n", update_metadata.json())
288 289 290 291

            # 4 - Test delete entry
            print("Testing the deletion of the dummy entry...")
            delete_test_entry = self.erase_entry(test_entry_id)
Vuillaume's avatar
Vuillaume committed
292
            if delete_test_entry.status_code == 204:
Vuillaume's avatar
Vuillaume committed
293
                print("Delete test entry status OK !")
Vuillaume's avatar
Vuillaume committed
294
            else:
Vuillaume's avatar
Vuillaume committed
295
                print("ERROR while deleting test entry\n", delete_test_entry.json())
296 297 298 299 300 301 302 303 304

            print("\n\tYAY ! Successful testing of the connection to Zenodo ! \n\n"
                  "You should not face any trouble when uploading a project to Zenodo through the "
                  "ESCAPE-GitLabCI pipeline.\n"
                  "In case you do, please contact us !\n")

        else:

            print("\n ! Please add a codemeta.json file to the ROOT directory of your project.\n")