5576946992

pending completion

Build # 5576946992

Build Type

Pull #929

github

Committed by

web-flow

Commit Message

Merge 63e01a8fb into d0f35c594

Pull Request Pull Request #929: URN: redirect to the right URL

Run Details

192 of 192 new or added lines in 9 files covered. (100.0%)

7994 of 8284 relevant lines covered (96.5%)

1.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.5

/sonar/modules/documents/dnb.py

# -*- coding: utf-8 -*-
#
# Swiss Open Access Repository
# Copyright (C) 2022 RERO
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Dnb URN service API."""

import base64
import json

import requests
from flask import current_app


class DnbServerError(Exception):
    """The Dnb Server returns an error."""

class DnbUrnService:
    """Dnb URN service class."""

    @classmethod
    def base_url(cls):
        """Base DBN URL."""
        return f"{current_app.config.get('SONAR_APP_URN_DNB_BASE_URL')}/urns"

    @classmethod
    def headers(cls):
        """Set headers for queries.

        :returns: headers dictionary.
        """
        username = current_app.config.get('SONAR_APP_URN_DNB_USERNAME')
        password = current_app.config.get('SONAR_APP_URN_DNB_PASSWORD')
        string = f"{username}:{password}"
        auth = base64.b64encode(string.encode()).decode('ascii')
        return {
            'Authorization': f'Basic {auth}',
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }

    @classmethod
    def exists(cls, urn_code):
        """Check the existence for a URN.

        :param urn_code: the urn code.
        :returns: True if urn is exists, otherwise False.
        """
        # Documentation: https://wiki.dnb.de/display/URNSERVDOK/URN-Service+API
        # https://wiki.dnb.de/display/URNSERVDOK/Beispiele%3A+URN-Verwaltung
        try:
            response = requests.request(
                'HEAD',
                f"{cls.base_url()}/urn/{urn_code}",
                headers=cls.headers()
            )
            if not response.status_code in [200, 404]:
                raise DnbServerError(
                    f'Bad DNB sever response status {response.status_code}, '
                    f'when we check the existence of the following '
                    f'urn: {urn_code}')
            return response.status_code == 200
        except Exception as error:
            raise DnbServerError(error)

    @classmethod
    def get_urls(cls, urn_code):
        """Get the URN information.

        :param urn_code: the urn code.
        :returns: True if urn is registered, otherwise False.
        """
        # Documentation: https://wiki.dnb.de/display/URNSERVDOK/URN-Service+API
        # https://wiki.dnb.de/display/URNSERVDOK/Beispiele%3A+URN-Verwaltung
        answer = False
        try:
            response = requests.get(
                f'{cls.base_url()}/urn/{urn_code}/urls',
                headers=cls.headers()
            )
            if not response.status_code == 200:
                raise DnbServerError(
                    f'Bad DNB sever response status {response.status_code}, '
                    f'when we get the information of the following '
                    f'urn: {urn_code}')
            return response.json()
        except Exception as error:
            raise DnbServerError(error)


    @classmethod
    def get(cls, urn_code):
        """Get the URN information.

        :param urn_code: the urn code.
        :returns: True if urn is registered, otherwise False.
        """
        # Documentation: https://wiki.dnb.de/display/URNSERVDOK/URN-Service+API
        # https://wiki.dnb.de/display/URNSERVDOK/Beispiele%3A+URN-Verwaltung
        answer = False
        try:
            response = requests.get(
                f'{cls.base_url()}/urn/{urn_code}',
                headers=cls.headers()
            )
            if response.status_code != 200:
                raise DnbServerError(
                    f'Bad DNB sever response status {response.status_code}, '
                    f'when we get the information of the following '
                    f'urn: {urn_code}')
            return response.json()
        except Exception as error:
            raise DnbServerError(error) from error

    @classmethod
    def update(cls, urn_code, urls):
        """Update the list of urs registered to a given URN."""
        response = requests.request(
            'PATCH',
            f"{cls.base_url()}/urn/{urn_code}/my-urls",
            headers=cls.headers(),
            data=json.dumps(urls)
        )
        if response.status_code != 204:
            raise DnbServerError(
                f'Bad DNB sever response status {response.status_code}, '
                f'when we update the information of the following '
                f'urn: {urn_code}')

    @classmethod
    def create(cls, data):
        """Register a new URN to the DBN service with a list of urls."""
        response = requests.request(
            'POST',
            cls.base_url(),
            headers=cls.headers(),
            data=json.dumps(data)
        )
        if response.status_code != 201:
            raise DnbServerError(
                f'Bad DNB sever response status {response.status_code}, '
                f'when we update the information of the following '
                f'urn: {data.get("urn")}')


    @classmethod
    def register_document(cls, document):
        """Register a new URN code.

        :param document: the document.
        :returns: True if urn is registered, otherwise False.
        """
        from sonar.modules.documents.api import DocumentRecord
        if not isinstance(document, DocumentRecord):
            document = DocumentRecord(document)
        if url := cls.get_url(document):
            urn = document.get_rero_urn_code(document)
            data = {
                'urn': urn,
                'urls': [
                    {
                        'url': url,
                        'priority': 1
                    }
                ]
            }
            if cls.exists(urn):
                cls.update(urn, data['urls'])
            else:
                cls.create(data)
            return True
        return False

    @classmethod
    def get_url(cls, document):
        """."""
        base_url = 'https://' + current_app.config.get(
            'SONAR_APP_SERVER_NAME')
        if orgs := document.resolve().get('organisation', []):
            org_code = current_app.config.get('SONAR_APP_DEFAULT_ORGANISATION')
            if org := orgs[0]:
                if org.get('isDedicated') or org.get('isShared'):
                    org_code = org.get('code')
                if org.get('isDedicated') and (server_name := org.get('serverName')):
                    base_url = f'https://{server_name}'
            return f"{base_url}/{org_code}/documents/{document.get('pid')}"

1	# -- coding: utf-8 --
2	#
3	# Swiss Open Access Repository
4	# Copyright (C) 2022 RERO
5	#
6	# This program is free software: you can redistribute it and/or modify
7	# it under the terms of the GNU Affero General Public License as published by
8	# the Free Software Foundation, version 3 of the License.
9	#
10	# This program is distributed in the hope that it will be useful,
11	# but WITHOUT ANY WARRANTY; without even the implied warranty of
12	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	# GNU Affero General Public License for more details.
14	#
15	# You should have received a copy of the GNU Affero General Public License
16	# along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18	"""Dnb URN service API."""	2✔
19
20	import base64	2✔
21	import json	2✔
22
23	import requests	2✔
24	from flask import current_app	2✔
25
26
27	class DnbServerError(Exception):	2✔
28	"""The Dnb Server returns an error."""
29
30	class DnbUrnService:	2✔
31	"""Dnb URN service class."""
32
33	@classmethod	2✔
34	def base_url(cls):	2✔
35	"""Base DBN URL."""
36	return f"{current_app.config.get('SONAR_APP_URN_DNB_BASE_URL')}/urns"	2✔
37
38	@classmethod	2✔
39	def headers(cls):	2✔
40	"""Set headers for queries.
41
42	:returns: headers dictionary.
43	"""
44	username = current_app.config.get('SONAR_APP_URN_DNB_USERNAME')	2✔
45	password = current_app.config.get('SONAR_APP_URN_DNB_PASSWORD')	2✔
46	string = f"{username}:{password}"	2✔
47	auth = base64.b64encode(string.encode()).decode('ascii')	2✔
48	return {	2✔
49	'Authorization': f'Basic {auth}',
50	'Content-Type': 'application/json',
51	'Accept': 'application/json'
52	}
53
54	@classmethod	2✔
55	def exists(cls, urn_code):	2✔
56	"""Check the existence for a URN.
57
58	:param urn_code: the urn code.
59	:returns: True if urn is exists, otherwise False.
60	"""
61	# Documentation: https://wiki.dnb.de/display/URNSERVDOK/URN-Service+API
62	# https://wiki.dnb.de/display/URNSERVDOK/Beispiele%3A+URN-Verwaltung
63	try:	2✔
64	response = requests.request(	2✔
65	'HEAD',
66	f"{cls.base_url()}/urn/{urn_code}",
67	headers=cls.headers()
68	)
69	if not response.status_code in [200, 404]:	2✔
70	raise DnbServerError(	×
71	f'Bad DNB sever response status {response.status_code}, '
72	f'when we check the existence of the following '
73	f'urn: {urn_code}')
74	return response.status_code == 200	2✔
75	except Exception as error:	×
76	raise DnbServerError(error)	×
77
78	@classmethod	2✔
79	def get_urls(cls, urn_code):	2✔
80	"""Get the URN information.
81
82	:param urn_code: the urn code.
83	:returns: True if urn is registered, otherwise False.
84	"""
85	# Documentation: https://wiki.dnb.de/display/URNSERVDOK/URN-Service+API
86	# https://wiki.dnb.de/display/URNSERVDOK/Beispiele%3A+URN-Verwaltung
87	answer = False	×
88	try:	×
89	response = requests.get(	×
90	f'{cls.base_url()}/urn/{urn_code}/urls',
91	headers=cls.headers()
92	)
93	if not response.status_code == 200:	×
94	raise DnbServerError(	×
95	f'Bad DNB sever response status {response.status_code}, '
96	f'when we get the information of the following '
97	f'urn: {urn_code}')
98	return response.json()	×
99	except Exception as error:	×
100	raise DnbServerError(error)	×
101
102
103	@classmethod	2✔
104	def get(cls, urn_code):	2✔
105	"""Get the URN information.
106
107	:param urn_code: the urn code.
108	:returns: True if urn is registered, otherwise False.
109	"""
110	# Documentation: https://wiki.dnb.de/display/URNSERVDOK/URN-Service+API
111	# https://wiki.dnb.de/display/URNSERVDOK/Beispiele%3A+URN-Verwaltung
112	answer = False	×
113	try:	×
114	response = requests.get(	×
115	f'{cls.base_url()}/urn/{urn_code}',
116	headers=cls.headers()
117	)
118	if response.status_code != 200:	×
119	raise DnbServerError(	×
120	f'Bad DNB sever response status {response.status_code}, '
121	f'when we get the information of the following '
122	f'urn: {urn_code}')
123	return response.json()	×
124	except Exception as error:	×
125	raise DnbServerError(error) from error	×
126
127	@classmethod	2✔
128	def update(cls, urn_code, urls):	2✔
129	"""Update the list of urs registered to a given URN."""
130	response = requests.request(	×
131	'PATCH',
132	f"{cls.base_url()}/urn/{urn_code}/my-urls",
133	headers=cls.headers(),
134	data=json.dumps(urls)
135	)
136	if response.status_code != 204:	×
137	raise DnbServerError(	×
138	f'Bad DNB sever response status {response.status_code}, '
139	f'when we update the information of the following '
140	f'urn: {urn_code}')
141
142	@classmethod	2✔
143	def create(cls, data):	2✔
144	"""Register a new URN to the DBN service with a list of urls."""
145	response = requests.request(	2✔
146	'POST',
147	cls.base_url(),
148	headers=cls.headers(),
149	data=json.dumps(data)
150	)
151	if response.status_code != 201:	2✔
152	raise DnbServerError(	×
153	f'Bad DNB sever response status {response.status_code}, '
154	f'when we update the information of the following '
155	f'urn: {data.get("urn")}')
156
157
158	@classmethod	2✔
159	def register_document(cls, document):	2✔
160	"""Register a new URN code.
161
162	:param document: the document.
163	:returns: True if urn is registered, otherwise False.
164	"""
165	from sonar.modules.documents.api import DocumentRecord	2✔
166	if not isinstance(document, DocumentRecord):	2✔
167	document = DocumentRecord(document)	×
168	if url := cls.get_url(document):	2✔
169	urn = document.get_rero_urn_code(document)	2✔
170	data = {	2✔
171	'urn': urn,
172	'urls': [
173	{
174	'url': url,
175	'priority': 1
176	}
177	]
178	}
179	if cls.exists(urn):	2✔
180	cls.update(urn, data['urls'])	×
181	else:
182	cls.create(data)	2✔
183	return True	2✔
184	return False	×
185
186	@classmethod	2✔
187	def get_url(cls, document):	2✔
188	"""."""
189	base_url = 'https://' + current_app.config.get(	2✔
190	'SONAR_APP_SERVER_NAME')
191	if orgs := document.resolve().get('organisation', []):	2✔
192	org_code = current_app.config.get('SONAR_APP_DEFAULT_ORGANISATION')	2✔
193	if org := orgs[0]:	2✔
194	if org.get('isDedicated') or org.get('isShared'):	2✔
195	org_code = org.get('code')	2✔
196	if org.get('isDedicated') and (server_name := org.get('serverName')):	2✔
197	base_url = f'https://{server_name}'	2✔
198	return f"{base_url}/{org_code}/documents/{document.get('pid')}"	2✔

rero / sonar / 5576946992

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous