21246856953

Committed 22 Jan 2026 11:32AM UTC coverage: 84.743% (+0.07%) from 84.673%

Build # 21246856953

Build Type

Pull #898

github

Committed by

web-flow

Commit Message

Merge 1a16115c6 into 7db713a49

Pull Request Pull Request #898: Observer role

Run Details

145 of 182 new or added lines in 8 files covered. (79.67%)

194 existing lines in 3 files now uncovered.

4860 of 5735 relevant lines covered (84.74%)

0.85 hits per line

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

93.1

/hepdata/modules/submission/api.py

#
# This file is part of HEPData.
# Copyright (C) 2016 CERN.
#
# HEPData is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; either version 2 of the
# License, or (at your option) any later version.
#
# HEPData 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
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with HEPData; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
#
import logging

from invenio_db import db
from hepdata.modules.submission.models import DataResource, SubmissionObserver
from hepdata.modules.permissions.models import SubmissionParticipant
from hepdata.modules.submission.models import HEPSubmission

"""Common utilities used across the code base."""

logging.basicConfig()
log = logging.getLogger(__name__)


def is_resource_added_to_submission(recid, version, resource_url):
    """
    Returns if a submission already has the given resource url
    :param recid:
    :param version:
    :param resource_url:
    :return:
    """
    return HEPSubmission.query.filter(HEPSubmission.publication_recid == recid,
                                      HEPSubmission.version == version,
                                      HEPSubmission.resources.any(
                                          DataResource.file_location == resource_url)).count() > 0


def get_latest_hepsubmission(*args, **kwargs):
    """
    Gets the latest HEPSubmission record matching the given kwargs

    :return: the HEPSubmission object or None
    """

    hepsubmissions = HEPSubmission.query.filter_by(**kwargs).all()

    last = None
    for hepsubmission in hepsubmissions:
        if last is None:
            last = hepsubmission
        else:
            if hepsubmission.version > last.version:
                last = hepsubmission

    return last


def get_submission_participants_for_record(publication_recid, roles=None, **kwargs):
    """Gets the participants for a given publication record id

    :param int publication_recid: publication_recid of a submission.
    :param ``**kwargs``: Additional filter parameters to pass to `filter_by`.
    :return: List of participants relating to that record
    :rtype: list[SubmissionParticipant]
    """
    query = SubmissionParticipant.query.filter_by(
            publication_recid=publication_recid,
            **kwargs
    )

    if roles:
        query = query.filter(SubmissionParticipant.role.in_(roles))

    return query.all()


def get_primary_submission_participants_for_record(publication_recid):
    submission_participants = get_submission_participants_for_record(publication_recid, status="primary")
    return submission_participants


def get_or_create_submission_observer(publication_recid, regenerate=False):
    """
    Gets or re/generates a SubmissionObserver key for a given recid.
    Where an observer does not exist for a recid (with existing sub),
    it is created and returned instead.

    :param publication_recid: The publication record id
    :param regenerate: Whether to regenerate/force generate the key
    :return: SubmissionObserver key, created, or None
    """
    submission_observer = SubmissionObserver.query.filter_by(publication_recid=publication_recid).first()
    created = False

    if submission_observer is None:
        submission = get_latest_hepsubmission(publication_recid=publication_recid)
        if submission:
            if submission.overall_status == "todo" or regenerate:
                submission_observer = SubmissionObserver(publication_recid=publication_recid)
                created = True
        else:
            # No submission, no observer, return None
            return None


    # If we are to regenerate, and SubmissionObserver was queried and not generated.
    # If just created, we don't need to generate anything.
    if not created and regenerate:
        submission_observer.generate_observer_key()

    # Only commit if we have created or regenerated
    if created or regenerate:
        db.session.add(submission_observer)
        db.session.commit()

    return submission_observer

def delete_submission_observer(recid):
    """
    Deletes a SubmissionObserver object from the database
    based on a given recid value.

    :param: recid: int - The recid to delete on
    """

    # Validate recid is an integer
    try:
        recid = int(recid)
    except (ValueError, TypeError) as e:
        log.error(f"Invalid recid provided for observer deletion: {recid}")
        raise ValueError(f"Supplied recid value ({recid}) for deletion is not an Integer.") from e

    try:
        submission_observer = SubmissionObserver.query.filter_by(publication_recid=recid).first()

        if submission_observer:
            db.session.delete(submission_observer)
            db.session.commit()
            log.info(f"Deleted observer for submission {recid}")
    except Exception as e:
        log.error(f"Error deleting observer for submission {recid}: {e}")
        db.session.rollback()
        raise

