28589338958

Committed 02 Jul 2026 10:50AM UTC coverage: 86.67% (+0.02%) from 86.649%

Build # 28589338958

Build Type

Pull #923

github

Committed by

ItIsJordan

Commit Message

Merge branch 'main' into mh-refactorAnalysesCI

Pull Request Pull Request #923: Refactor analyses CI

Coverage Stats

5039 of 5814 relevant lines covered (86.67%)

0.87 hits per line

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

92.16

hepdata/modules/records/utils/analyses.py

# -*- coding: utf-8 -*-
#
# This file is part of HEPData.
# Copyright (C) 2021 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.
#
# In applying this license, CERN does not
# waive the privileges and immunities granted to it by virtue of its status
# as an Intergovernmental Organization or submit itself to any jurisdiction.

import logging
import os

from celery import shared_task
from flask import current_app
from invenio_db import db
from hepdata.resilient_requests import resilient_requests
from sqlalchemy import select
import json
import jsonschema

from hepdata.ext.opensearch.api import index_record_ids
from hepdata.modules.submission.api import get_latest_hepsubmission, is_resource_added_to_submission
from hepdata.modules.submission.models import DataResource, HEPSubmission, data_reference_link
from hepdata.utils.users import get_user_from_id
from hepdata.modules.records.subscribers.rest import subscribe
from hepdata.modules.records.subscribers.api import is_current_user_subscribed_to_record
from hepdata.modules.records.utils.common import get_license

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

def test_analyses_schema(json_file, schema_version="1.0.0"):
    schema_path = os.path.join("hepdata", "templates", "analyses_schema", schema_version, "analyses_schema.json")
    with open(schema_path) as f:
        schema = json.load(f)
    jsonschema.validate(instance=json_file, schema=schema)

