src/pds_doi_service/core/outputs/osti/osti_web_client.py
def query_doi(self, query, url=None, username=None, password=None, content_type=CONTENT_TYPE_XML):
        """
        Queries the status of a DOI from the OSTI server and returns the
        response text.

        Parameters
        ----------
        query : dict
            Key/value pairs to append as parameters to the URL for the GET
            endpoint.
        url : str, optional
            The URL to submit the request to. If not submitted, it is pulled
            from the INI config for the appropriate service provider.
        username : str, optional
            The username to authenticate the request as. If not submitted, it
            is pulled from the INI config OSTI user field.
        password : str, optional
            The password to authenticate the request with. If not submitted, it
            is pulled from the INI config OSTI password field.
        content_type : str, optional
            The content type to specify the format of the payload, as well as
            the format of the response from OSTI. Currently, 'xml' and 'json'
            are supported. Defaults to xml.

        Returns
        -------
        response_text : str
            Body of the response text from the endpoint.

        """
        config = self._config_util.get_config()

        if content_type not in self._content_type_map:
            raise ValueError(
                "Invalid content type requested, must be one of " f'{",".join(list(self._content_type_map.keys()))}'
            )

        auth = HTTPBasicAuth(username or config.get("OSTI", "user"), password or config.get("OSTI", "password"))

        headers = {"Accept": self._content_type_map[content_type], "Content-Type": self._content_type_map[content_type]}

        # OSTI server requires 'rows' field to know how many max rows to fetch at once.
        initial_payload = {"rows": self.MAX_TOTAL_ROWS_RETRIEVE}

        # If user provided a query_dict, append to our initial payload.
        if query:
            initial_payload.update(query)
            # Do a sanity check and only fetch valid field names.
            query = self._validate_field_names(initial_payload)
        else:
            query = initial_payload

        url = url or config.get("OSTI", "url")

        logger.debug("initial_payload: %s", initial_payload)
        logger.debug("query_dict: %s", query)
        logger.debug("url: %s", url)

        osti_response = requests.get(url=url, auth=auth, params=query, headers=headers)

        try:
            osti_response.raise_for_status()
        except requests.exceptions.HTTPError as http_err:
            # Detail text is not always present, which can cause json parsing
            # issues
            details = f"Details: {pprint.pformat(json.loads(osti_response.text))}" if osti_response.text else ""

            raise WebRequestException(
                "DOI submission request to OSTI service failed, " f"reason: {str(http_err)}\n{details}"
            )

        return osti_response.text

        Queries the status of a DOI from the OSTI server and returns the
        response text.

        Parameters
        ----------
        query : dict
            Key/value pairs to append as parameters to the URL for the GET
            endpoint.
        url : str, optional
            The URL to submit the request to. If not submitted, it is pulled
            from the INI config for the appropriate service provider.
        username : str, optional
            The username to authenticate the request as. If not submitted, it
            is pulled from the INI config OSTI user field.
        password : str, optional
            The password to authenticate the request with. If not submitted, it
            is pulled from the INI config OSTI password field.
        content_type : str, optional
            The content type to specify the format of the payload, as well as
            the format of the response from OSTI. Currently, 'xml' and 'json'
            are supported. Defaults to xml.

        Returns
        -------
        response_text : str
            Body of the response text from the endpoint.

        
Queries the status of a DOI from the OSTI server and returns the
response text.

Parameters

query : dict
Key/value pairs to append as parameters to the URL for the GET
endpoint.
url : str, optional
The URL to submit the request to. If not submitted, it is pulled
from the INI config for the appropriate service provider.
username : str, optional
The username to authenticate the request as. If not submitted, it
is pulled from the INI config OSTI user field.
password : str, optional
The password to authenticate the request with. If not submitted, it
is pulled from the INI config OSTI password field.
content_type : str, optional
The content type to specify the format of the payload, as well as
the format of the response from OSTI. Currently, 'xml' and 'json'
are supported. Defaults to xml.

Returns

