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

95.24

/rero_mef/tasks.py

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

"""Celery tasks to index records."""

import click
from celery import shared_task
from flask import current_app

from .api import Action, EntityIndexer
from .utils import get_entity_class


@shared_task(ignore_result=True)
def process_bulk_queue(version_type=None, search_bulk_kwargs=None, stats_only=True):
    """Process bulk indexing queue.

    :param str version_type: Elasticsearch version type.
    :param dict search_bulk_kwargs: Passed to
        :func:`elasticsearch:elasticsearch.helpers.bulk`.
    :param boolean stats_only: if `True` only report number of
            successful/failed operations instead of just number of
            successful and a list of error responses.
    Note: You can start multiple versions of this task.
    """
    return EntityIndexer(version_type=version_type).process_bulk_queue(
        search_bulk_kwargs=search_bulk_kwargs, stats_only=stats_only
    )


@shared_task
def create_or_update(
    idx, record, entity, dbcommit=True, reindex=True, test_md5=False, verbose=False
):
    """Create or update record task.

    :param index: index of record
    :param record: record data to use
    :param agent: agent to use
    :param dbcommit: db commit or not
    :param reindex: reindex or not
    :param test_md5: test md5 or not
    :param verbose: verbose or not
    :returns: id type, pid or id, agent action, MEF action
    """
    entity_class = get_entity_class(entity)
    record, agent_action = entity_class.create_or_update(
        data=record, dbcommit=dbcommit, reindex=reindex, test_md5=test_md5
    )
    entities = current_app.config.get("RERO_ENTITIES", [])
    mef_record = None
    if entity in entities and agent_action in (
        Action.CREATE,
        Action.UPDATE,
        Action.REPLACE,
    ):
        mef_record, mef_actions = record.create_or_update_mef(
            dbcommit=dbcommit, reindex=reindex
        )
    rec_id = record.get("pid")
    id_type = "pid:"
    if not rec_id:
        id_type = "uuid:"
        rec_id = record.id
    if verbose:
        msg = f"{idx:<10} {entity:<6} {id_type:<5} {rec_id:<25} {agent_action.name}"
        if mef_record:
            for mef_pid, mef_action in mef_actions.items():
                msg = f"{msg} | mef: {mef_pid} {mef_action.name}"
        click.echo(msg)
    return id_type, str(rec_id), agent_action


@shared_task
def delete(idx, pid, entity, dbcommit=True, delindex=True, verbose=False):
    """Delete record task.

    :param index: index of record
    :param pid: pid to delete
    :param agent: agent to use
    :param dbcommit: db commit or not
    :param delindex: delete index or not
    :param verbose: verbose or not
    :returns: action
    """
    entity_class = get_entity_class(entity)
    if entity_record := entity_class.get_record_by_pid(pid):
        entity_record.delete(dbcommit=dbcommit, delindex=delindex)
        if verbose:
            msg = f"{idx:<10} {entity:<6} pid: {pid:<25} DELETED"
            click.echo(msg)
        return "DELETED: {entity} {pid}"
    msg = f"{idx:<10} {entity:<6} pid: {pid:<25} NOT FOUND"
    if verbose:
        click.secho(msg, fg="yellow")
    current_app.logger.warning(msg)
    return f"DELETE NOT FOUND: {entity} {pid}"

1	# RERO MEF
2	# Copyright (C) 2020 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	"""Celery tasks to index records."""
17
18	import click	1✔
19	from celery import shared_task	1✔
20	from flask import current_app	1✔
21
22	from .api import Action, EntityIndexer	1✔
23	from .utils import get_entity_class	1✔
24
25
26	@shared_task(ignore_result=True)	1✔
27	def process_bulk_queue(version_type=None, search_bulk_kwargs=None, stats_only=True):	1✔
28	"""Process bulk indexing queue.
29
30	:param str version_type: Elasticsearch version type.
31	:param dict search_bulk_kwargs: Passed to
32	:func:`elasticsearch:elasticsearch.helpers.bulk`.
33	:param boolean stats_only: if `True` only report number of
34	successful/failed operations instead of just number of
35	successful and a list of error responses.
36	Note: You can start multiple versions of this task.
37	"""
38	return EntityIndexer(version_type=version_type).process_bulk_queue(	1✔
39	search_bulk_kwargs=search_bulk_kwargs, stats_only=stats_only
40	)
41
42
43	@shared_task	1✔
44	def create_or_update(	1✔
45	idx, record, entity, dbcommit=True, reindex=True, test_md5=False, verbose=False
46	):
47	"""Create or update record task.
48
49	:param index: index of record
50	:param record: record data to use
51	:param agent: agent to use
52	:param dbcommit: db commit or not
53	:param reindex: reindex or not
54	:param test_md5: test md5 or not
55	:param verbose: verbose or not
56	:returns: id type, pid or id, agent action, MEF action
57	"""
58	entity_class = get_entity_class(entity)	1✔
59	record, agent_action = entity_class.create_or_update(	1✔
60	data=record, dbcommit=dbcommit, reindex=reindex, test_md5=test_md5
61	)
62	entities = current_app.config.get("RERO_ENTITIES", [])	1✔
63	mef_record = None	1✔
64	if entity in entities and agent_action in (	1✔
65	Action.CREATE,
66	Action.UPDATE,
67	Action.REPLACE,
68	):
69	mef_record, mef_actions = record.create_or_update_mef(	1✔
70	dbcommit=dbcommit, reindex=reindex
71	)
72	rec_id = record.get("pid")	1✔
73	id_type = "pid:"	1✔
74	if not rec_id:	1✔
75	id_type = "uuid:"	×
76	rec_id = record.id	×
77	if verbose:	1✔
78	msg = f"{idx:<10} {entity:<6} {id_type:<5} {rec_id:<25} {agent_action.name}"	1✔
79	if mef_record:	1✔
80	for mef_pid, mef_action in mef_actions.items():	1✔
81	msg = f"{msg} \| mef: {mef_pid} {mef_action.name}"	1✔
82	click.echo(msg)	1✔
83	return id_type, str(rec_id), agent_action	1✔
84
85
86	@shared_task	1✔
87	def delete(idx, pid, entity, dbcommit=True, delindex=True, verbose=False):	1✔
88	"""Delete record task.
89
90	:param index: index of record
91	:param pid: pid to delete
92	:param agent: agent to use
93	:param dbcommit: db commit or not
94	:param delindex: delete index or not
95	:param verbose: verbose or not
96	:returns: action
97	"""
98	entity_class = get_entity_class(entity)	1✔
99	if entity_record := entity_class.get_record_by_pid(pid):	1✔
100	entity_record.delete(dbcommit=dbcommit, delindex=delindex)	1✔
101	if verbose:	1✔
102	msg = f"{idx:<10} {entity:<6} pid: {pid:<25} DELETED"	1✔
103	click.echo(msg)	1✔
104	return "DELETED: {entity} {pid}"	1✔
105	msg = f"{idx:<10} {entity:<6} pid: {pid:<25} NOT FOUND"	1✔
106	if verbose:	1✔
107	click.secho(msg, fg="yellow")	1✔
108	current_app.logger.warning(msg)	1✔
109	return f"DELETE NOT FOUND: {entity} {pid}"	1✔

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