@shared_task
def update_analyses_single_tool(analysis_endpoint):
    """
    Update tool (Rivet, MadAnalysis 5, etc.) analyses and remove outdated resources.
    Allow bulk subscription to record update notifications if "subscribe_user_id" in endpoint.
    Add optional "description" and "license" fields if present in endpoint.

    Raise exceptions if errors encountered

    :param analysis_endpoint: any one from config.ANALYSES_ENDPOINTS ("rivet", "MadAnalysis", etc.)
    """

    endpoints = current_app.config["ANALYSES_ENDPOINTS"]
    if "endpoint_url" in endpoints[analysis_endpoint]:

        log.info("Updating analyses from {0}...".format(analysis_endpoint))

        response = resilient_requests('get', endpoints[analysis_endpoint]["endpoint_url"])

        if response.ok:

            analysis_resources = db.session.execute(
                select(DataResource).filter_by(file_type=analysis_endpoint)
            ).scalars().all()

            r_json = response.json()

            schema_version = "0.1.0"  # default to 0.1.0 for backward compatibility when schema_version field is missing
            if "schema_version" in r_json:
                schema_version = r_json["schema_version"]

            # Validate analyses JSON file against the schema.
            test_analyses_schema(r_json, schema_version=schema_version)

            if schema_version == "0.1.0":
                analyses = r_json

                # Check for missing analyses.
                for record in analyses:
                    submission = get_latest_hepsubmission(inspire_id=record, overall_status='finished')

                    if submission:
                        num_new_resources = 0

                        for analysis in analyses[record]:
                            _resource_url = endpoints[analysis_endpoint]["url_template"].format(analysis)

                            if not is_resource_added_to_submission(submission.publication_recid, submission.version,
                                                                _resource_url):

                                log.info('Adding {} analysis to ins{} with URL {}'.format(
                                    analysis_endpoint, record, _resource_url)
                                )
                                new_resource = DataResource(
                                    file_location=_resource_url,
                                    file_type=analysis_endpoint)

                                if "description" in endpoints[analysis_endpoint]:
                                    new_resource.file_description = str(endpoints[analysis_endpoint]["description"])

                                if "license" in endpoints[analysis_endpoint]:
                                    resource_license = get_license(endpoints[analysis_endpoint]["license"])
                                    new_resource.file_license = resource_license.id

                                submission.resources.append(new_resource)
                                num_new_resources += 1

                            else:

                                # Remove resources from 'analysis_resources' list.
                                resources = list(filter(lambda a: a.file_location == _resource_url, analysis_resources))
                                for resource in resources:
                                    analysis_resources.remove(resource)

                        if num_new_resources:

                            try:
                                db.session.add(submission)
                                db.session.commit()
                                latest_submission = get_latest_hepsubmission(inspire_id=record)
                                if submission.version == latest_submission.version:
                                    index_record_ids([submission.publication_recid])
                            except Exception as e:
                                db.session.rollback()
                                log.error(e)

                    else:
                        log.debug("An analysis is available in {0} but with no equivalent in HEPData (ins{1}).".format(
                            analysis_endpoint, record))

            else:  # schema_version >= "1.0.0"
                # Check for missing analyses.
                for ana in r_json["analyses"]:
                    inspire_id = str(ana["inspire_id"])  # inspire_id is stored as a string in the database
                    submission = get_latest_hepsubmission(inspire_id=inspire_id, overall_status='finished')

                    if submission:
                        num_new_resources = 0

                        for implementation in ana["implementations"]:
                            _resource_url = r_json["url_templates"]["main_url"].format(**implementation)

                            if not is_resource_added_to_submission(submission.publication_recid, submission.version,
                                                                _resource_url):

                                log.info('Adding {} analysis to ins{} with URL {}'.format(
                                    analysis_endpoint, inspire_id, _resource_url)
                                )
                                new_resource = DataResource(
                                    file_location=_resource_url,
                                    file_type=analysis_endpoint,
                                    file_description=r_json["implementations_description"]
                                )

                                if "implementations_license" in r_json:
                                    resource_license = get_license(r_json["implementations_license"])
                                    new_resource.file_license = resource_license.id

                                submission.resources.append(new_resource)
                                num_new_resources += 1

                            else:

                                # Remove resources from 'analysis_resources' list.
                                resources = list(filter(lambda a: a.file_location == _resource_url, analysis_resources))
                                for resource in resources:
                                    analysis_resources.remove(resource)

                        if num_new_resources:

                            try:
                                db.session.add(submission)
                                db.session.commit()
                                latest_submission = get_latest_hepsubmission(inspire_id=inspire_id)
                                if submission.version == latest_submission.version:
                                    index_record_ids([submission.publication_recid])
                            except Exception as e:
                                db.session.rollback()
                                log.error(e)

                    else:
                        log.debug("An analysis is available in {0} but with no equivalent in HEPData (ins{1}).".format(
                            analysis_endpoint, inspire_id))

            if analysis_resources:
                # Extra resources that were not found in the analyses JSON file.
                # Need to delete extra resources then reindex affected submissions.
                # Only take action if latest version is finished (most important case).
                try:
                    recids_to_reindex = []
                    for extra_analysis_resource in analysis_resources:
                        if not extra_analysis_resource.file_location.lower().startswith('http'):
                            continue  # don't delete local files from database

                        query = select(data_reference_link.columns.submission_id).where(data_reference_link.columns.dataresource_id == extra_analysis_resource.id)
                        results = db.session.execute(query)
                        for result in results:
                            submission_id = result[0]
                        submission = HEPSubmission.query.filter_by(id=submission_id).first()
                        latest_submission = get_latest_hepsubmission(
                            publication_recid=submission.publication_recid, overall_status='finished'
                        )
                        if submission and latest_submission and submission.version == latest_submission.version:
                            log.info('Removing {} analysis with URL {} from submission {} version {}'
                                        .format(analysis_endpoint, extra_analysis_resource.file_location,
                                                submission.publication_recid, submission.version))
                            db.session.delete(extra_analysis_resource)
                            recids_to_reindex.append(submission.publication_recid)
                    db.session.commit()
                    if recids_to_reindex:
                        unique_recids = list(set(recids_to_reindex))  # remove duplicates before indexing
                        # For large numbers of records, batch the indexing to reduce memory usage
                        batch_size = 100
                        for i in range(0, len(unique_recids), batch_size):
                            batch_recids = unique_recids[i:i+batch_size]
                            index_record_ids(batch_recids)
                except Exception as e:
                    db.session.rollback()
                    log.error(e)

            # Allow bulk subscription to record update notifications.
            if "subscribe_user_id" in endpoints[analysis_endpoint]:
                user = get_user_from_id(endpoints[analysis_endpoint]["subscribe_user_id"])
                if user:
                    # Check for missing analyses.
                    if schema_version == "0.1.0":
                        for record in analyses:
                            submission = get_latest_hepsubmission(inspire_id=record, overall_status='finished')
                            if submission and not is_current_user_subscribed_to_record(submission.publication_recid, user):
                                subscribe(submission.publication_recid, user)

                    else:  # schema_version >= "1.0.0"
                        for ana in r_json["analyses"]:
                            submission = get_latest_hepsubmission(inspire_id=str(ana["inspire_id"]), overall_status='finished')
                            if submission and not is_current_user_subscribed_to_record(submission.publication_recid, user):
                                subscribe(submission.publication_recid, user)

        else:  # if not response.ok
            raise LookupError(f"Error accessing {endpoints[analysis_endpoint]['endpoint_url']}, status {response.status_code}")

    else:  # if "endpoint_url" not in endpoints[analysis_endpoint]
        raise KeyError(f"No endpoint_url configured for {analysis_endpoint}")

