16621609190

Committed 30 Jul 2025 11:43AM UTC coverage: 84.491% (+0.008%) from 84.483%

Build # 16621609190

Build Type

push

github

Committed by

rerowep

Commit Message

chore: update dependencies

Co-Authored-by: Peter Weber <peter.weber@rero.ch>

Run Details

4560 of 5397 relevant lines covered (84.49%)

0.84 hits per line

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

91.43

/rero_mef/places/mef/api.py

# RERO MEF
# Copyright (C) 2024 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/>.

"""API for manipulating MEF records."""

from copy import deepcopy

from flask import current_app
from invenio_search.api import RecordsSearch

from rero_mef.api import EntityIndexer
from rero_mef.api_mef import EntityMefRecord

from .fetchers import mef_id_fetcher
from .minters import mef_id_minter
from .models import PlaceMefMetadata
from .providers import PlaceMefProvider


class PlaceMefSearch(RecordsSearch):
    """RecordsSearch."""

    class Meta:
        """Search only on index."""

        index = "places_mef"
        doc_types = None
        fields = ("*",)
        facets = {}

        default_filter = None


class PlaceMefRecord(EntityMefRecord):
    """Mef place class."""

    minter = mef_id_minter
    fetcher = mef_id_fetcher
    provider = PlaceMefProvider
    name = "mef"
    model_cls = PlaceMefMetadata
    search = PlaceMefSearch
    mef_type = "PLACES"
    entities = ["idref", "gnd"]

    @classmethod
    def create(
        cls,
        data,
        id_=None,
        delete_pid=False,
        dbcommit=False,
        reindex=False,
        md5=True,
        **kwargs,
    ):
        """Create a new place record."""
        data["type"] = "bf:Place"
        return super().create(
            data=data,
            id_=id_,
            delete_pid=delete_pid,
            dbcommit=dbcommit,
            reindex=reindex,
            md5=False,
            **kwargs,
        )

    def replace_refs(self):
        """Replace $ref with real data."""
        data = deepcopy(self)
        data = super().replace_refs()
        data["sources"] = [place for place in self.entities if data.get(place)]
        return data

    def add_information(self, resolve=False, sources=False):
        """Add information to record.

        Sources will be also added if resolve is True.
        :param resolve: resolve $refs
        :param sources: Add sources information to record
        :returns: record
        """
        replace_refs_data = PlaceMefRecord(deepcopy(self).replace_refs())
        data = replace_refs_data if resolve else deepcopy(self)
        my_sources = []
        for place in self.entities:
            if place_data := data.get(place):
                # we got a error status in data
                if place_data.get("status"):
                    data.pop(place)
                    current_app.logger.error(
                        f"MEF replace refs {data.get('pid')} {place}"
                        f" status: {place_data.get('status')}"
                        f" {place_data.get('message')}"
                    )
                else:
                    my_sources.append(place)
                for place in self.entities:
                    if metadata := replace_refs_data.get(place, {}).get("metadata"):
                        data[place] = metadata
        if my_sources and (resolve or sources):
            data["sources"] = my_sources
        return data

    @classmethod
    def get_latest(cls, pid_type, pid):
        """Get latest Mef record for pid_type and pid.

        :param pid_type: pid type to use.
        :param pid: pid to use..
        :returns: latest record.
        """
        search = PlaceMefSearch().filter({"term": {f"{pid_type}.pid": pid}})
        if search.count() > 0:
            data = next(search.scan()).to_dict()
            new_pid = None
            if relation_pid := data.get(pid_type, {}).get("relation_pid"):
                if relation_pid["type"] == "redirect_to":
                    new_pid = relation_pid["value"]
            elif pid_type == "idref":
                # Find new pid from redirect_pid redirect_from
                search = PlaceMefSearch().filter("term", idref__relation_pid__value=pid)
                if search.count() > 0:
                    new_data = next(search.scan()).to_dict()
                    new_pid = new_data.get("idref", {}).get("pid")
            return cls.get_latest(pid_type=pid_type, pid=new_pid) if new_pid else data
        return {}


class PlaceMefIndexer(EntityIndexer):
    """Place MEF indexer."""

    record_cls = PlaceMefRecord

    def bulk_index(self, record_id_iterator):
        """Bulk index records.

        :param record_id_iterator: Iterator yielding record UUIDs.
        """
        super().bulk_index(
            record_id_iterator, index=PlaceMefSearch.Meta.index, doc_type="plmef"
        )

