17425918180

Committed 03 Sep 2025 07:11AM UTC coverage: 95.796% (-0.6%) from 96.378%

Build # 17425918180

Build Type

push

github

Committed by

PascalRepond

Commit Message

translations: extract messages

Co-Authored-by: Pascal Repond <pascal.repond@rero.ch>

Run Details

7816 of 8159 relevant lines covered (95.8%)

0.96 hits per line

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

99.39

/sonar/modules/documents/marshmallow/json.py

# Swiss Open Access Repository
# Copyright (C) 2021 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/>.

"""JSON Schemas."""

from functools import partial

from flask import current_app, request
from invenio_records_rest.schemas import Nested, StrictKeysMixin
from invenio_records_rest.schemas.fields import (
    GenFunction,
    PersistentIdentifier,
    SanitizedUnicode,
)
from marshmallow import EXCLUDE, fields, pre_dump, pre_load, validate

from sonar.modules.documents.api import DocumentRecord
from sonar.modules.documents.permissions import (
    DocumentFilesPermission,
    DocumentPermission,
)
from sonar.modules.documents.utils import (
    has_external_urls_for_files,
    populate_files_properties,
)
from sonar.modules.documents.views import (
    contribution_text,
    create_publication_statement,
    dissertation,
    part_of_format,
)
from sonar.modules.serializers import schema_from_context
from sonar.modules.users.api import current_user_record

schema_from_document = partial(schema_from_context, schema=DocumentRecord.schema)


class ThumbnailSchemaV1(StrictKeysMixin):
    """Thumbnail only schema."""

    class Meta:
        """Meta for file schema."""

        unknown = EXCLUDE

    key = fields.Str()
    label = SanitizedUnicode()
    type = SanitizedUnicode()
    order = fields.Integer()
    restriction = fields.Dict(dump_only=True)
    links = fields.Dict(dump_only=True)
    thumbnail = SanitizedUnicode(dump_only=True)


class FileSchemaV1(ThumbnailSchemaV1):
    """File schema."""

    class Meta:
        """Meta for file schema."""

        # Specifically exclude unknown fields, as in the new version of
        # marshmallow, dump_only fields are treated as included.
        # https://github.com/marshmallow-code/marshmallow/issues/875
        unknown = EXCLUDE

    bucket = SanitizedUnicode()
    file_id = SanitizedUnicode()
    version_id = SanitizedUnicode()
    mimetype = SanitizedUnicode()
    checksum = SanitizedUnicode()
    size = fields.Integer()
    external_url = SanitizedUnicode()
    access = SanitizedUnicode()
    restricted_outside_organisation = fields.Boolean()
    embargo_date = SanitizedUnicode()
    permissions = fields.Dict(dump_only=True)

    @pre_dump
    def add_permissions(self, item, **kwargs):
        """Add permissions to record.

        :param item: Dict representing the record.
        :returns: Modified dict.
        """
        if not item.get("bucket"):
            return item
        doc = DocumentRecord.get_record_by_bucket(item.get("bucket"))
        item["permissions"] = {
            "read": DocumentFilesPermission.read(current_user_record, item, doc["pid"], doc),
            "update": DocumentFilesPermission.update(current_user_record, item, doc["pid"], doc),
            "delete": DocumentFilesPermission.delete(current_user_record, item, doc["pid"], doc),
        }

        return item

    @pre_load
    def remove_fields(self, data, **kwargs):
        """Removes computed fields.

        :param data: Dict of record data.
        :returns: Modified data.
        """
        data.pop("permissions", None)
        return data


