Coverage for datacite/datacite.py: 100%

1import logging 

2from typing import Any 

3 

4import humps 

5import requests 

6from django.conf import settings 

7from requests.auth import HTTPBasicAuth 

8 

9# Specific User-Agent is asked by Datacite for contacting the client in case 

10# of abuse: https://support.datacite.org/docs/api#api-versions 

11DATACITE_USER_AGENT = f"Dalima ({settings.DC_URL}; mailto:{settings.DC_MAIL})" 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16class DataciteRESTClient: 

17 """ 

18 REST client for the Datacite API. 

19 """ 

20 

21 headers = {"User-Agent": DATACITE_USER_AGENT} 

22 authentication = HTTPBasicAuth(settings.DATACITE_USER, settings.DATACITE_PASSWORD) 

23 base_url = settings.DATACITE_BASE_URL 

24 

25 def __init__( 

26 self, 

27 base_url: str | None = None, 

28 user: str | None = None, 

29 password: str | None = None, 

30 ) -> None: 

31 if user and password: 

32 self.authentication = HTTPBasicAuth(user, password) 

33 if base_url: 

34 self.base_url = base_url 

35 

36 def get_doi_metadata(self, doi: str = "", params: dict | None = None) -> Any: 

37 """ 

38 Get the metadata associated to the given DOI from Datacite database 

39 :param doi: DOI to query 

40 :param params: parameters to the query 

41 :return: metadata of the given DOI or None if it could not be fetched 

42 """ 

43 resp = requests.get( 

44 url=self.base_url + "dois/" + doi, 

45 params=params, 

46 headers=self.headers, 

47 timeout=settings.DATACITE_TIMEOUT, 

48 auth=self.authentication, 

49 ) 

50 if resp.status_code == requests.codes.ok: 

51 logger.info( 

52 "Fetched metadata for %s from Datacite.", doi if doi else "__all__" 

53 ) 

54 return humps.decamelize(resp.json()["data"]) 

55 logger.warning( 

56 "Failed to fetched metadata for %s from Datacite.", 

57 doi if doi else "__all__", 

58 ) 

59 return None 

60 

61 def get_metadata_of_doi_list( 

62 self, doi_list: list, params: dict | None = None 

63 ) -> list: 

64 """ 

65 Get the metadata the DOIs of the given list from Datacite database 

66 :param doi_list: list of DOI to query 

67 :param params: parameters to the query 

68 :return: the list of metadata of the given DOI list. 

69 If none was fetched, it is not included 

70 """ 

71 metadata_list: list = [] 

72 for doi in doi_list: 

73 if not doi: 

74 continue 

75 doi_metadata = self.get_doi_metadata(doi, params) 

76 if doi_metadata is not None: 

77 metadata_list.append(doi_metadata) 

78 return metadata_list 

79 

80 def upsert_doi( 

81 self, 

82 doi: str, 

83 doi_metadata: dict, 

84 params: dict | None = None, 

85 ) -> Any: 

86 """ 

87 Creates or updates a DOI in the Datacite database with the given metadata 

88 and parameters 

89 :param doi: doi to modify. If None, it will create one. 

90 :param doi_metadata: doi metadata to insert into the Datacite database 

91 :param params: parameters for the requests 

92 :return: the dictionary of the response or none if status code != 200 or 201. 

93 """ 

94 resp = requests.put( 

95 url=self.base_url + "dois/" + doi, 

96 json={ 

97 "data": { 

98 "type": "dois", 

99 "attributes": humps.camelize(doi_metadata), 

100 } 

101 }, 

102 params=params, 

103 headers=self.headers, 

104 timeout=settings.DATACITE_TIMEOUT, 

105 auth=self.authentication, 

106 ) 

107 resp.raise_for_status() 

108 

109 return humps.decamelize(resp.json()["data"])