1	# RERO MEF
2	# Copyright (C) 2024 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	"""API for manipulating MEF records."""
17
18	from copy import deepcopy	1✔
19
20	from flask import current_app	1✔
21	from invenio_search.api import RecordsSearch	1✔
22
23	from rero_mef.api import EntityIndexer	1✔
24	from rero_mef.api_mef import EntityMefRecord	1✔
25
26	from .fetchers import mef_id_fetcher	1✔
27	from .minters import mef_id_minter	1✔
28	from .models import PlaceMefMetadata	1✔
29	from .providers import PlaceMefProvider	1✔
30
31
32	class PlaceMefSearch(RecordsSearch):	1✔
33	"""RecordsSearch."""
34
35	class Meta:	1✔
36	"""Search only on index."""
37
38	index = "places_mef"	1✔
39	doc_types = None	1✔
40	fields = ("*",)	1✔
41	facets = {}	1✔
42
43	default_filter = None	1✔
44
45
46	class PlaceMefRecord(EntityMefRecord):	1✔
47	"""Mef place class."""
48
49	minter = mef_id_minter	1✔
50	fetcher = mef_id_fetcher	1✔
51	provider = PlaceMefProvider	1✔
52	name = "mef"	1✔
53	model_cls = PlaceMefMetadata	1✔
54	search = PlaceMefSearch	1✔
55	mef_type = "PLACES"	1✔
56	entities = ["idref", "gnd"]	1✔
57
58	@classmethod	1✔
59	def create(	1✔
60	cls,
61	data,
62	id_=None,
63	delete_pid=False,
64	dbcommit=False,
65	reindex=False,
66	md5=True,
67	**kwargs,
68	):
69	"""Create a new place record."""
70	data["type"] = "bf:Place"	1✔
71	return super().create(	1✔
72	data=data,
73	id_=id_,
74	delete_pid=delete_pid,
75	dbcommit=dbcommit,
76	reindex=reindex,
77	md5=False,
78	**kwargs,
79	)
80
81	def replace_refs(self):	1✔
82	"""Replace $ref with real data."""
83	data = deepcopy(self)	1✔
84	data = super().replace_refs()	1✔
85	data["sources"] = [place for place in self.entities if data.get(place)]	1✔
86	return data	1✔
87
88	def add_information(self, resolve=False, sources=False):	1✔
89	"""Add information to record.
90
91	Sources will be also added if resolve is True.
92	:param resolve: resolve $refs
93	:param sources: Add sources information to record
94	:returns: record
95	"""
96	replace_refs_data = PlaceMefRecord(deepcopy(self).replace_refs())	1✔
97	data = replace_refs_data if resolve else deepcopy(self)	1✔
98	my_sources = []	1✔
99	for place in self.entities:	1✔
100	if place_data := data.get(place):	1✔
101	# we got a error status in data
102	if place_data.get("status"):	1✔
103	data.pop(place)	×
104	current_app.logger.error(	×
105	f"MEF replace refs {data.get('pid')} {place}"
106	f" status: {place_data.get('status')}"
107	f" {place_data.get('message')}"
108	)
109	else:
110	my_sources.append(place)	1✔
111	for place in self.entities:	1✔
112	if metadata := replace_refs_data.get(place, {}).get("metadata"):	1✔
113	data[place] = metadata	×
114	if my_sources and (resolve or sources):	1✔
115	data["sources"] = my_sources	1✔
116	return data	1✔
117
118	@classmethod	1✔
119	def get_latest(cls, pid_type, pid):	1✔
120	"""Get latest Mef record for pid_type and pid.
121
122	:param pid_type: pid type to use.
123	:param pid: pid to use..
124	:returns: latest record.
125	"""
126	search = PlaceMefSearch().filter({"term": {f"{pid_type}.pid": pid}})	1✔
127	if search.count() > 0:	1✔
128	data = next(search.scan()).to_dict()	1✔
129	new_pid = None	1✔
130	if relation_pid := data.get(pid_type, {}).get("relation_pid"):	1✔
131	if relation_pid["type"] == "redirect_to":	1✔
132	new_pid = relation_pid["value"]	×
133	elif pid_type == "idref":	1✔
134	# Find new pid from redirect_pid redirect_from
135	search = PlaceMefSearch().filter("term", idref__relation_pid__value=pid)	1✔
136	if search.count() > 0:	1✔
137	new_data = next(search.scan()).to_dict()	1✔
138	new_pid = new_data.get("idref", {}).get("pid")	1✔
139	return cls.get_latest(pid_type=pid_type, pid=new_pid) if new_pid else data	1✔
140	return {}	×
141
142
143	class PlaceMefIndexer(EntityIndexer):	1✔
144	"""Place MEF indexer."""
145
146	record_cls = PlaceMefRecord	1✔
147
148	def bulk_index(self, record_id_iterator):	1✔
149	"""Bulk index records.
150
151	:param record_id_iterator: Iterator yielding record UUIDs.
152	"""
153	super().bulk_index(	×
154	record_id_iterator, index=PlaceMefSearch.Meta.index, doc_type="plmef"
155	)

rero / rero-mef / 16621609190

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