Source code for nmdc_client.minter

# -*- coding: utf-8 -*-
import json
import logging

import requests

from nmdc_client.api_client import NMDCAPIClient
from nmdc_client.auth import NMDCAuth
from nmdc_client.config import API_BASE_URL
from nmdc_client.decorators import has_deprecated_parameter, requires_auth

logger = logging.getLogger(__name__)


[docs] @has_deprecated_parameter("env", reason="Use ``api_base_url`` instead.") class Minter(NMDCAPIClient): """ Class to interact with the NMDC API to mint new identifiers. Parameters ---------- api_base_url The base URL of the NMDC API. auth An instance of the NMDCAuth class for authentication. """ def __init__( self, api_base_url: str = API_BASE_URL, auth: NMDCAuth | None = None, env: str = "", ): super().__init__( api_base_url=api_base_url, env=env, ) self.auth = auth or NMDCAuth(api_base_url=self.api_base_url)
[docs] @requires_auth def mint( self, nmdc_type: str, count: int = 1, client_id: str | None = None, client_secret: str | None = None, ) -> str | list[str]: """ Mint new identifier(s) for a specified type of record. Parameters ---------- nmdc_type The type of NMDC ID to mint (e.g., 'nmdc:MassSpectrometry', 'nmdc:DataObject'). count The number of identifiers to mint. client_id The client ID for authentication. Kept for backwards compatibility. client_secret The client secret for authentication. Kept for backwards compatibility. Returns ------- str or list[str] If count is 1, returns a single minted identifier as a string. If count is greater than 1, returns a list of minted identifiers. Raises ------ RuntimeError If the API request fails. ValueError If count is less than 1. Notes ----- If ``client_id`` and ``client_secret`` are provided, a new instance of the ``NMDCAuth`` class will be created. The newest and preferred method for authentication is to use the ``NMDCAuth`` class directly. """ # if they are passed into the function, create the auth object if client_id and client_secret: self.auth = NMDCAuth( client_id=client_id, client_secret=client_secret, api_base_url=self.api_base_url, ) # Validate count parameter if count < 1: raise ValueError("count must be at least 1") # get the token token = self.auth.get_token() url = f"{self.api_base_url}/pids/mint" payload = {"schema_class": {"id": nmdc_type}, "how_many": count} try: response = requests.post( url, headers=self._build_http_request_headers( access_token=token, content_type="application/json", ), data=json.dumps(payload), ) response.raise_for_status() except requests.exceptions.RequestException as e: logger.error("API request failed", exc_info=True) raise RuntimeError("Failed to mint new identifier from NMDC API") from e else: logging.debug( f"API request response: {response.json()}\n API Status Code: {response.status_code}" ) # return the response response_data = response.json() if count == 1: return response_data[0] else: return response_data