response_text : str
Body of the response text from the endpoint.
Queries the status of a DOI from the OSTI server and returns the response text . Parameters query : dict Key / value pairs to append as parameters to the URL for the GET endpoint . url : str optional The URL to submit the request to . If not submitted it is pulled from the INI config for the appropriate service provider . username : str optional The username to authenticate the request as . If not submitted it is pulled from the INI config OSTI user field . password : str optional The password to authenticate the request with . If not submitted it is pulled from the INI config OSTI password field . content_type : str optional The content type to specify the format of the payload as well as the format of the response from OSTI . Currently ' xml ' and ' json ' are supported . Defaults to xml . Returns response_text : str Body of the response text from the endpoint .
def query_doi(self, query, url=None, username=None, password=None, content_type=CONTENT_TYPE_XML):
        config = self._config_util.get_config()
        if content_type not in self._content_type_map:
            raise ValueError(
                "Invalid content type requested, must be one of " f'{",".join(list(self._content_type_map.keys()))}'
            )
        auth = HTTPBasicAuth(username or config.get("OSTI", "user"), password or config.get("OSTI", "password"))
        headers = {"Accept": self._content_type_map[content_type], "Content-Type": self._content_type_map[content_type]}
        initial_payload = {"rows": self.MAX_TOTAL_ROWS_RETRIEVE}
        if query:
            initial_payload.update(query)
            query = self._validate_field_names(initial_payload)
        else:
            query = initial_payload
        url = url or config.get("OSTI", "url")
        logger.debug("initial_payload: %s", initial_payload)
        logger.debug("query_dict: %s", query)
        logger.debug("url: %s", url)
        osti_response = requests.get(url=url, auth=auth, params=query, headers=headers)
        try:
            osti_response.raise_for_status()
        except requests.exceptions.HTTPError as http_err:
            details = f"Details: {pprint.pformat(json.loads(osti_response.text))}" if osti_response.text else ""
            raise WebRequestException(
                "DOI submission request to OSTI service failed, " f"reason: {str(http_err)}\n{details}"
            )
        return osti_response.text
def query_doi ( self , query , url = None , username = None , password = None , content_type = CONTENT_TYPE_XML ) : config = self . _config_util . get_config ( ) if content_type not in self . _content_type_map : raise ValueError ( "Invalid content type requested, must be one of " f'{",".join(list(self._content_type_map.keys()))}' ) auth = HTTPBasicAuth ( username or config . get ( "OSTI" , "user" ) , password or config . get ( "OSTI" , "password" ) ) headers = { "Accept" : self . _content_type_map [ content_type ] , "Content-Type" : self . _content_type_map [ content_type ] } initial_payload = { "rows" : self . MAX_TOTAL_ROWS_RETRIEVE } if query : initial_payload . update ( query ) query = self . _validate_field_names ( initial_payload ) else : query = initial_payload url = url or config . get ( "OSTI" , "url" ) logger . debug ( "initial_payload: %s" , initial_payload ) logger . debug ( "query_dict: %s" , query ) logger . debug ( "url: %s" , url ) osti_response = requests . get ( url = url , auth = auth , params = query , headers = headers ) try : osti_response . raise_for_status ( ) except requests . exceptions . HTTPError as http_err : details = f"Details: {pprint.pformat(json.loads(osti_response.text))}" if osti_response . text else "" raise WebRequestException ( "DOI submission request to OSTI service failed, " f"reason: {str(http_err)}\n{details}" ) return osti_response . text
Queries the status of a DOI from the OSTI server and returns the
response text.
Queries the status of a DOI from the OSTI server and returns the response text .
"""
        Queries the status of a DOI from the OSTI server and returns the
        response text.

        Parameters
        ----------
        query : dict
            Key/value pairs to append as parameters to the URL for the GET
            endpoint.
        url : str, optional
            The URL to submit the request to. If not submitted, it is pulled
            from the INI config for the appropriate service provider.
        username : str, optional
            The username to authenticate the request as. If not submitted, it
            is pulled from the INI config OSTI user field.
        password : str, optional
            The password to authenticate the request with. If not submitted, it
            is pulled from the INI config OSTI password field.
        content_type : str, optional
            The content type to specify the format of the payload, as well as
            the format of the response from OSTI. Currently, 'xml' and 'json'
            are supported. Defaults to xml.

        Returns
        -------
        response_text : str
            Body of the response text from the endpoint.

        """ # OSTI server requires 'rows' field to know how many max rows to fetch at once. # If user provided a query_dict, append to our initial payload. # Do a sanity check and only fetch valid field names. # Detail text is not always present, which can cause json parsing # issues
self
query
url
username
password
content_type