class DocumentListMetadataSchemaV1(StrictKeysMixin):
    """Schema for the document metadata."""

    pid = PersistentIdentifier()
    ark = SanitizedUnicode()
    documentType = SanitizedUnicode()
    title = fields.List(fields.Dict())
    partOf = fields.List(fields.Dict())
    abstracts = fields.List(fields.Dict())
    contribution = fields.List(fields.Dict())
    organisation = fields.List(fields.Dict())
    language = fields.List(fields.Dict())
    copyrightDate = fields.List(fields.String())
    editionStatement = fields.Dict()
    provisionActivity = fields.List(fields.Dict())
    extent = SanitizedUnicode()
    otherMaterialCharacteristics = SanitizedUnicode()
    formats = fields.List(SanitizedUnicode())
    additionalMaterials = SanitizedUnicode()
    series = fields.List(fields.Dict())
    notes = fields.List(fields.String())
    identifiedBy = fields.List(fields.Dict())
    identifiers = fields.Dict()
    subjects = fields.List(fields.Dict())
    classification = fields.List(fields.Dict())
    collections = fields.List(fields.Dict())
    dissertation = fields.Dict()
    otherEdition = fields.List(fields.Dict())
    relatedTo = fields.List(fields.Dict())
    usageAndAccessPolicy = fields.Dict()
    projects = fields.List(fields.Dict())
    oa_status = SanitizedUnicode()
    subdivisions = fields.List(fields.Dict())
    harvested = fields.Boolean()
    contentNote = fields.List(SanitizedUnicode())
    customField1 = fields.List(fields.String(validate=validate.Length(min=1)))
    customField2 = fields.List(fields.String(validate=validate.Length(min=1)))
    customField3 = fields.List(fields.String(validate=validate.Length(min=1)))
    masked = SanitizedUnicode()
    _bucket = SanitizedUnicode()
    _files = Nested(ThumbnailSchemaV1, many=True)
    _oai = fields.Dict()
    # When loading, if $schema is not provided, it's retrieved by
    # Record.schema property.
    schema = GenFunction(
        load_only=True,
        attribute="$schema",
        data_key="$schema",
        deserialize=schema_from_document,
    )
    permissions = fields.Dict(dump_only=True)
    permalink = SanitizedUnicode(dump_only=True)

    @pre_dump
    def populate_files_properties(self, item, **kwargs):
        """Add some customs properties to file before dumping it.

        :param item: Item object to process
        :returns: Modified item
        """
        if not item.get("_files"):
            return item

        # Check if organisation record forces to point file to an external url
        item["external_url"] = has_external_urls_for_files(item)

        # Add restriction, link and thumbnail to files
        populate_files_properties(item)

        # Sort files to have the main file in first position
        item["_files"] = sorted(item["_files"], key=lambda file: file.get("order", 100))

        return item

    @pre_dump
    def add_permissions(self, item, **kwargs):
        """Add permissions to record.

        :param item: Dict representing the record.
        :returns: Modified dict.
        """
        # For public views, no check for permissions
        if request.args.get("view"):
            return item

        item["permissions"] = {
            "read": DocumentPermission.read(current_user_record, item),
            "update": DocumentPermission.update(current_user_record, item),
            "delete": DocumentPermission.delete(current_user_record, item),
        }

        return item

    @pre_dump
    def add_permalink(self, item, **kwargs):
        """Add permanent link to document."""
        item["permalink"] = DocumentRecord.get_permanent_link(host=request.host_url, pid=item["pid"])
        return item

    @pre_dump
    def add_ark_uri(self, item, **kwargs):
        """Add permanent link to document."""
        resolver_url = current_app.config.get("SONAR_APP_ARK_RESOLVER")
        for itm in item.get("identifiedBy", []):
            if itm.get("type") == "ark":
                itm["uri"] = f"{resolver_url}/{itm['value']}"
                break
        return item

    @pre_dump
    def add_formatted_texts(self, item, **kwargs):
        """Add formatted texts for objects which are processing in backend.

        :param item: Dict of record data.
        :returns: Modified data.
        """
        # Provision activity processing
        for index, provision_activity in enumerate(item.get("provisionActivity", [])):
            item["provisionActivity"][index]["text"] = create_publication_statement(provision_activity)

        # Part of proccessing
        for index, part_of in enumerate(item.get("partOf", [])):
            item["partOf"][index]["text"] = part_of_format(part_of)

        # Contribution
        for index, contribution in enumerate(item.get("contribution", [])):
            item["contribution"][index]["text"] = contribution_text(contribution)

        if item.get("dissertation"):
            item["dissertation"]["text"] = dissertation(item)

        return item

    @pre_load
    def guess_organisation(self, data, **kwargs):
        """Guess organisation from current logged user.

        :param data: Dict of record data.
        :returns: Modified dict of record data.
        """
        # Organisation already attached to document, we do nothing.
        if data.get("organisation"):
            return data

        # Store current user organisation in new document.
        if current_user_record.get("organisation"):
            data["organisation"] = [current_user_record["organisation"]]

        return data

    @pre_load
    def remove_fields(self, data, **kwargs):
        """Removes computed fields.

        :param data: Dict of record data.
        :returns: Modified data.
        """
        for provision_activity in data.get("provisionActivity", []):
            provision_activity.pop("text", None)

        for part_of in data.get("partOf", []):
            part_of.pop("text", None)

        for contribution in data.get("contribution", []):
            contribution.pop("text", None)

        data.get("dissertation", {}).pop("text", None)

        data.pop("permalink", None)
        data.pop("permissions", None)
        data.pop("external_url", None)
        for itm in data.get("identifiedBy", []):
            if itm.get("type") == "ark":
                itm.pop("uri", None)
                break

        return data


