Skip to content

tpm_connector

TPMConnector

This class handles all communication with the Typed PID Maker (TPM) (see https://kit-data-manager.github.io/webpage/typed-pid-maker for more information). The Typed PID Maker is a service that allows the creation and management of FAIR-DOs (Findable, Accessible, Interoperable, Reusable Digital Objects). It communicates with the Handle.net service to create PIDs and stores PID records in a globally persistent manner.

Attributes:

Name Type Description
_tpm_url

str The URL of the Typed PID Maker instance
Source code in src/nmr_FAIR_DOs/connectors/tpm_connector.py 
 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
class TPMConnector:
    """
    This class handles all communication with the Typed PID Maker (TPM) (see https://kit-data-manager.github.io/webpage/typed-pid-maker for more information).
    The Typed PID Maker is a service that allows the creation and management of FAIR-DOs (Findable, Accessible, Interoperable, Reusable Digital Objects).
    It communicates with the Handle.net service to create PIDs and stores PID records in a globally persistent manner.

    Attributes:
        _tpm_url:str The URL of the Typed PID Maker instance
    """

    def __init__(self, tpm_url: str):
        """
        Creates a new TPMConnector

        Args:
            tpm_url:str The URL of the Typed PID Maker instance

        Raises:
            ValueError: If the TPM URL is None or empty
        """
        if tpm_url is None or len(tpm_url) == 0:
            raise ValueError("TPM URL must not be None or empty")
        self._tpm_url = tpm_url

    def createSingleFAIRDO(self, pidRecord: PIDRecord) -> PIDRecord:
        """
        Creates a single FAIR-DO in the TPM

        Args:
            pidRecord:PIDRecord The FAIR-DO to create

        Returns:
            PIDRecord The response from the TPM. This response contains the PID and the PID record.
        """
        logger.info(f"Creating FAIR-DO {pidRecord.getPID()}")

        if pidRecord is None or not isinstance(
            pidRecord, PIDRecord
        ):  # if the PID record is None or not an instance of PIDRecord, raise an error
            raise ValueError(
                "FAIR-DO must not be None and must be an instance of PIDRecord"
            )

        headers = {"Content-Type": "application/json"}

        # content = self._applyTypeAPIFixes(pidRecord.toJSON()) # Possible fix for Type API issues
        content = pidRecord.toJSON()  # get the JSON representation of the PID record

        endpoint = "/api/v1/pit/pid"

        if (
            content is None or len(content) == 0
        ):  # if the content is None or empty, raise an error
            raise ValueError("No content to create due to invalid input")

        resource_response = requests.post(
            self._tpm_url + endpoint, headers=headers, json=content
        )  # send a POST request to the TPM to create the PID record
        logger.debug(f"Response for URL {self._tpm_url + endpoint}", resource_response)

        if (
            resource_response.status_code != 201
        ):  # if the status code is not 201, raise an error
            raise Exception("Error creating PID record: ", resource_response)

        return PIDRecord.fromJSON(
            resource_response.json()
        )  # parse a PID record from the response JSON and return it

    def createMultipleFAIRDOs(self, pidRecord: list[PIDRecord]) -> list[PIDRecord]:
        """
        Creates multiple FAIR-DOs in the TPM.
        This function uses the bulk-create endpoint of the TPM to create multiple FAIR-DOs at once.
        One advantage of using this endpoint is that it can create multiple connected FAIR-DOs in one request and automatically replaces "placeholder"/"fantasy"/"preliminary" PIDs in the records with the real deal.

        Args:
            pidRecord:list[PIDRecord] The FAIR-DOs to create

        Returns:
            list[PIDRecord] The response from the TPM which is a list of all created FAIR-DOs
        """
        logger.info(f"Creating {len(pidRecord)} FAIR-DOs")

        headers = {"Content-Type": "application/json"}

        content = []

        for fairdo in pidRecord:  # iterate over all PID records
            if fairdo is None or not isinstance(
                fairdo, PIDRecord
            ):  # Check the validity or raise an exception
                raise ValueError(
                    "FAIR-DO must not be None and must be an instance of PIDRecord"
                )

            # content.append(self._applyTypeAPIFixes(fairdo.toJSON())) # Possible fix for Type API issues
            content.append(fairdo.toJSON())  # Mark this record ready for creation

        endpoint = "/api/v1/pit/pids"

        if (
            content is None or len(content) == 0
        ):  # if the content is None or empty, raise an error
            raise ValueError("No content to create due to invalid input")

        logger.debug(
            f"Creating FAIR-DOs at {self._tpm_url + endpoint} : {json.dumps(content)[:250]}"
        )
        resource_response = requests.post(
            self._tpm_url + endpoint, headers=headers, json=content, timeout=None
        )  # send a POST request to the TPM to create the PID records

        logger.debug(f"Response for URL {self._tpm_url + endpoint}", resource_response)

        if (
            resource_response.status_code != 201
        ):  # if the status code is not 201, raise an error
            raise Exception(
                "Error creating PID records. API response from TPM: ",
                repr(resource_response),
            )

        result = []
        for (
            i
        ) in resource_response.json():  # iterate over all PID records in the response
            result.append(
                PIDRecord.fromJSON(i)
            )  # parse a PID record from the response JSON and add it to the result list

        logger.info("Successfully created FAIR-DOs")
        return result  # return the list of all created PID records (with their actual PIDs)

    def getPIDRecord(self, pid: str) -> PIDRecord:
        """
        Retrieves a PID record from the TPM

        Args:
            pid (str): The PID to retrieve

        Returns:
            PIDRecord: The PID record retrieved from the TPM

        Raises:
            ValueError: If the PID is None or empty
            Exception: If the PID record cannot be retrieved
        """
        if pid is None or len(pid) == 0:  # if the PID is None or empty, raise an error
            raise ValueError("PID must not be None or empty")

        endpoint = "/api/v1/pit/pid/" + pid

        resource_response = requests.get(
            self._tpm_url + endpoint, headers={"Accept": "application/json"}
        )  # send a GET request to the TPM to retrieve the PID record

        if (
            resource_response.status_code != 200
        ):  # if the status code is not 200, raise an error
            raise Exception("Error retrieving PID record: ", resource_response)

        return PIDRecord.fromJSON(
            resource_response.json()
        )  # parse a PID record from the response JSON and return it

    def updatePIDRecord(self, pidRecord: PIDRecord) -> PIDRecord:
        """
        Updates a PID record in the TPM

        Args:
            pidRecord:PIDRecord The PID record to update

        Returns:
            PIDRecord The response from the TPM

        Raises:
            ValueError: If the PID record is None or not an instance of PIDRecord
            Exception: If the PID record cannot be updated
        """
        if pidRecord is None or not isinstance(
            pidRecord, PIDRecord
        ):  # if the PID record is None or not an instance of PIDRecord, raise an error
            raise ValueError(
                "PID record must not be None and must be an instance of PIDRecord"
            )

        headers = {"Content-Type": "application/json"}

        content = pidRecord.toJSON()  # get the JSON representation of the PID record

        endpoint = "/api/v1/pit/pid/" + pidRecord.getPID()  # create the endpoint URL

        if (
            content is None or len(content) == 0
        ):  # if the content is None or empty, raise an error
            raise ValueError("No content to update due to invalid input")

        resource_response = requests.put(
            self._tpm_url + endpoint, headers=headers, json=content
        )  # send a PUT request to the TPM to update the PID record

        if (
            resource_response.status_code != 200
        ):  # if the status code is not 200, raise an error
            raise Exception(
                "Error updating PID record: ",
                resource_response,
            )

        return PIDRecord.fromJSON(
            resource_response.json()
        )  # parse a PID record from the response JSON and return it

    async def getAllPIDRecords(self) -> list[PIDRecord]:
        """
        Retrieves all PID records from the TPM.
        This function uses the known-pid endpoint of the TPM to retrieve all PID records.
        Keep in mind that the TPM does not necessarily know all PIDs in the system or even in the prefix or same instance.

        Returns:
            list[PIDRecord] The list of all PID records
        """
        endpoint = "/api/v1/pit/known-pid"

        resource_response = requests.get(
            self._tpm_url + endpoint, headers={"Accept": "application/json"}
        )  # send a GET request to the TPM to retrieve all PID records

        if (
            resource_response.status_code != 200
        ):  # if the status code is not 200, raise an error
            raise Exception("Error retrieving PID records: ", resource_response)

        url_template = Template(
            "$tpmURL/api/v1/pit/pid/$pid"
        )  # create a template for the URL

        single_pidRecord_urls = []
        for (
            i
        ) in resource_response.json():  # iterate over all PID records in the response
            # Create the URL
            url = url_template.safe_substitute(
                tpmURL=self._tpm_url,
                pid=i["pid"],
            )  # create the URL for the PID record
            single_pidRecord_urls.append(url)  # add the URL to the list

        # Fetch all PID records in a parallelized manner. Do not use the cache.
        json_records = await fetch_multiple(single_pidRecord_urls, True)

        result = []
        for i in json_records:  # iterate over all PID records in the response
            result.append(
                PIDRecord.fromJSON(i)
            )  # parse a PID record from the response JSON and add it to the result list

        return result  # return the list of all fetched PID records

    @staticmethod
    def _applyTypeAPIFixes(content: dict) -> dict:
        """
        Applies fixes to the content to match the TPM API.
        This is due to an issue in the schema generation of the Type API.

        Args:
            content:dict The content to fix

        Returns:
            dict The fixed content

        Raises:
            ValueError: If the content is None or not a dict
        """
        # Define a set of types that need to be fixed and the new name of the internal key
        types_to_fix = {
            "21.T11969/8710d753ad10f371189b": "landingPageLocation",
            "21.T11148/f3f0cbaa39fa9966b279": "identifier",
            "21.T11969/7a19f6d5c8e63dd6bfcb": "NMR_Method",
            "21.T11148/7fdada5846281ef5d461": "locationPreview/Sample",
        }

        if content is None or not isinstance(
            content, dict
        ):  # if the content is None or not a dict, raise an error
            raise ValueError("Content must not be None and must be a dict")

        result = {"pid": content["pid"], "entries": {}}

        for key, value in content[
            "entries"
        ].items():  # iterate over all entries in the content
            fix_name = types_to_fix.get(key)  # get the fix name for the current key

            if fix_name is not None:  # if a fix name is available
                logger.debug(f"Fixing content for type {key} to {fix_name}")
                values = []
                for item in value:  # iterate over all values to this key
                    newEntry = {
                        "key": item["key"],
                        "value": '{"' + fix_name + '": "' + item["value"] + '"}',
                    }  # create a new entry with the fixed name and value in an internal JSON string
                    values.append(newEntry)  # add the new entry to the list
                result["entries"][key] = values  # define the new entries for the key
                logger.debug(f"Fixed content for type {key} to {values}")
            else:  # if no fix name is available
                result["entries"][key] = value  # use the original value
                logger.debug(f"No fix for type {key}")

        return result  # return the fixed content

__init__ 

__init__(tpm_url: str)

Creates a new TPMConnector

Parameters:

Name Type Description Default
tpm_url str

str The URL of the Typed PID Maker instance

 required

Raises:

Type Description
ValueError

If the TPM URL is None or empty
Source code in src/nmr_FAIR_DOs/connectors/tpm_connector.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(self, tpm_url: str):
    """
    Creates a new TPMConnector

    Args:
        tpm_url:str The URL of the Typed PID Maker instance

    Raises:
        ValueError: If the TPM URL is None or empty
    """
    if tpm_url is None or len(tpm_url) == 0:
        raise ValueError("TPM URL must not be None or empty")
    self._tpm_url = tpm_url

createSingleFAIRDO 

createSingleFAIRDO(pidRecord: PIDRecord) -> PIDRecord

Creates a single FAIR-DO in the TPM

Parameters:

Name Type Description Default
pidRecord PIDRecord

PIDRecord The FAIR-DO to create

 required

Returns:

Type Description
PIDRecord

PIDRecord The response from the TPM. This response contains the PID and the PID record.
Source code in src/nmr_FAIR_DOs/connectors/tpm_connector.py 
 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
def createSingleFAIRDO(self, pidRecord: PIDRecord) -> PIDRecord:
    """
    Creates a single FAIR-DO in the TPM

    Args:
        pidRecord:PIDRecord The FAIR-DO to create

    Returns:
        PIDRecord The response from the TPM. This response contains the PID and the PID record.
    """
    logger.info(f"Creating FAIR-DO {pidRecord.getPID()}")

    if pidRecord is None or not isinstance(
        pidRecord, PIDRecord
    ):  # if the PID record is None or not an instance of PIDRecord, raise an error
        raise ValueError(
            "FAIR-DO must not be None and must be an instance of PIDRecord"
        )

    headers = {"Content-Type": "application/json"}

    # content = self._applyTypeAPIFixes(pidRecord.toJSON()) # Possible fix for Type API issues
    content = pidRecord.toJSON()  # get the JSON representation of the PID record

    endpoint = "/api/v1/pit/pid"

    if (
        content is None or len(content) == 0
    ):  # if the content is None or empty, raise an error
        raise ValueError("No content to create due to invalid input")

    resource_response = requests.post(
        self._tpm_url + endpoint, headers=headers, json=content
    )  # send a POST request to the TPM to create the PID record
    logger.debug(f"Response for URL {self._tpm_url + endpoint}", resource_response)

    if (
        resource_response.status_code != 201
    ):  # if the status code is not 201, raise an error
        raise Exception("Error creating PID record: ", resource_response)

    return PIDRecord.fromJSON(
        resource_response.json()
    )  # parse a PID record from the response JSON and return it

createMultipleFAIRDOs 

createMultipleFAIRDOs(
    pidRecord: list[PIDRecord],
) -> list[PIDRecord]

Creates multiple FAIR-DOs in the TPM. This function uses the bulk-create endpoint of the TPM to create multiple FAIR-DOs at once. One advantage of using this endpoint is that it can create multiple connected FAIR-DOs in one request and automatically replaces "placeholder"/"fantasy"/"preliminary" PIDs in the records with the real deal.

Parameters:

Name Type Description Default
pidRecord list[PIDRecord]

list[PIDRecord] The FAIR-DOs to create

 required

Returns:

Type Description
list[PIDRecord]

list[PIDRecord] The response from the TPM which is a list of all created FAIR-DOs
Source code in src/nmr_FAIR_DOs/connectors/tpm_connector.py 
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
def createMultipleFAIRDOs(self, pidRecord: list[PIDRecord]) -> list[PIDRecord]:
    """
    Creates multiple FAIR-DOs in the TPM.
    This function uses the bulk-create endpoint of the TPM to create multiple FAIR-DOs at once.
    One advantage of using this endpoint is that it can create multiple connected FAIR-DOs in one request and automatically replaces "placeholder"/"fantasy"/"preliminary" PIDs in the records with the real deal.

    Args:
        pidRecord:list[PIDRecord] The FAIR-DOs to create

    Returns:
        list[PIDRecord] The response from the TPM which is a list of all created FAIR-DOs
    """
    logger.info(f"Creating {len(pidRecord)} FAIR-DOs")

    headers = {"Content-Type": "application/json"}

    content = []

    for fairdo in pidRecord:  # iterate over all PID records
        if fairdo is None or not isinstance(
            fairdo, PIDRecord
        ):  # Check the validity or raise an exception
            raise ValueError(
                "FAIR-DO must not be None and must be an instance of PIDRecord"
            )

        # content.append(self._applyTypeAPIFixes(fairdo.toJSON())) # Possible fix for Type API issues
        content.append(fairdo.toJSON())  # Mark this record ready for creation

    endpoint = "/api/v1/pit/pids"

    if (
        content is None or len(content) == 0
    ):  # if the content is None or empty, raise an error
        raise ValueError("No content to create due to invalid input")

    logger.debug(
        f"Creating FAIR-DOs at {self._tpm_url + endpoint} : {json.dumps(content)[:250]}"
    )
    resource_response = requests.post(
        self._tpm_url + endpoint, headers=headers, json=content, timeout=None
    )  # send a POST request to the TPM to create the PID records

    logger.debug(f"Response for URL {self._tpm_url + endpoint}", resource_response)

    if (
        resource_response.status_code != 201
    ):  # if the status code is not 201, raise an error
        raise Exception(
            "Error creating PID records. API response from TPM: ",
            repr(resource_response),
        )

    result = []
    for (
        i
    ) in resource_response.json():  # iterate over all PID records in the response
        result.append(
            PIDRecord.fromJSON(i)
        )  # parse a PID record from the response JSON and add it to the result list

    logger.info("Successfully created FAIR-DOs")
    return result  # return the list of all created PID records (with their actual PIDs)

getPIDRecord 

getPIDRecord(pid: str) -> PIDRecord

Retrieves a PID record from the TPM

Parameters:

Name Type Description Default
pid str

The PID to retrieve

 required

Returns:

Name Type Description
PIDRecord PIDRecord

The PID record retrieved from the TPM

Raises:

Type Description
ValueError

If the PID is None or empty
Exception

If the PID record cannot be retrieved
Source code in src/nmr_FAIR_DOs/connectors/tpm_connector.py 
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
def getPIDRecord(self, pid: str) -> PIDRecord:
    """
    Retrieves a PID record from the TPM

    Args:
        pid (str): The PID to retrieve

    Returns:
        PIDRecord: The PID record retrieved from the TPM

    Raises:
        ValueError: If the PID is None or empty
        Exception: If the PID record cannot be retrieved
    """
    if pid is None or len(pid) == 0:  # if the PID is None or empty, raise an error
        raise ValueError("PID must not be None or empty")

    endpoint = "/api/v1/pit/pid/" + pid

    resource_response = requests.get(
        self._tpm_url + endpoint, headers={"Accept": "application/json"}
    )  # send a GET request to the TPM to retrieve the PID record

    if (
        resource_response.status_code != 200
    ):  # if the status code is not 200, raise an error
        raise Exception("Error retrieving PID record: ", resource_response)

    return PIDRecord.fromJSON(
        resource_response.json()
    )  # parse a PID record from the response JSON and return it

updatePIDRecord 

updatePIDRecord(pidRecord: PIDRecord) -> PIDRecord

Updates a PID record in the TPM

Parameters:

Name Type Description Default
pidRecord PIDRecord

PIDRecord The PID record to update

 required

Returns:

Type Description
PIDRecord

PIDRecord The response from the TPM

Raises:

Type Description
ValueError

If the PID record is None or not an instance of PIDRecord
Exception

If the PID record cannot be updated
Source code in src/nmr_FAIR_DOs/connectors/tpm_connector.py 
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
def updatePIDRecord(self, pidRecord: PIDRecord) -> PIDRecord:
    """
    Updates a PID record in the TPM

    Args:
        pidRecord:PIDRecord The PID record to update

    Returns:
        PIDRecord The response from the TPM

    Raises:
        ValueError: If the PID record is None or not an instance of PIDRecord
        Exception: If the PID record cannot be updated
    """
    if pidRecord is None or not isinstance(
        pidRecord, PIDRecord
    ):  # if the PID record is None or not an instance of PIDRecord, raise an error
        raise ValueError(
            "PID record must not be None and must be an instance of PIDRecord"
        )

    headers = {"Content-Type": "application/json"}

    content = pidRecord.toJSON()  # get the JSON representation of the PID record

    endpoint = "/api/v1/pit/pid/" + pidRecord.getPID()  # create the endpoint URL

    if (
        content is None or len(content) == 0
    ):  # if the content is None or empty, raise an error
        raise ValueError("No content to update due to invalid input")

    resource_response = requests.put(
        self._tpm_url + endpoint, headers=headers, json=content
    )  # send a PUT request to the TPM to update the PID record

    if (
        resource_response.status_code != 200
    ):  # if the status code is not 200, raise an error
        raise Exception(
            "Error updating PID record: ",
            resource_response,
        )

    return PIDRecord.fromJSON(
        resource_response.json()
    )  # parse a PID record from the response JSON and return it

getAllPIDRecords async 

getAllPIDRecords() -> list[PIDRecord]

Retrieves all PID records from the TPM. This function uses the known-pid endpoint of the TPM to retrieve all PID records. Keep in mind that the TPM does not necessarily know all PIDs in the system or even in the prefix or same instance.

Returns:

Type Description
list[PIDRecord]

list[PIDRecord] The list of all PID records
Source code in src/nmr_FAIR_DOs/connectors/tpm_connector.py 
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
async def getAllPIDRecords(self) -> list[PIDRecord]:
    """
    Retrieves all PID records from the TPM.
    This function uses the known-pid endpoint of the TPM to retrieve all PID records.
    Keep in mind that the TPM does not necessarily know all PIDs in the system or even in the prefix or same instance.

    Returns:
        list[PIDRecord] The list of all PID records
    """
    endpoint = "/api/v1/pit/known-pid"

    resource_response = requests.get(
        self._tpm_url + endpoint, headers={"Accept": "application/json"}
    )  # send a GET request to the TPM to retrieve all PID records

    if (
        resource_response.status_code != 200
    ):  # if the status code is not 200, raise an error
        raise Exception("Error retrieving PID records: ", resource_response)

    url_template = Template(
        "$tpmURL/api/v1/pit/pid/$pid"
    )  # create a template for the URL

    single_pidRecord_urls = []
    for (
        i
    ) in resource_response.json():  # iterate over all PID records in the response
        # Create the URL
        url = url_template.safe_substitute(
            tpmURL=self._tpm_url,
            pid=i["pid"],
        )  # create the URL for the PID record
        single_pidRecord_urls.append(url)  # add the URL to the list

    # Fetch all PID records in a parallelized manner. Do not use the cache.
    json_records = await fetch_multiple(single_pidRecord_urls, True)

    result = []
    for i in json_records:  # iterate over all PID records in the response
        result.append(
            PIDRecord.fromJSON(i)
        )  # parse a PID record from the response JSON and add it to the result list

    return result  # return the list of all fetched PID records