@shared_task
def update_analyses(endpoint=None):
    """
    Update tools (Rivet, MadAnalysis 5, etc.) analyses and remove outdated resources.
    Allow bulk subscription to record update notifications if "subscribe_user_id" in endpoint.
    Add optional "description" and "license" fields if present in endpoint.

    Logs exceptions when errors encountered but continues with next tool.

    :param endpoint: any one from config.ANALYSES_ENDPOINTS ("rivet", "MadAnalysis", etc.) or None (default) for all

    :return: True if all analyses endpoints were updated successfully, False if any error was encountered
    """

    for analysis_endpoint in current_app.config["ANALYSES_ENDPOINTS"]:

        if endpoint and endpoint != analysis_endpoint:
            continue

        try:
            update_analyses_single_tool(analysis_endpoint)
        except jsonschema.exceptions.ValidationError as e:
            log.error("Validation error for analyses schema in {}: {}".format(analysis_endpoint, e))
            return False
        except KeyError:
            # KeyError is on HEPData's side and should be raised
            raise
        except Exception as e:
            # need to support LookupError and json.JSONDecodeError
            # but also ConnectionError, urllib3.exceptions.HTTPError, requests.RequestException, ...
            # => maybe best to be pragmatic and catch all exceptions
            log.error(str(e))
            return False

    return True