class DocumentMetadataSchemaV1(DocumentListMetadataSchemaV1):
    """Schema for the document metadata."""

    _files = Nested(FileSchemaV1, many=True)


class DocumentSchemaV1(StrictKeysMixin):
    """Document schema."""

    id = PersistentIdentifier()
    metadata = fields.Nested(DocumentMetadataSchemaV1)
    links = fields.Dict(dump_only=True)
    explanation = fields.Raw(dump_only=True)


class DocumentListSchemaV1(StrictKeysMixin):
    """Document schema."""

    id = PersistentIdentifier()
    metadata = fields.Nested(DocumentListMetadataSchemaV1)
    links = fields.Dict(dump_only=True)
    explanation = fields.Raw(dump_only=True)

1	# Swiss Open Access Repository
2	# Copyright (C) 2021 RERO
3	#
4	# This program is free software: you can redistribute it and/or modify
5	# it under the terms of the GNU Affero General Public License as published by
6	# the Free Software Foundation, version 3 of the License.
7	#
8	# This program is distributed in the hope that it will be useful,
9	# but WITHOUT ANY WARRANTY; without even the implied warranty of
10	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11	# GNU Affero General Public License for more details.
12	#
13	# You should have received a copy of the GNU Affero General Public License
14	# along with this program. If not, see <http://www.gnu.org/licenses/>.
15
16	"""JSON Schemas."""
17
18	from functools import partial	1✔
19
20	from flask import current_app, request	1✔
21	from invenio_records_rest.schemas import Nested, StrictKeysMixin	1✔
22	from invenio_records_rest.schemas.fields import (	1✔
23	GenFunction,
24	PersistentIdentifier,
25	SanitizedUnicode,
26	)
27	from marshmallow import EXCLUDE, fields, pre_dump, pre_load, validate	1✔
28
29	from sonar.modules.documents.api import DocumentRecord	1✔
30	from sonar.modules.documents.permissions import (	1✔
31	DocumentFilesPermission,
32	DocumentPermission,
33	)
34	from sonar.modules.documents.utils import (	1✔
35	has_external_urls_for_files,
36	populate_files_properties,
37	)
38	from sonar.modules.documents.views import (	1✔
39	contribution_text,
40	create_publication_statement,
41	dissertation,
42	part_of_format,
43	)
44	from sonar.modules.serializers import schema_from_context	1✔
45	from sonar.modules.users.api import current_user_record	1✔
46
47	schema_from_document = partial(schema_from_context, schema=DocumentRecord.schema)	1✔
48
49
50	class ThumbnailSchemaV1(StrictKeysMixin):	1✔
51	"""Thumbnail only schema."""
52
53	class Meta:	1✔
54	"""Meta for file schema."""
55
56	unknown = EXCLUDE	1✔
57
58	key = fields.Str()	1✔
59	label = SanitizedUnicode()	1✔
60	type = SanitizedUnicode()	1✔
61	order = fields.Integer()	1✔
62	restriction = fields.Dict(dump_only=True)	1✔
63	links = fields.Dict(dump_only=True)	1✔
64	thumbnail = SanitizedUnicode(dump_only=True)	1✔
65
66
67	class FileSchemaV1(ThumbnailSchemaV1):	1✔
68	"""File schema."""
69
70	class Meta:	1✔
71	"""Meta for file schema."""
72
73	# Specifically exclude unknown fields, as in the new version of
74	# marshmallow, dump_only fields are treated as included.
75	# https://github.com/marshmallow-code/marshmallow/issues/875
76	unknown = EXCLUDE	1✔
77
78	bucket = SanitizedUnicode()	1✔
79	file_id = SanitizedUnicode()	1✔
80	version_id = SanitizedUnicode()	1✔
81	mimetype = SanitizedUnicode()	1✔
82	checksum = SanitizedUnicode()	1✔
83	size = fields.Integer()	1✔
84	external_url = SanitizedUnicode()	1✔
85	access = SanitizedUnicode()	1✔
86	restricted_outside_organisation = fields.Boolean()	1✔
87	embargo_date = SanitizedUnicode()	1✔
88	permissions = fields.Dict(dump_only=True)	1✔
89
90	@pre_dump	1✔
91	def add_permissions(self, item, **kwargs):	1✔
92	"""Add permissions to record.
93
94	:param item: Dict representing the record.
95	:returns: Modified dict.
96	"""
97	if not item.get("bucket"):	1✔
98	return item	×
99	doc = DocumentRecord.get_record_by_bucket(item.get("bucket"))	1✔
100	item["permissions"] = {	1✔
101	"read": DocumentFilesPermission.read(current_user_record, item, doc["pid"], doc),
102	"update": DocumentFilesPermission.update(current_user_record, item, doc["pid"], doc),
103	"delete": DocumentFilesPermission.delete(current_user_record, item, doc["pid"], doc),
104	}
105
106	return item	1✔
107
108	@pre_load	1✔
109	def remove_fields(self, data, **kwargs):	1✔
110	"""Removes computed fields.
111
112	:param data: Dict of record data.
113	:returns: Modified data.
114	"""
115	data.pop("permissions", None)	1✔
116	return data	1✔
117
118
119	class DocumentListMetadataSchemaV1(StrictKeysMixin):	1✔
120	"""Schema for the document metadata."""
121
122	pid = PersistentIdentifier()	1✔
123	ark = SanitizedUnicode()	1✔
124	documentType = SanitizedUnicode()	1✔
125	title = fields.List(fields.Dict())	1✔
126	partOf = fields.List(fields.Dict())	1✔
127	abstracts = fields.List(fields.Dict())	1✔
128	contribution = fields.List(fields.Dict())	1✔
129	organisation = fields.List(fields.Dict())	1✔
130	language = fields.List(fields.Dict())	1✔
131	copyrightDate = fields.List(fields.String())	1✔
132	editionStatement = fields.Dict()	1✔
133	provisionActivity = fields.List(fields.Dict())	1✔
134	extent = SanitizedUnicode()	1✔
135	otherMaterialCharacteristics = SanitizedUnicode()	1✔
136	formats = fields.List(SanitizedUnicode())	1✔
137	additionalMaterials = SanitizedUnicode()	1✔
138	series = fields.List(fields.Dict())	1✔
139	notes = fields.List(fields.String())	1✔
140	identifiedBy = fields.List(fields.Dict())	1✔
141	identifiers = fields.Dict()	1✔
142	subjects = fields.List(fields.Dict())	1✔
143	classification = fields.List(fields.Dict())	1✔
144	collections = fields.List(fields.Dict())	1✔
145	dissertation = fields.Dict()	1✔
146	otherEdition = fields.List(fields.Dict())	1✔
147	relatedTo = fields.List(fields.Dict())	1✔
148	usageAndAccessPolicy = fields.Dict()	1✔
149	projects = fields.List(fields.Dict())	1✔
150	oa_status = SanitizedUnicode()	1✔
151	subdivisions = fields.List(fields.Dict())	1✔
152	harvested = fields.Boolean()	1✔
153	contentNote = fields.List(SanitizedUnicode())	1✔
154	customField1 = fields.List(fields.String(validate=validate.Length(min=1)))	1✔
155	customField2 = fields.List(fields.String(validate=validate.Length(min=1)))	1✔
156	customField3 = fields.List(fields.String(validate=validate.Length(min=1)))	1✔
157	masked = SanitizedUnicode()	1✔
158	_bucket = SanitizedUnicode()	1✔
159	_files = Nested(ThumbnailSchemaV1, many=True)	1✔
160	_oai = fields.Dict()	1✔
161	# When loading, if $schema is not provided, it's retrieved by
162	# Record.schema property.
163	schema = GenFunction(	1✔
164	load_only=True,
165	attribute="$schema",
166	data_key="$schema",
167	deserialize=schema_from_document,
168	)
169	permissions = fields.Dict(dump_only=True)	1✔
170	permalink = SanitizedUnicode(dump_only=True)	1✔
171
172	@pre_dump	1✔
173	def populate_files_properties(self, item, **kwargs):	1✔
174	"""Add some customs properties to file before dumping it.
175
176	:param item: Item object to process
177	:returns: Modified item
178	"""
179	if not item.get("_files"):	1✔
180	return item	1✔
181
182	# Check if organisation record forces to point file to an external url
183	item["external_url"] = has_external_urls_for_files(item)	1✔
184
185	# Add restriction, link and thumbnail to files
186	populate_files_properties(item)	1✔
187
188	# Sort files to have the main file in first position
189	item["_files"] = sorted(item["_files"], key=lambda file: file.get("order", 100))	1✔
190
191	return item	1✔
192
193	@pre_dump	1✔
194	def add_permissions(self, item, **kwargs):	1✔
195	"""Add permissions to record.
196
197	:param item: Dict representing the record.
198	:returns: Modified dict.
199	"""
200	# For public views, no check for permissions
201	if request.args.get("view"):	1✔
202	return item	1✔
203
204	item["permissions"] = {	1✔
205	"read": DocumentPermission.read(current_user_record, item),
206	"update": DocumentPermission.update(current_user_record, item),
207	"delete": DocumentPermission.delete(current_user_record, item),
208	}
209
210	return item	1✔
211
212	@pre_dump	1✔
213	def add_permalink(self, item, **kwargs):	1✔
214	"""Add permanent link to document."""
215	item["permalink"] = DocumentRecord.get_permanent_link(host=request.host_url, pid=item["pid"])	1✔
216	return item	1✔
217
218	@pre_dump	1✔
219	def add_ark_uri(self, item, **kwargs):	1✔
220	"""Add permanent link to document."""
221	resolver_url = current_app.config.get("SONAR_APP_ARK_RESOLVER")	1✔
222	for itm in item.get("identifiedBy", []):	1✔
223	if itm.get("type") == "ark":	1✔
224	itm["uri"] = f"{resolver_url}/{itm['value']}"	1✔
225	break	1✔
226	return item	1✔
227
228	@pre_dump	1✔
229	def add_formatted_texts(self, item, **kwargs):	1✔
230	"""Add formatted texts for objects which are processing in backend.
231
232	:param item: Dict of record data.
233	:returns: Modified data.
234	"""
235	# Provision activity processing
236	for index, provision_activity in enumerate(item.get("provisionActivity", [])):	1✔
237	item["provisionActivity"][index]["text"] = create_publication_statement(provision_activity)	1✔
238
239	# Part of proccessing
240	for index, part_of in enumerate(item.get("partOf", [])):	1✔
241	item["partOf"][index]["text"] = part_of_format(part_of)	1✔
242
243	# Contribution
244	for index, contribution in enumerate(item.get("contribution", [])):	1✔
245	item["contribution"][index]["text"] = contribution_text(contribution)	1✔
246
247	if item.get("dissertation"):	1✔
248	item["dissertation"]["text"] = dissertation(item)	1✔
249
250	return item	1✔
251
252	@pre_load	1✔
253	def guess_organisation(self, data, **kwargs):	1✔
254	"""Guess organisation from current logged user.
255
256	:param data: Dict of record data.
257	:returns: Modified dict of record data.
258	"""
259	# Organisation already attached to document, we do nothing.
260	if data.get("organisation"):	1✔
261	return data	1✔
262
263	# Store current user organisation in new document.
264	if current_user_record.get("organisation"):	1✔
265	data["organisation"] = [current_user_record["organisation"]]	1✔
266
267	return data	1✔
268
269	@pre_load	1✔
270	def remove_fields(self, data, **kwargs):	1✔
271	"""Removes computed fields.
272
273	:param data: Dict of record data.
274	:returns: Modified data.
275	"""
276	for provision_activity in data.get("provisionActivity", []):	1✔
277	provision_activity.pop("text", None)	1✔
278
279	for part_of in data.get("partOf", []):	1✔
280	part_of.pop("text", None)	1✔
281
282	for contribution in data.get("contribution", []):	1✔
283	contribution.pop("text", None)	1✔
284
285	data.get("dissertation", {}).pop("text", None)	1✔
286
287	data.pop("permalink", None)	1✔
288	data.pop("permissions", None)	1✔
289	data.pop("external_url", None)	1✔
290	for itm in data.get("identifiedBy", []):	1✔
291	if itm.get("type") == "ark":	1✔
292	itm.pop("uri", None)	1✔
293	break	1✔
294
295	return data	1✔
296
297
298	class DocumentMetadataSchemaV1(DocumentListMetadataSchemaV1):	1✔
299	"""Schema for the document metadata."""
300
301	_files = Nested(FileSchemaV1, many=True)	1✔
302
303
304	class DocumentSchemaV1(StrictKeysMixin):	1✔
305	"""Document schema."""
306
307	id = PersistentIdentifier()	1✔
308	metadata = fields.Nested(DocumentMetadataSchemaV1)	1✔
309	links = fields.Dict(dump_only=True)	1✔
310	explanation = fields.Raw(dump_only=True)	1✔
311
312
313	class DocumentListSchemaV1(StrictKeysMixin):	1✔
314	"""Document schema."""
315
316	id = PersistentIdentifier()	1✔
317	metadata = fields.Nested(DocumentListMetadataSchemaV1)	1✔
318	links = fields.Dict(dump_only=True)	1✔
319	explanation = fields.Raw(dump_only=True)	1✔

rero / sonar / 17425918180

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