1	#
2	# This file is part of HEPData.
3	# Copyright (C) 2016 CERN.
4	#
5	# HEPData is free software; you can redistribute it and/or
6	# modify it under the terms of the GNU General Public License as
7	# published by the Free Software Foundation; either version 2 of the
8	# License, or (at your option) any later version.
9	#
10	# HEPData is distributed in the hope that it will be useful, but
11	# WITHOUT ANY WARRANTY; without even the implied warranty of
12	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13	# General Public License for more details.
14	#
15	# You should have received a copy of the GNU General Public License
16	# along with HEPData; if not, write to the Free Software Foundation, Inc.,
17	# 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
18	#
19	import logging	1✔
20
21	from invenio_db import db	1✔
22	from hepdata.modules.submission.models import DataResource, SubmissionObserver	1✔
23	from hepdata.modules.permissions.models import SubmissionParticipant	1✔
24	from hepdata.modules.submission.models import HEPSubmission	1✔
25
26	"""Common utilities used across the code base."""
27
28	logging.basicConfig()	1✔
29	log = logging.getLogger(__name__)	1✔
30
31
32	def is_resource_added_to_submission(recid, version, resource_url):	1✔
33	"""
34	Returns if a submission already has the given resource url
35	:param recid:
36	:param version:
37	:param resource_url:
38	:return:
39	"""
40	return HEPSubmission.query.filter(HEPSubmission.publication_recid == recid,	1✔
41	HEPSubmission.version == version,
42	HEPSubmission.resources.any(
43	DataResource.file_location == resource_url)).count() > 0
44
45
46	def get_latest_hepsubmission(args, *kwargs):	1✔
47	"""
48	Gets the latest HEPSubmission record matching the given kwargs
49
50	:return: the HEPSubmission object or None
51	"""
52
53	hepsubmissions = HEPSubmission.query.filter_by(**kwargs).all()	1✔
54
55	last = None	1✔
56	for hepsubmission in hepsubmissions:	1✔
57	if last is None:	1✔
58	last = hepsubmission	1✔
59	else:
60	if hepsubmission.version > last.version:	1✔
61	last = hepsubmission	1✔
62
63	return last	1✔
64
65
66	def get_submission_participants_for_record(publication_recid, roles=None, **kwargs):	1✔
67	"""Gets the participants for a given publication record id
68
69	:param int publication_recid: publication_recid of a submission.
70	:param ``**kwargs``: Additional filter parameters to pass to `filter_by`.
71	:return: List of participants relating to that record
72	:rtype: list[SubmissionParticipant]
73	"""
74	query = SubmissionParticipant.query.filter_by(	1✔
75	publication_recid=publication_recid,
76	**kwargs
77	)
78
79	if roles:	1✔
80	query = query.filter(SubmissionParticipant.role.in_(roles))	1✔
81
82	return query.all()	1✔
83
84
85	def get_primary_submission_participants_for_record(publication_recid):	1✔
86	submission_participants = get_submission_participants_for_record(publication_recid, status="primary")	1✔
87	return submission_participants	1✔
88
89
90	def get_or_create_submission_observer(publication_recid, regenerate=False):	1✔
91	"""
92	Gets or re/generates a SubmissionObserver key for a given recid.
93	Where an observer does not exist for a recid (with existing sub),
94	it is created and returned instead.
95
96	:param publication_recid: The publication record id
97	:param regenerate: Whether to regenerate/force generate the key
98	:return: SubmissionObserver key, created, or None
99	"""
100	submission_observer = SubmissionObserver.query.filter_by(publication_recid=publication_recid).first()	1✔
101	created = False	1✔
102
103	if submission_observer is None:	1✔
104	submission = get_latest_hepsubmission(publication_recid=publication_recid)	1✔
105	if submission:	1✔
106	if submission.overall_status == "todo" or regenerate:	1✔
107	submission_observer = SubmissionObserver(publication_recid=publication_recid)	1✔
108	created = True	1✔
109	else:
110	# No submission, no observer, return None
111	return None	1✔
112
113
114	# If we are to regenerate, and SubmissionObserver was queried and not generated.
115	# If just created, we don't need to generate anything.
116	if not created and regenerate:	1✔
117	submission_observer.generate_observer_key()	1✔
118
119	# Only commit if we have created or regenerated
120	if created or regenerate:	1✔
121	db.session.add(submission_observer)	1✔
122	db.session.commit()	1✔
123
124	return submission_observer	1✔
125
126	def delete_submission_observer(recid):	1✔
127	"""
128	Deletes a SubmissionObserver object from the database
129	based on a given recid value.
130
131	:param: recid: int - The recid to delete on
132	"""
133
134	# Validate recid is an integer
135	try:	1✔
136	recid = int(recid)	1✔
137	except (ValueError, TypeError) as e:	1✔
138	log.error(f"Invalid recid provided for observer deletion: {recid}")	1✔
139	raise ValueError(f"Supplied recid value ({recid}) for deletion is not an Integer.") from e	1✔
140
141	try:	1✔
142	submission_observer = SubmissionObserver.query.filter_by(publication_recid=recid).first()	1✔
143
144	if submission_observer:	1✔
145	db.session.delete(submission_observer)	1✔
146	db.session.commit()	1✔
147	log.info(f"Deleted observer for submission {recid}")	1✔
NEW 148	except Exception as e:	×
NEW 149	log.error(f"Error deleting observer for submission {recid}: {e}")	×
NEW 150	db.session.rollback()	×
NEW 151	raise	×

HEPData / hepdata / 21246856953

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