1	# -- coding: utf-8 --
2	#
3	# This file is part of HEPData.
4	# Copyright (C) 2021 CERN.
5	#
6	# HEPData is free software; you can redistribute it
7	# and/or modify it under the terms of the GNU General Public License as
8	# published by the Free Software Foundation; either version 2 of the
9	# License, or (at your option) any later version.
10	#
11	# HEPData is distributed in the hope that it will be
12	# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14	# General Public License for more details.
15	#
16	# You should have received a copy of the GNU General Public License
17	# along with HEPData; if not, write to the
18	# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
19	# MA 02111-1307, USA.
20	#
21	# In applying this license, CERN does not
22	# waive the privileges and immunities granted to it by virtue of its status
23	# as an Intergovernmental Organization or submit itself to any jurisdiction.
24
25	import logging	1✔
26	import os	1✔
27
28	from celery import shared_task	1✔
29	from flask import current_app	1✔
30	from invenio_db import db	1✔
31	from hepdata.resilient_requests import resilient_requests	1✔
32	from sqlalchemy import select	1✔
33	import json	1✔
34	import jsonschema	1✔
35
36	from hepdata.ext.opensearch.api import index_record_ids	1✔
37	from hepdata.modules.submission.api import get_latest_hepsubmission, is_resource_added_to_submission	1✔
38	from hepdata.modules.submission.models import DataResource, HEPSubmission, data_reference_link	1✔
39	from hepdata.utils.users import get_user_from_id	1✔
40	from hepdata.modules.records.subscribers.rest import subscribe	1✔
41	from hepdata.modules.records.subscribers.api import is_current_user_subscribed_to_record	1✔
42	from hepdata.modules.records.utils.common import get_license	1✔
43
44	logging.basicConfig()	1✔
45	log = logging.getLogger(__name__)	1✔
46
47	def test_analyses_schema(json_file, schema_version="1.0.0"):	1✔
48	schema_path = os.path.join("hepdata", "templates", "analyses_schema", schema_version, "analyses_schema.json")	1✔
49	with open(schema_path) as f:	1✔
50	schema = json.load(f)	1✔
51	jsonschema.validate(instance=json_file, schema=schema)	1✔
52
53	@shared_task	1✔
54	def update_analyses_single_tool(analysis_endpoint):	1✔
55	"""
56	Update tool (Rivet, MadAnalysis 5, etc.) analyses and remove outdated resources.
57	Allow bulk subscription to record update notifications if "subscribe_user_id" in endpoint.
58	Add optional "description" and "license" fields if present in endpoint.
59
60	Raise exceptions if errors encountered
61
62	:param analysis_endpoint: any one from config.ANALYSES_ENDPOINTS ("rivet", "MadAnalysis", etc.)
63	"""
64
65	endpoints = current_app.config["ANALYSES_ENDPOINTS"]	1✔
66	if "endpoint_url" in endpoints[analysis_endpoint]:	1✔
67
68	log.info("Updating analyses from {0}...".format(analysis_endpoint))	1✔
69
70	response = resilient_requests('get', endpoints[analysis_endpoint]["endpoint_url"])	1✔
71
72	if response.ok:	1✔
73
74	analysis_resources = db.session.execute(	1✔
75	select(DataResource).filter_by(file_type=analysis_endpoint)
76	).scalars().all()
77
78	r_json = response.json()	1✔
79
80	schema_version = "0.1.0" # default to 0.1.0 for backward compatibility when schema_version field is missing	1✔
81	if "schema_version" in r_json:	1✔
82	schema_version = r_json["schema_version"]	1✔
83
84	# Validate analyses JSON file against the schema.
85	test_analyses_schema(r_json, schema_version=schema_version)	1✔
86
87	if schema_version == "0.1.0":	1✔
88	analyses = r_json	1✔
89
90	# Check for missing analyses.
91	for record in analyses:	1✔
92	submission = get_latest_hepsubmission(inspire_id=record, overall_status='finished')	1✔
93
94	if submission:	1✔
95	num_new_resources = 0	1✔
96
97	for analysis in analyses[record]:	1✔
98	_resource_url = endpoints[analysis_endpoint]["url_template"].format(analysis)	1✔
99
100	if not is_resource_added_to_submission(submission.publication_recid, submission.version,	1✔
101	_resource_url):
102
103	log.info('Adding {} analysis to ins{} with URL {}'.format(	1✔
104	analysis_endpoint, record, _resource_url)
105	)
106	new_resource = DataResource(	1✔
107	file_location=_resource_url,
108	file_type=analysis_endpoint)
109
110	if "description" in endpoints[analysis_endpoint]:	1✔
111	new_resource.file_description = str(endpoints[analysis_endpoint]["description"])	1✔
112
113	if "license" in endpoints[analysis_endpoint]:	1✔
114	resource_license = get_license(endpoints[analysis_endpoint]["license"])	1✔
115	new_resource.file_license = resource_license.id	1✔
116
117	submission.resources.append(new_resource)	1✔
118	num_new_resources += 1	1✔
119
120	else:
121
122	# Remove resources from 'analysis_resources' list.
123	resources = list(filter(lambda a: a.file_location == _resource_url, analysis_resources))	×
124	for resource in resources:	×
125	analysis_resources.remove(resource)	×
126
127	if num_new_resources:	1✔
128
129	try:	1✔
130	db.session.add(submission)	1✔
131	db.session.commit()	1✔
132	latest_submission = get_latest_hepsubmission(inspire_id=record)	1✔
133	if submission.version == latest_submission.version:	1✔
134	index_record_ids([submission.publication_recid])	1✔
135	except Exception as e:	×
136	db.session.rollback()	×
137	log.error(e)	×
138
139	else:
140	log.debug("An analysis is available in {0} but with no equivalent in HEPData (ins{1}).".format(	1✔
141	analysis_endpoint, record))
142
143	else: # schema_version >= "1.0.0"
144	# Check for missing analyses.
145	for ana in r_json["analyses"]:	1✔
146	inspire_id = str(ana["inspire_id"]) # inspire_id is stored as a string in the database	1✔
147	submission = get_latest_hepsubmission(inspire_id=inspire_id, overall_status='finished')	1✔
148
149	if submission:	1✔
150	num_new_resources = 0	1✔
151
152	for implementation in ana["implementations"]:	1✔
153	_resource_url = r_json["url_templates"]["main_url"].format(**implementation)	1✔
154
155	if not is_resource_added_to_submission(submission.publication_recid, submission.version,	1✔
156	_resource_url):
157
158	log.info('Adding {} analysis to ins{} with URL {}'.format(	1✔
159	analysis_endpoint, inspire_id, _resource_url)
160	)
161	new_resource = DataResource(	1✔
162	file_location=_resource_url,
163	file_type=analysis_endpoint,
164	file_description=r_json["implementations_description"]
165	)
166
167	if "implementations_license" in r_json:	1✔
168	resource_license = get_license(r_json["implementations_license"])	1✔
169	new_resource.file_license = resource_license.id	1✔
170
171	submission.resources.append(new_resource)	1✔
172	num_new_resources += 1	1✔
173
174	else:
175
176	# Remove resources from 'analysis_resources' list.
177	resources = list(filter(lambda a: a.file_location == _resource_url, analysis_resources))	1✔
178	for resource in resources:	1✔
179	analysis_resources.remove(resource)	1✔
180
181	if num_new_resources:	1✔
182
183	try:	1✔
184	db.session.add(submission)	1✔
185	db.session.commit()	1✔
186	latest_submission = get_latest_hepsubmission(inspire_id=inspire_id)	1✔
187	if submission.version == latest_submission.version:	1✔
188	index_record_ids([submission.publication_recid])	1✔
189	except Exception as e:	×
190	db.session.rollback()	×
191	log.error(e)	×
192
193	else:
194	log.debug("An analysis is available in {0} but with no equivalent in HEPData (ins{1}).".format(	1✔
195	analysis_endpoint, inspire_id))
196
197	if analysis_resources:	1✔
198	# Extra resources that were not found in the analyses JSON file.
199	# Need to delete extra resources then reindex affected submissions.
200	# Only take action if latest version is finished (most important case).
201	try:	1✔
202	recids_to_reindex = []	1✔
203	for extra_analysis_resource in analysis_resources:	1✔
204	if not extra_analysis_resource.file_location.lower().startswith('http'):	1✔
205	continue # don't delete local files from database	1✔
206
207	query = select(data_reference_link.columns.submission_id).where(data_reference_link.columns.dataresource_id == extra_analysis_resource.id)	1✔
208	results = db.session.execute(query)	1✔
209	for result in results:	1✔
210	submission_id = result[0]	1✔
211	submission = HEPSubmission.query.filter_by(id=submission_id).first()	1✔
212	latest_submission = get_latest_hepsubmission(	1✔
213	publication_recid=submission.publication_recid, overall_status='finished'
214	)
215	if submission and latest_submission and submission.version == latest_submission.version:	1✔
216	log.info('Removing {} analysis with URL {} from submission {} version {}'	1✔
217	.format(analysis_endpoint, extra_analysis_resource.file_location,
218	submission.publication_recid, submission.version))
219	db.session.delete(extra_analysis_resource)	1✔
220	recids_to_reindex.append(submission.publication_recid)	1✔
221	db.session.commit()	1✔
222	if recids_to_reindex:	1✔
223	unique_recids = list(set(recids_to_reindex)) # remove duplicates before indexing	1✔
224	# For large numbers of records, batch the indexing to reduce memory usage
225	batch_size = 100	1✔
226	for i in range(0, len(unique_recids), batch_size):	1✔
227	batch_recids = unique_recids[i:i+batch_size]	1✔
228	index_record_ids(batch_recids)	1✔
229	except Exception as e:	×
230	db.session.rollback()	×
231	log.error(e)	×
232
233	# Allow bulk subscription to record update notifications.
234	if "subscribe_user_id" in endpoints[analysis_endpoint]:	1✔
235	user = get_user_from_id(endpoints[analysis_endpoint]["subscribe_user_id"])	1✔
236	if user:	1✔
237	# Check for missing analyses.
238	if schema_version == "0.1.0":	1✔
239	for record in analyses:	1✔
240	submission = get_latest_hepsubmission(inspire_id=record, overall_status='finished')	1✔
241	if submission and not is_current_user_subscribed_to_record(submission.publication_recid, user):	1✔
242	subscribe(submission.publication_recid, user)	1✔
243
244	else: # schema_version >= "1.0.0"
245	for ana in r_json["analyses"]:	1✔
246	submission = get_latest_hepsubmission(inspire_id=str(ana["inspire_id"]), overall_status='finished')	1✔
247	if submission and not is_current_user_subscribed_to_record(submission.publication_recid, user):	1✔
248	subscribe(submission.publication_recid, user)	1✔
249
250	else: # if not response.ok
251	raise LookupError(f"Error accessing {endpoints[analysis_endpoint]['endpoint_url']}, status {response.status_code}")	1✔
252
253	else: # if "endpoint_url" not in endpoints[analysis_endpoint]
254	raise KeyError(f"No endpoint_url configured for {analysis_endpoint}")	1✔
255
256	@shared_task	1✔
257	def update_analyses(endpoint=None):	1✔
258	"""
259	Update tools (Rivet, MadAnalysis 5, etc.) analyses and remove outdated resources.
260	Allow bulk subscription to record update notifications if "subscribe_user_id" in endpoint.
261	Add optional "description" and "license" fields if present in endpoint.
262
263	Logs exceptions when errors encountered but continues with next tool.
264
265	:param endpoint: any one from config.ANALYSES_ENDPOINTS ("rivet", "MadAnalysis", etc.) or None (default) for all
266
267	:return: True if all analyses endpoints were updated successfully, False if any error was encountered
268	"""
269
270	for analysis_endpoint in current_app.config["ANALYSES_ENDPOINTS"]:	1✔
271
272	if endpoint and endpoint != analysis_endpoint:	1✔
273	continue	1✔
274
275	try:	1✔
276	update_analyses_single_tool(analysis_endpoint)	1✔
277	except jsonschema.exceptions.ValidationError as e:	1✔
278	log.error("Validation error for analyses schema in {}: {}".format(analysis_endpoint, e))	1✔
279	return False	1✔
280	except KeyError:	1✔
281	# KeyError is on HEPData's side and should be raised
282	raise	1✔
283	except Exception as e:	1✔
284	# need to support LookupError and json.JSONDecodeError
285	# but also ConnectionError, urllib3.exceptions.HTTPError, requests.RequestException, ...
286	# => maybe best to be pragmatic and catch all exceptions
287	log.error(str(e))	1✔
288	return False	1✔
289
290	return True	1✔

HEPData / hepdata / 28589338958

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