diff --git a/core/common/checksums.py b/core/common/checksums.py index 164b1b1cd..a9e488846 100644 --- a/core/common/checksums.py +++ b/core/common/checksums.py @@ -5,6 +5,8 @@ from ocldev.checksum import Checksum as ChecksumBase from pydash import get +from core.common.utils import chunks + # Return only aggregate counts for changed resource groups. SUMMARY_VERBOSITY = 0 @@ -403,12 +405,29 @@ def get_db_id_for(self, diff_key, identity): class ChecksumChangelog: + CONCEPTS_BATCH_SIZE = 1000 + DIFF_KEYS = ('new', 'removed', 'changed_retired', 'changed_major', 'changed_minor') + MAPPING_SUMMARY_FIELDS = ( + 'id', 'mnemonic', 'map_type', 'external_id', 'from_concept_code', 'to_concept_code', 'from_source_url', + 'to_source_url', 'from_concept__mnemonic', 'from_concept__versioned_object_id', 'from_concept__parent_id', + 'to_concept__mnemonic', + ) + def __init__(self, concepts_diff, mappings_diff, identity='mnemonic', verbosity=0): # pylint: disable=too-many-arguments self.concepts_diff = concepts_diff self.mappings_diff = mappings_diff self.identity = identity self.verbosity = verbosity self.result = {} + self.enrichment_skipped = False + + def get_diff_size(self): + return sum( + len(entry[self.identity]) + for diff in (self.concepts_diff, self.mappings_diff) + for entry in (diff.result.get(key) for key in self.DIFF_KEYS) + if isinstance(entry, dict) + ) def get_mapping_summary(self, mapping, mapping_id=None, v1_mapping=None): summary = { @@ -419,7 +438,7 @@ def get_mapping_summary(self, mapping, mapping_id=None, v1_mapping=None): 'to_source': mapping.to_source_url, 'map_type': mapping.map_type, } - if self.verbosity >= CHANGELOG_ENRICHMENT_VERBOSITY: + if self.verbosity >= CHANGELOG_ENRICHMENT_VERBOSITY and not self.enrichment_skipped: summary['external_id'] = getattr(mapping, 'external_id', None) if v1_mapping is not None: summary['prev_to_concept'] = ( @@ -446,66 +465,88 @@ def _collect_v1_v2_ids(self, diff, diff_obj): ids.add(v1_id) return ids - def _build_concepts_cache(self, diff_keys): - """Batch-fetch concepts with names/descriptions for enrich mode (avoids N+1 queries).""" + def _get_concepts_cache(self, key, mnemonics, enriched): + """Fetch one batch of concepts (and their v1 state) with names, and descriptions when enriched.""" from core.concepts.models import Concept - ids = self._collect_v1_v2_ids( - {k: self.concepts_diff.result.get(k, False) for k in diff_keys}, - self.concepts_diff, - ) - return { - c.id: c - for c in Concept.objects.filter(id__in=ids).prefetch_related('names', 'descriptions') - } + ids = self._collect_v1_v2_ids({key: {self.identity: mnemonics}}, self.concepts_diff) + parents = {} + cache = {} + prefetch = ('names', 'descriptions') if enriched else ('names',) + for concept in Concept.objects.filter(id__in=ids).prefetch_related(*prefetch): + if concept.parent_id not in parents: + parents[concept.parent_id] = concept.parent + concept.parent = parents[concept.parent_id] + cache[concept.id] = concept + return cache + + def _get_mapping_db_ids(self, key, mnemonics, enriched): + """The db ids of one batch of diff'd mappings, plus their v1 ids for prev_* when enriched.""" + ids = set() + for mnemonic in mnemonics: + db_id = self.mappings_diff.get_db_id_for(key, mnemonic) + if db_id: + ids.add(db_id) + if enriched and key in ('changed_major', 'changed_minor'): + v1_id = get_map_id(self.mappings_diff.resources1_map, mnemonic) + if v1_id and v1_id != db_id: + ids.add(v1_id) + return ids - def _build_mappings_cache(self, diff_keys): - """Batch-fetch mappings for enrich mode (to resolve v1 state of changed mappings).""" + def _get_mappings_cache(self, ids): + """Fetch one batch of mappings with only the fields the summary uses.""" from core.mappings.models import Mapping - ids = self._collect_v1_v2_ids( - {k: self.mappings_diff.result.get(k, False) for k in diff_keys}, - self.mappings_diff, - ) return { - m.id: m - for m in Mapping.objects.filter(id__in=ids).select_related('from_concept', 'to_concept') + mapping.id: mapping + for batch in chunks(sorted(ids), 5000) + for mapping in Mapping.objects.filter(id__in=batch).select_related( + 'from_concept', 'to_concept').only(*self.MAPPING_SUMMARY_FIELDS) } - def _build_mapping_diff_index(self, diff_keys, mappings_cache=None): + @staticmethod + def _get_display_names(concept_ids): + """Display names of one batch of concepts, with their names prefetched and their sources loaded once.""" + from core.concepts.models import Concept + parents = {} + display_names = {} + for concept in Concept.objects.filter(id__in=concept_ids).prefetch_related('names'): + if concept.parent_id not in parents: + parents[concept.parent_id] = concept.parent + concept.parent = parents[concept.parent_id] + display_name = concept.display_name + display_names[concept.id] = display_name.replace('"', "'") if display_name else display_name + return display_names + + @staticmethod + def _get_new_changed_only_concept_ids(mappings_cache, same_concept_ids, changed_only): + concept_ids = set() + for mapping in mappings_cache.values(): + code = mapping.from_concept_code or get(mapping, 'from_concept.mnemonic') + if mapping.from_concept_id and code in same_concept_ids and code not in changed_only: + concept_ids.add(mapping.from_concept_id) + return concept_ids + + def _build_mapping_diff_index(self, diff_keys): """ - Group every new/removed/changed mapping by its from_concept's versioned_object_id, in a - single batched pass. Without this, attaching diff'd mappings to their concept required - one query PER concept PER diff key, each with an IN-clause spanning every mapping in that - category -- for a source with tens of thousands of changed mappings and concepts, that's - effectively quadratic and dominates both runtime and memory. + Group every new/removed/changed mapping by its from_concept's versioned_object_id, as + (mnemonic, db id) pairs only: the mappings themselves are fetched per concept batch. """ from core.mappings.models import Mapping - ids_by_key = {} - all_ids = set() + index = {} for key in diff_keys: diff = self.mappings_diff.result.get(key, False) - if isinstance(diff, dict): - db_ids = [self.mappings_diff.get_db_id_for(key, mnemonic) for mnemonic in diff[self.identity]] - ids_by_key[key] = db_ids - all_ids.update(db_ids) - - if mappings_cache: - # mappings_cache already covers every id here (plus v1 state for changed items) -- - # reuse it instead of fetching and holding the same rows in memory a second time. - mappings_by_id = mappings_cache - else: - mappings_by_id = { - m.id: m - for m in Mapping.objects.filter(id__in=all_ids).select_related('from_concept', 'to_concept') - } - - index = {} - for key, ids in ids_by_key.items(): - for mapping_id in ids: - mapping = mappings_by_id.get(mapping_id) - from_concept = get(mapping, 'from_concept') - if not from_concept: - continue - index.setdefault(from_concept.versioned_object_id, {}).setdefault(key, []).append(mapping) + if not isinstance(diff, dict): + continue + mnemonics_by_id = {} + for mnemonic in diff[self.identity]: + db_id = self.mappings_diff.get_db_id_for(key, mnemonic) + if db_id: + mnemonics_by_id.setdefault(db_id, []).append(mnemonic) + for batch in chunks(sorted(mnemonics_by_id), 5000): + for mapping_id, versioned_object_id in Mapping.objects.filter(id__in=batch).values_list( + 'id', 'from_concept__versioned_object_id'): + if versioned_object_id: + index.setdefault(versioned_object_id, {}).setdefault(key, []).extend( + (mnemonic, mapping_id) for mnemonic in mnemonics_by_id[mapping_id]) return index @staticmethod @@ -536,75 +577,83 @@ def _v1_mapping_for(self, mnemonic, mapping_db_id, mappings_cache): return None def process(self): # pylint: disable=too-many-locals,too-many-branches,too-many-statements - from core.mappings.models import Mapping - from core.concepts.models import Concept concepts_result = {} mappings_result = {} traversed_mappings = set() traversed_concepts = set() - diff_keys = ['new', 'removed', 'changed_retired', 'changed_major', 'changed_minor'] + diff_keys = self.DIFF_KEYS include_changelog_enrichment = self.verbosity >= CHANGELOG_ENRICHMENT_VERBOSITY + if include_changelog_enrichment and self.get_diff_size() > settings.CHANGELOG_ENRICHMENT_MAX_RESOURCES: + include_changelog_enrichment = False + self.enrichment_skipped = True - concepts_cache = self._build_concepts_cache(diff_keys) if include_changelog_enrichment else {} - mappings_cache = self._build_mappings_cache(diff_keys) if include_changelog_enrichment else {} - mapping_diff_index = self._build_mapping_diff_index(diff_keys, mappings_cache) + mapping_diff_index = self._build_mapping_diff_index(diff_keys) for key in diff_keys: # pylint: disable=too-many-nested-blocks diff = self.concepts_diff.result.get(key, False) if isinstance(diff, dict): section_summary = {} - for concept_id in diff[self.identity]: - if concept_id in traversed_concepts: - continue - traversed_concepts.add(concept_id) - concept_db_id = self.concepts_diff.get_db_id_for(key, concept_id) - if include_changelog_enrichment: + for batch in chunks(diff[self.identity], self.CONCEPTS_BATCH_SIZE): + concepts_cache = self._get_concepts_cache(key, batch, include_changelog_enrichment) + batch_mapping_diffs = {} + for concept in concepts_cache.values(): + for mapping_diff_key, entries in mapping_diff_index.get( + concept.versioned_object_id, {}).items(): + batch_mapping_diffs.setdefault(mapping_diff_key, []).extend( + mnemonic for mnemonic, _ in entries) + mappings_cache = self._get_mappings_cache({ + _id for mapping_diff_key, mnemonics in batch_mapping_diffs.items() + for _id in self._get_mapping_db_ids( + mapping_diff_key, mnemonics, include_changelog_enrichment)}) + for concept_id in batch: + if concept_id in traversed_concepts: + continue + traversed_concepts.add(concept_id) + concept_db_id = self.concepts_diff.get_db_id_for(key, concept_id) concept = concepts_cache.get(concept_db_id) - else: - concept = Concept.objects.filter(id=concept_db_id).first() - concept_display_name = get(concept, 'display_name') - if concept_display_name: - concept_display_name = concept_display_name.replace('"', "'") - summary = { - 'id': concept_id, - 'display_name': concept_display_name - } - if include_changelog_enrichment and concept: - summary['concept_class'] = getattr(concept, 'concept_class', None) - summary['datatype'] = getattr(concept, 'datatype', None) - summary['names'] = self._names_list(concept) - summary['descriptions'] = self._descriptions_list(concept) - if key in ('changed_major', 'changed_minor'): - v1_id = get_map_id(self.concepts_diff.resources1_map, concept_id) - if v1_id and v1_id != concept_db_id: - v1_concept = concepts_cache.get(v1_id) - if v1_concept: - summary['prev_names'] = self._names_list(v1_concept) - summary['prev_descriptions'] = self._descriptions_list(v1_concept) - summary['prev_concept_class'] = getattr(v1_concept, 'concept_class', None) - summary['prev_datatype'] = getattr(v1_concept, 'datatype', None) - mappings_diff_summary = {} - for mapping_diff_key, mappings in mapping_diff_index.get( - get(concept, 'versioned_object_id'), {}).items(): - for mapping in mappings: - mnemonic = get(mapping, self.identity) - if mnemonic in traversed_mappings: - continue - if mapping_diff_key not in mappings_diff_summary: - mappings_diff_summary[mapping_diff_key] = [] - v1_mapping = None - if ( - include_changelog_enrichment and - mapping_diff_key in ('changed_major', 'changed_minor') - ): - v1_mapping = self._v1_mapping_for(mnemonic, mapping.id, mappings_cache) - mappings_diff_summary[mapping_diff_key].append( - self.get_mapping_summary(mapping, v1_mapping=v1_mapping) - ) - traversed_mappings.add(mnemonic) - if mappings_diff_summary: - summary['mappings'] = mappings_diff_summary - section_summary[concept_id] = summary + concept_display_name = get(concept, 'display_name') + if concept_display_name: + concept_display_name = concept_display_name.replace('"', "'") + summary = { + 'id': concept_id, + 'display_name': concept_display_name + } + if include_changelog_enrichment and concept: + summary['concept_class'] = getattr(concept, 'concept_class', None) + summary['datatype'] = getattr(concept, 'datatype', None) + summary['names'] = self._names_list(concept) + summary['descriptions'] = self._descriptions_list(concept) + if key in ('changed_major', 'changed_minor'): + v1_id = get_map_id(self.concepts_diff.resources1_map, concept_id) + if v1_id and v1_id != concept_db_id: + v1_concept = concepts_cache.get(v1_id) + if v1_concept: + summary['prev_names'] = self._names_list(v1_concept) + summary['prev_descriptions'] = self._descriptions_list(v1_concept) + summary['prev_concept_class'] = getattr(v1_concept, 'concept_class', None) + summary['prev_datatype'] = getattr(v1_concept, 'datatype', None) + mappings_diff_summary = {} + for mapping_diff_key, entries in mapping_diff_index.get( + get(concept, 'versioned_object_id'), {}).items(): + for mnemonic, mapping_db_id in entries: + mapping = mappings_cache.get(mapping_db_id) + if mapping is None or mnemonic in traversed_mappings: + continue + if mapping_diff_key not in mappings_diff_summary: + mappings_diff_summary[mapping_diff_key] = [] + v1_mapping = None + if ( + include_changelog_enrichment and + mapping_diff_key in ('changed_major', 'changed_minor') + ): + v1_mapping = self._v1_mapping_for(mnemonic, mapping.id, mappings_cache) + mappings_diff_summary[mapping_diff_key].append( + self.get_mapping_summary(mapping, v1_mapping=v1_mapping) + ) + traversed_mappings.add(mnemonic) + if mappings_diff_summary: + summary['mappings'] = mappings_diff_summary + section_summary[concept_id] = summary if section_summary: concepts_result[key] = section_summary # Read the raw same_standard/same_smart dicts directly rather than the exposed @@ -619,36 +668,34 @@ def process(self): # pylint: disable=too-many-locals,too-many-branches,too-many diff = self.mappings_diff.result.get(key, False) if isinstance(diff, dict): section_summary = {} - for mapping_id in diff[self.identity]: - if mapping_id in traversed_mappings: - continue - traversed_mappings.add(mapping_id) - mapping_db_id = self.mappings_diff.get_db_id_for(key, mapping_id) - if mappings_cache: + for batch in chunks(diff[self.identity], self.CONCEPTS_BATCH_SIZE): + batch = [mapping_id for mapping_id in batch if mapping_id not in traversed_mappings] + mappings_cache = self._get_mappings_cache( + self._get_mapping_db_ids(key, batch, include_changelog_enrichment)) + display_names = self._get_display_names(self._get_new_changed_only_concept_ids( + mappings_cache, same_concept_ids, concepts_result.get('changed_mappings_only', {}))) + for mapping_id in batch: + if mapping_id in traversed_mappings: + continue + traversed_mappings.add(mapping_id) + mapping_db_id = self.mappings_diff.get_db_id_for(key, mapping_id) mapping = mappings_cache.get(mapping_db_id) - else: - mapping = Mapping.objects.filter(id=mapping_db_id).first() - v1_mapping = None - if include_changelog_enrichment and key in ('changed_major', 'changed_minor'): - v1_mapping = self._v1_mapping_for(mapping_id, mapping_db_id, mappings_cache) - mapping_summary = self.get_mapping_summary(mapping, mapping_id, v1_mapping=v1_mapping) - from_concept_code = get(mapping, 'from_concept_code') or get(mapping.from_concept, 'mnemonic') - if from_concept_code and from_concept_code in same_concept_ids: - if 'changed_mappings_only' not in concepts_result: - concepts_result['changed_mappings_only'] = {} - if from_concept_code not in concepts_result['changed_mappings_only']: - concept_display_name = get(mapping.from_concept, 'display_name') - if concept_display_name: - concept_display_name = concept_display_name.replace('"', "'") - concepts_result['changed_mappings_only'][from_concept_code] = { - 'id': from_concept_code, - 'display_name': concept_display_name, - 'mappings': {} - } - mappings_dict = concepts_result['changed_mappings_only'][from_concept_code]['mappings'] - mappings_dict.setdefault(key, []).append(mapping_summary) - else: - section_summary[mapping_id] = mapping_summary + v1_mapping = None + if include_changelog_enrichment and key in ('changed_major', 'changed_minor'): + v1_mapping = self._v1_mapping_for(mapping_id, mapping_db_id, mappings_cache) + mapping_summary = self.get_mapping_summary(mapping, mapping_id, v1_mapping=v1_mapping) + from_concept_code = get(mapping, 'from_concept_code') or get(mapping, 'from_concept.mnemonic') + if from_concept_code and from_concept_code in same_concept_ids: + changed_only = concepts_result.setdefault('changed_mappings_only', {}) + if from_concept_code not in changed_only: + changed_only[from_concept_code] = { + 'id': from_concept_code, + 'display_name': display_names.get(mapping.from_concept_id), + 'mappings': {} + } + changed_only[from_concept_code]['mappings'].setdefault(key, []).append(mapping_summary) + else: + section_summary[mapping_id] = mapping_summary if section_summary: mappings_result[key] = section_summary self.result = { @@ -771,6 +818,8 @@ def get_changelog_result( # pylint: disable=too-many-arguments }, **log.result } + if log.enrichment_skipped: + result['meta']['enrichment_skipped'] = True if verbosity > 0: concepts_diff.set_concise_result() mappings_diff.set_concise_result() diff --git a/core/common/mixins.py b/core/common/mixins.py index d15cda8e7..2a51ad139 100644 --- a/core/common/mixins.py +++ b/core/common/mixins.py @@ -1,4 +1,5 @@ import logging +from copy import deepcopy from math import ceil from urllib import parse from urllib.parse import urlencode @@ -666,6 +667,8 @@ class SourceChildMixin(ChecksumModel): _counted = models.BooleanField(default=True, null=True, blank=True) _index = models.BooleanField(default=True) + ISOLATED_VERSION_DEFAULTS = {'extras': {}, 'retired': False} + # relations the list serializers read on every row, prefetched once per page by load_list_fields LIST_PREFETCHES = () # relations named like the serializer field that reads them; prefetched by load_list_fields when it's serialized @@ -1134,6 +1137,26 @@ def __update_latest_version(self, index, is_latest_version, parent, remove_paren if add_parent: self.sources.add(parent) + def set_version_fields(self, data, fields, isolated=False): + for field in fields: + if isolated: + value = data.get(field) + setattr(self, field, deepcopy(self.ISOLATED_VERSION_DEFAULTS.get(field)) if value is None else value) + else: + setattr(self, field, data.get(field, getattr(self, field))) + + def restore_to_head(self): + versioned_object = self.versioned_object + parent = versioned_object.parent + if not versioned_object.is_active: + versioned_object.is_active = True + versioned_object.save(update_fields=['is_active']) + versioned_object.sources.add(parent) + latest_version = versioned_object.get_latest_version() + if latest_version: + latest_version.sources.add(parent) + return latest_version + def unmark_latest_version(self, index=True, parent=None): parent = parent or self.parent self.__update_latest_version(index, False, parent, True, False) diff --git a/core/common/models.py b/core/common/models.py index de6aa7dd8..86a3a9246 100644 --- a/core/common/models.py +++ b/core/common/models.py @@ -1287,6 +1287,7 @@ def persist_new_version_in_transaction(cls, obj, head, user, sync, is_repeatable cursor.execute('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ') # Serialize version creation with destructive child operations on the HEAD repository. head = cls.objects.select_for_update().get(id=head.id) + obj.apply_version_metadata_to_head(head) obj.update_version_data(head) obj.save(**kwargs) @@ -1329,6 +1330,9 @@ def index_resources_for_self_as_latest_released(self, only_update=False): # pyl def can_seed_from_head_expansion(self): return False + def apply_version_metadata_to_head(self, head): + pass + @classmethod def persist_changes(cls, obj, updated_by, original_schema, **kwargs): # pylint: disable=too-many-locals,too-many-branches errors = {} diff --git a/core/common/swagger_parameters.py b/core/common/swagger_parameters.py index 0f1b4a621..0628916dd 100644 --- a/core/common/swagger_parameters.py +++ b/core/common/swagger_parameters.py @@ -111,6 +111,12 @@ f"When omitted, imports of up to {settings.IMPORT_INDEX_MAX_LINES:,} lines are indexed and larger " "ones aren't" ) +isolated_version_param = openapi.Parameter( + 'isolated_version', openapi.IN_QUERY, type=openapi.TYPE_STRING, default='false', + description="true | false. The import is one whole source version: its concepts and mappings, then exactly one " + "Source Version line. HEAD keeps only what the import contains, and the version is created isolated, " + "unless some of the content fails to import" +) file_upload_param = openapi.Parameter( 'file', openapi.IN_FORM, description="JSON Content File (json, csv or zip)", type=openapi.TYPE_FILE ) diff --git a/core/common/tasks.py b/core/common/tasks.py index d3a02976c..9d17b901b 100644 --- a/core/common/tasks.py +++ b/core/common/tasks.py @@ -16,7 +16,7 @@ from django.core.management import call_command from django.core.paginator import Paginator from django.db import DatabaseError -from django.db.models import Exists, OuterRef, F +from django.db.models import Exists, OuterRef from django.template.loader import render_to_string from django.utils import timezone from django_elasticsearch_dsl.registries import registry @@ -265,40 +265,47 @@ def bulk_import(to_import, username, update_if_exists): @app.task(base=QueueOnceCustomTask, bind=True, retry_kwargs={'max_retries': 0}) -def bulk_import_parallel_inline(self, to_import, username, update_if_exists, threads=5, index=None): # pylint: disable=too-many-arguments +def bulk_import_parallel_inline( # pylint: disable=too-many-arguments + self, to_import, username, update_if_exists, threads=5, index=None, isolated_version=False): from core.importers.models import BulkImportParallelRunner try: importer = BulkImportParallelRunner( content=to_import, username=username, update_if_exists=update_if_exists, - parallel=threads, self_task_id=self.request.id, index=index + parallel=threads, self_task_id=self.request.id, index=index, isolated_version=isolated_version ) except JSONDecodeError as ex: return {'error': f"Invalid JSON ({ex.msg})"} except ValidationError as ex: - return {'error': f"Invalid Input ({ex.message})"} + return {'error': f"Invalid Input ({'; '.join(ex.messages)})"} return importer.run() @app.task(base=QueueOnceCustomTask, retry_kwargs={'max_retries': 0}) -def bulk_import_inline(to_import, username, update_if_exists, index=None): +def bulk_import_inline(to_import, username, update_if_exists, index=None, isolated_version=False): from core.importers.models import BulkImportInline return BulkImportInline( - content=to_import, username=username, update_if_exists=update_if_exists, index=index).run() + content=to_import, username=username, update_if_exists=update_if_exists, index=index, + isolated_version=isolated_version).run() # pylint: disable=too-many-arguments @app.task(bind=True, base=QueueOnceCustomTask, retry_kwargs={'max_retries': 0}) -def bulk_import_new(self, path, username, owner_type, owner, import_type='default', index=None): +def bulk_import_new( # pylint: disable=too-many-arguments + self, path, username, owner_type, owner, import_type='default', index=None, isolated_version=False): from core.importers.importer import Importer task_id = self.request.id - return Importer(task_id, path, username, owner_type, owner, import_type, index).run() + return Importer(task_id, path, username, owner_type, owner, import_type, index, isolated_version).run() # pylint: disable=too-many-arguments @app.task(retry_kwargs={'max_retries': 0}, compression='gzip') -def bulk_import_subtask(path, username, owner_type, owner, resource_type, files, index=True): +def bulk_import_subtask( # pylint: disable=too-many-arguments + path, username, owner_type, owner, resource_type, files, index=True, isolated=False, isolated_version=False, + previous_subtask_ids=None): from core.importers.importer import ImporterSubtask - return ImporterSubtask(path, username, owner_type, owner, resource_type, files, index).run() + return ImporterSubtask( + path, username, owner_type, owner, resource_type, files, index, isolated, isolated_version, + previous_subtask_ids).run() @app.task(retry_kwargs={'max_retries': 0}, compression='gzip') @@ -334,11 +341,11 @@ def import_finisher(task_id): @app.task(bind=True, retry_kwargs={'max_retries': 0}) -def bulk_import_parts_inline(self, input_list, username, update_if_exists, index=True): +def bulk_import_parts_inline(self, input_list, username, update_if_exists, index=True, isolated=False): # pylint: disable=too-many-arguments from core.importers.models import BulkImportInline return BulkImportInline( content=None, username=username, update_if_exists=update_if_exists, input_list=input_list, - self_task_id=self.request.id, skip_hierarchy_tasks=True, index=index + self_task_id=self.request.id, skip_hierarchy_tasks=True, index=index, isolated=isolated ).run() @@ -524,7 +531,8 @@ def process_hierarchy_for_new_concept(concept_id, initial_version_id, parent_con acks_late=True, reject_on_worker_lost=True ) def process_hierarchy_for_concept_version( - latest_version_id, prev_version_id, parent_concept_uris, create_parent_version): + latest_version_id, prev_version_id, parent_concept_uris, create_parent_version, + create_removed_parent_versions=True): """ Executed when a new concept version is created with new, updated or existing hierarchy 1. Associates parent concepts to the latest concept version. @@ -546,7 +554,7 @@ def process_hierarchy_for_concept_version( latest_version.set_parent_concepts_from_uris(create_parent_version) latest_version.versioned_object.parent_concepts.set(latest_version.parent_concepts.all()) - if prev_version: + if prev_version and create_removed_parent_versions: removed_parent_urls = [ url for url in old_parents if url not in list(latest_version.parent_concept_urls) ] @@ -686,72 +694,14 @@ def index_expansion_mappings(expansion_id, count=None, mapping_ids=None): # pyl @app.task -def make_hierarchy(concept_map): # pragma: no cover # pylint: disable=too-many-locals - from core.concepts.models import Concept, HierarchicalConcepts - - def chunks(values, size=5000): - values = list(values) - for index in range(0, len(values), size): - yield values[index:index + size] - - def resolve(uris): - """uri -> (id, versioned_object_id); the lowest id wins for duplicate uris, like .first()""" - resolved = {} - for batch in chunks(uris): - for concept_id, versioned_object_id, uri in Concept.objects.filter(uri__in=batch).order_by( - '-id').values_list('id', 'versioned_object_id', 'uri'): - resolved[uri] = (concept_id, versioned_object_id) - return resolved - - def latest_versions(versioned_object_ids): - """versioned_object_id -> id of its newest active latest version, like Concept.get_latest_version()""" - latest = {} - for batch in chunks(versioned_object_ids): - for versioned_object_id, concept_id in Concept.objects.filter( - versioned_object_id__in=batch, is_latest_version=True, is_active=True - ).exclude(id=F('versioned_object_id')).order_by('created_at').values_list('versioned_object_id', 'id'): - latest[versioned_object_id] = concept_id # ascending created_at, so the newest wins - return latest - - parents = resolve(concept_map.keys()) - parent_latest = latest_versions({versioned_object_id for _, versioned_object_id in parents.values()}) - children = resolve({ - url for parent_uri, urls in concept_map.items() - if parent_uri in parents and parents[parent_uri][1] in parent_latest for url in urls}) - child_latest = latest_versions({versioned_object_id for _, versioned_object_id in children.values()}) - - pairs = set() - for parent_uri, urls in concept_map.items(): - if parent_uri not in parents: - logger.info('Could not find parent %s', parent_uri) - continue - parent_id = parent_latest.get(parents[parent_uri][1]) - if not parent_id: - logger.info('Could not find parent %s latest_version', parent_uri) - continue - for url in urls: - if url not in children: - continue - child_id, child_versioned_object_id = children[url] - pairs.add((child_id, parent_id)) - if child_versioned_object_id in child_latest: - pairs.add((child_latest[child_versioned_object_id], parent_id)) - else: - logger.info('Could not find child %s latest_version', url) - - if not pairs: - return - existing = set() - parent_id_set = {parent for _, parent in pairs} - for batch in chunks({child for child, _ in pairs}): - existing.update( - HierarchicalConcepts.objects.filter( - child_id__in=batch, parent_id__in=parent_id_set).values_list('child_id', 'parent_id')) - HierarchicalConcepts.objects.bulk_create( - [HierarchicalConcepts(child_id=child, parent_id=parent) for child, parent in pairs - existing], - batch_size=5000 - ) - logger.info('Added %s hierarchy links', len(pairs - existing)) +def make_hierarchy(concept_map): + """concept_map: parent uri -> child uris. Adds the links, never removes any.""" + from core.concepts.models import ImportHierarchy + child_map = {} + for parent_uri, child_uris in concept_map.items(): + for child_uri in child_uris: + child_map.setdefault(child_uri, []).append(parent_uri) + ImportHierarchy.from_uri_map(child_map).apply() @app.task(ignore_result=True, base=QueueOnceCustomTask) @@ -973,6 +923,21 @@ def index_source_mappings( source.clear_mappings_cache() +@app.task( + autoretry_for=(Exception, WorkerLostError, ), retry_backoff=True, retry_backoff_max=600, max_retries=5, + acks_late=True, reject_on_worker_lost=True +) +def index_dropped_source_children(source_id, concept_ids, mapping_ids): + """ + Reindexes the concepts and mappings an isolated version dropped from HEAD and recounts HEAD, after the commit. + Retried with backoff; its Task records a final failure. + """ + from core.sources.models import Source + source = Source.objects.filter(id=source_id).first() + if source: + source.index_dropped_children(concept_ids, mapping_ids) + + @app.task( bind=True, autoretry_for=(Exception, WorkerLostError, ), retry_backoff=True, retry_backoff_max=600, max_retries=2, acks_late=True, reject_on_worker_lost=True diff --git a/core/common/tests.py b/core/common/tests.py index 2917a50ac..6885aaad3 100644 --- a/core/common/tests.py +++ b/core/common/tests.py @@ -2387,14 +2387,15 @@ def test_bulk_import_new_passes_index(self, importer_mock): 'some/path', 'ocladmin', 'Organization', 'org1', 'npm', False ) - importer_mock.assert_called_once_with(ANY, 'some/path', 'ocladmin', 'Organization', 'org1', 'npm', False) + importer_mock.assert_called_once_with( + ANY, 'some/path', 'ocladmin', 'Organization', 'org1', 'npm', False, False) @patch('core.importers.importer.ImporterSubtask') def test_bulk_import_subtask_passes_index(self, subtask_mock): bulk_import_subtask('some/path', 'ocladmin', 'Organization', 'org1', 'Concept', ['f1.json'], False) subtask_mock.assert_called_once_with( - 'some/path', 'ocladmin', 'Organization', 'org1', 'Concept', ['f1.json'], False) + 'some/path', 'ocladmin', 'Organization', 'org1', 'Concept', ['f1.json'], False, False, False, None) @patch('core.common.tasks.chord') def test_bulk_import_queue(self, chord_mock): diff --git a/core/common/utils.py b/core/common/utils.py index 791155fda..963d8ab97 100644 --- a/core/common/utils.py +++ b/core/common/utils.py @@ -429,7 +429,8 @@ def es_get(url, **kwargs): def queue_bulk_import( # pylint: disable=too-many-arguments,too-many-locals - to_import, import_queue, username, update_if_exists, threads=None, inline=False, sub_task=False, index=None + to_import, import_queue, username, update_if_exists, threads=None, inline=False, sub_task=False, index=None, + isolated_version=False ): """ Used to queue bulk imports. It assigns a bulk import task to a specified import queue or a random one. @@ -462,7 +463,7 @@ def queue_bulk_import( # pylint: disable=too-many-arguments,too-many-locals # Consider removing bulk_import_inline and this branch once confirmed no external # callers rely on it (check API consumers and the test suite first). task_func = bulk_import_inline - args = (*args, index) + args = (*args, index, True) if isolated_version else (*args, index) else: task_func = bulk_import task = get_queue_task_names(import_queue, username, name=task_func.__name__) diff --git a/core/concepts/models.py b/core/concepts/models.py index 991558f7d..a64f51ec0 100644 --- a/core/concepts/models.py +++ b/core/concepts/models.py @@ -15,7 +15,7 @@ from core.common.tasks import process_hierarchy_for_new_concept, process_hierarchy_for_concept_version, \ process_hierarchy_for_new_parent_concept_version, update_mappings_concept from core.common.utils import generate_temp_version, drop_version, \ - startswith_temp_version, is_versioned_uri, decode_string + startswith_temp_version, is_versioned_uri, decode_string, chunks, is_url_encoded_string, encode_string from core.concepts.constants import CONCEPT_TYPE, LOCALES_FULLY_SPECIFIED, LOCALES_SHORT, LOCALES_SEARCH_INDEX_TERM, \ CONCEPT_WAS_RETIRED, CONCEPT_IS_ALREADY_RETIRED, CONCEPT_IS_ALREADY_NOT_RETIRED, CONCEPT_WAS_UNRETIRED, \ ALREADY_EXISTS, CONCEPT_REGEX, MAX_LOCALES_LIMIT, \ @@ -303,6 +303,7 @@ def get_validation_errors(error): CHECKSUM_INCLUSIONS = [ 'extras', 'concept_class', 'datatype', 'retired' ] + ISOLATED_VERSION_DEFAULTS = {**SourceChildMixin.ISOLATED_VERSION_DEFAULTS, 'datatype': 'None'} # $cascade as hierarchy attributes cascaded_entries = None @@ -622,7 +623,7 @@ def create_initial_version(cls, concept, **kwargs): @classmethod def create_new_version_for( cls, instance, data, user, create_parent_version=True, add_prev_version_children=True, - _hierarchy_processing=False, is_patch=False + _hierarchy_processing=False, is_patch=False, isolated=False ): # pylint: disable=too-many-arguments,too-many-locals mappings_payload = data.pop('mappings_payload', None) prev_latest = Concept.objects.filter( @@ -631,12 +632,10 @@ def create_new_version_for( instance.id = None # Clear id so it is persisted as a new object instance.version = data.get('version', None) instance.concept_class = data.get('concept_class', instance.concept_class) - instance.datatype = data.get('datatype', instance.datatype) - instance.extras = data.get('extras', instance.extras) - instance.external_id = data.get('external_id', instance.external_id) instance.comment = data.get('update_comment') or data.get('comment') - instance.retire_reason = data.get('retire_reason', instance.retire_reason) - instance.retired = data.get('retired', instance.retired) + instance.set_version_fields( + data, ('datatype', 'extras', 'external_id', 'retire_reason', 'retired'), isolated) + instance._isolated = isolated # pylint: disable=protected-access if is_patch: prev = instance.versions.exclude(id=instance.id).filter(is_latest_version=True).first() new_names = ConceptName.build(data.get('names', [])) if 'names' in data else ( @@ -653,11 +652,13 @@ def create_new_version_for( instance.cloned_names = compact(new_names) instance.cloned_descriptions = compact(new_descriptions) - if has_parent_concept_uris_attr: + prev_parent_uris = list(prev_latest.parent_concept_urls) if prev_latest else [] + if isolated or has_parent_concept_uris_attr: parent_concept_uris = cls.get_viewable_parent_uris( - parent_concept_uris or [], user, lambda: prev_latest.parent_concept_urls if prev_latest else []) + parent_concept_uris or [], user, lambda: prev_parent_uris) else: - parent_concept_uris = list(prev_latest.parent_concept_urls) if prev_latest else [] + parent_concept_uris = prev_parent_uris + has_parent_change = isolated and {drop_version(uri) for uri in parent_concept_uris} != set(prev_parent_uris) errors = instance.save_as_new_version( user=user, @@ -665,7 +666,7 @@ def create_new_version_for( parent_concept_uris=parent_concept_uris, add_prev_version_children=add_prev_version_children, _hierarchy_processing=_hierarchy_processing, - skip_duplicate_version_check=bool(mappings_payload) + skip_duplicate_version_check=bool(mappings_payload) or has_parent_change ) if errors or mappings_payload is None: @@ -1270,12 +1271,13 @@ def _process_prev_latest_version_hierarchy(self, prev_latest, add_prev_version_c def _process_latest_version_hierarchy(self, prev_latest, parent_concept_uris=None, create_parent_version=True): task = process_hierarchy_for_concept_version + args = (self.id, get(prev_latest, 'id'), parent_concept_uris, create_parent_version) + if get(self, '_isolated'): + args = (*args, False) if get(settings, 'TEST_MODE', False): - task(self.id, get(prev_latest, 'id'), parent_concept_uris, create_parent_version) + task(*args) else: - task.apply_async( - (self.id, get(prev_latest, 'id'), parent_concept_uris, create_parent_version), - queue='concurrent', permanent=False) + task.apply_async(args, queue='concurrent', permanent=False) @staticmethod def validate_locales_limit(names, descriptions): @@ -1408,6 +1410,58 @@ def get_hierarchy_queryset(self, relation, repo_version, filters=None): queryset = Concept.objects.filter(uri__in=self.get_hierarchy_concept_urls(relation, not repo_version.is_head)) return queryset.filter(**filters) + @staticmethod + def get_import_hierarchy_map(input_list, include_parentless=False): + hierarchy_map = {} + for line in input_list: + if (line.get('type', '') or '').lower() != 'concept': + continue + parent_urls = line.get('parent_concept_urls') or [] + concept_id = line.get('id') + owner = line.get('owner') + source = line.get('source') + if (parent_urls or include_parentless) and concept_id and owner and source: + owner_type = line.get('owner_type', '').lower() + owner_prefix = 'users' if owner_type in ['user', 'users'] else 'orgs' + if not is_url_encoded_string(concept_id): + concept_id = encode_string(concept_id, safe='') + hierarchy_map[f'/{owner_prefix}/{owner}/sources/{source}/concepts/{concept_id}/'] = parent_urls + return hierarchy_map + + @classmethod + def reconcile_import_hierarchy(cls, hierarchy_map, user, replace_head=False): + ImportHierarchy.from_uri_map(hierarchy_map, user).apply(replace_head) + + @classmethod + def get_versioned_objects_by_uri(cls, uris): + resolved = {} + for batch in chunks(sorted(uris), 5000): + for uri, concept_id, source_id in cls.objects.filter( + uri__in=batch, id=F('versioned_object_id')).values_list('uri', 'id', 'parent_id'): + resolved[uri] = (concept_id, source_id) + return resolved + + @classmethod + def get_latest_version_ids_by_versioned_object_id(cls, versioned_object_ids): + latest = {} + for batch in chunks(sorted(compact(versioned_object_ids)), 5000): + for versioned_object_id, concept_id in cls.objects.filter( + versioned_object_id__in=batch, is_latest_version=True, is_active=True + ).exclude(id=F('versioned_object_id')).order_by('created_at').values_list('versioned_object_id', 'id'): + latest[versioned_object_id] = concept_id + return latest + + def get_version_hierarchy_queryset(self, relation, repo_version): + if relation == 'parent_concepts': + member = repo_version.concepts.filter(versioned_object_id=self.versioned_object_id).first() + if not member: + return Concept.objects.none() + return repo_version.concepts.filter( + versioned_object_id__in=member.parent_concepts.values('versioned_object_id')) + return repo_version.concepts.filter( + id__in=Concept.objects.filter( + parent_concepts__versioned_object_id=self.versioned_object_id).values('id')) + def child_concept_queryset(self): urls = self.child_concept_urls if urls: @@ -1939,3 +1993,79 @@ def get_serializer_class(verbose=False, version=False, brief=False, cascade=Fals @staticmethod def get_brief_serializer(): return Concept.get_serializer_class(brief=True) + + +class ImportHierarchy: + """ + An import's parent links, resolved once into versioned object ids. apply() links each child's versioned object + and latest version to its parents' latest versions, add-only, or replaces the versioned objects' links. + """ + CHUNK_SIZE = 5000 + + def __init__(self, links): + self.links = links # child versioned object id -> parent versioned object ids + self.latest = Concept.get_latest_version_ids_by_versioned_object_id( + {*links, *(parent_id for parent_ids in links.values() for parent_id in parent_ids)}) + + @classmethod + def from_uri_map(cls, hierarchy_map, user=None): + """hierarchy_map: child uri -> parent uris. With a user, only children it can edit and parents it can view.""" + from core.sources.models import Source + resolved = Concept.get_versioned_objects_by_uri( + {*hierarchy_map, *(uri for uris in hierarchy_map.values() for uri in uris)}) + editable = viewable = None + if user: + sources = Source.objects.filter(id__in={source_id for _, source_id in resolved.values()}) + editable = {source.id for source in sources if source.has_edit_access(user)} + viewable = {source.id for source in sources if source.has_view_access(user)} + links = {} + for child_uri, parent_uris in hierarchy_map.items(): + child_id, child_source_id = resolved.get(child_uri, (None, None)) + if not child_id or (editable is not None and child_source_id not in editable): + continue + links.setdefault(child_id, set()).update( + resolved[uri][0] for uri in parent_uris + if uri in resolved and (viewable is None or resolved[uri][1] in viewable)) + return cls(links) + + def apply(self, replace_head=False): + for batch in chunks(sorted(self.links), self.CHUNK_SIZE): + parents = { + child_id: {self.latest[parent_id] for parent_id in self.links[child_id] if parent_id in self.latest} + for child_id in batch + } + pairs = { + (self.latest[child_id], parent_id) for child_id, parent_ids in parents.items() + if child_id in self.latest for parent_id in parent_ids} + if replace_head: + self.replace_links(parents) + else: + pairs.update( + (child_id, parent_id) for child_id, parent_ids in parents.items() for parent_id in parent_ids) + self.add_links(pairs) + + @staticmethod + def add_links(pairs): + if not pairs: + return + existing = set(HierarchicalConcepts.objects.filter( + child_id__in={child_id for child_id, _ in pairs}, parent_id__in={parent_id for _, parent_id in pairs} + ).values_list('child_id', 'parent_id')) + HierarchicalConcepts.objects.bulk_create( + [HierarchicalConcepts(child_id=child_id, parent_id=parent_id) for child_id, parent_id in pairs - existing], + batch_size=5000) + + @staticmethod + def replace_links(parents): + """parents: child versioned object id -> the parent ids it should have, and no others.""" + existing = {} + for child_id, parent_id in HierarchicalConcepts.objects.filter( + child_id__in=parents).values_list('child_id', 'parent_id'): + existing.setdefault(child_id, set()).add(parent_id) + changed = [child_id for child_id, parent_ids in parents.items() if existing.get(child_id, set()) != parent_ids] + if changed: + HierarchicalConcepts.objects.filter(child_id__in=changed).delete() + HierarchicalConcepts.objects.bulk_create( + [HierarchicalConcepts(child_id=child_id, parent_id=parent_id) + for child_id in changed for parent_id in parents[child_id]], + batch_size=5000) diff --git a/core/concepts/views.py b/core/concepts/views.py index a0c472ef2..a1ea4f1a7 100644 --- a/core/concepts/views.py +++ b/core/concepts/views.py @@ -602,7 +602,13 @@ def get_clone_to_source(self): return source -class ConceptChildrenView(ConceptBaseView, ListWithHeadersMixin): +class ConceptHierarchyMixin: + def is_source_version(self): + parent = get(self, 'parent_resource') + return bool(parent) and parent.__class__.__name__ == 'Source' and not parent.is_head + + +class ConceptChildrenView(ConceptHierarchyMixin, ConceptBaseView, ListWithHeadersMixin): is_page_size_capped = True is_searchable = False # a search would list the repo's concepts, not this concept's hierarchy serializer_class = ConceptChildrenSerializer @@ -617,13 +623,16 @@ def get_queryset(self): raise Http404() self.check_object_permissions(self.request, instance) + if self.is_source_version(): + return get_viewable_concepts( + instance.get_version_hierarchy_queryset('child_concepts', self.parent_resource), self.request) return get_viewable_concepts(instance.child_concept_queryset(), self.request) def get(self, request, *args, **kwargs): return self.list(request, *args, **kwargs) -class ConceptParentsView(ConceptBaseView, ListWithHeadersMixin): +class ConceptParentsView(ConceptHierarchyMixin, ConceptBaseView, ListWithHeadersMixin): is_page_size_capped = True is_searchable = False # a search would list the repo's concepts, not this concept's hierarchy serializer_class = ConceptParentsSerializer @@ -637,6 +646,9 @@ def get_queryset(self): raise Http404() self.check_object_permissions(self.request, instance) + if self.is_source_version(): + return get_viewable_concepts( + instance.get_version_hierarchy_queryset('parent_concepts', self.parent_resource), self.request) return get_viewable_concepts(instance.parent_concept_queryset(), self.request) def get(self, request, *args, **kwargs): diff --git a/core/importers/constants.py b/core/importers/constants.py index 7df02df53..146930f11 100644 --- a/core/importers/constants.py +++ b/core/importers/constants.py @@ -2,3 +2,5 @@ INVALID_UPDATE_IF_EXISTS = "update_if_exists must be either 'true' or 'false'" INVALID_INDEX = "index must be either 'true' or 'false'" NO_CONTENT_TO_IMPORT = 'No content to import' +INVALID_ISOLATED_VERSION = "'isolated_version' must be 'true' or 'false'." +ISOLATED_VERSION_IMPORT_TYPE = 'isolated_version is not supported for npm imports.' diff --git a/core/importers/importer.py b/core/importers/importer.py index f118ba985..83d974e23 100644 --- a/core/importers/importer.py +++ b/core/importers/importer.py @@ -19,6 +19,7 @@ from packaging.version import Version from pydantic import BaseModel, computed_field, PrivateAttr +from django.core.exceptions import ValidationError as DjangoValidationError from django.utils import timezone from pydash import get from rest_framework.exceptions import ValidationError @@ -26,12 +27,14 @@ from core import settings from core.common.serializers import IdentifierSerializer from core.common.tasks import bulk_import_subtask, bulk_import_subtask_empty, bulk_import_queue +from core.concepts.models import Concept +from core.sources.constants import ISOLATED_VERSION_NEEDS_ONE_SOURCE_VERSION from core.common.tasks import import_finisher from core.code_systems.converter import CodeSystemConverter from core.common.utils import get_export_service from core.importers.models import SourceImporter, SourceVersionImporter, ConceptImporter, OrganizationImporter, \ CollectionImporter, CollectionVersionImporter, MappingImporter, ReferenceImporter, ImportIndexer, CREATED, \ - UPDATED, FAILED, DELETED, NOT_FOUND, PERMISSION_DENIED, UNCHANGED, should_index_import + UPDATED, FAILED, DELETED, NOT_FOUND, PERMISSION_DENIED, UNCHANGED, should_index_import, IsolatedVersionCommit from core.orgs.models import Organization from core.sources.models import Source from core.users.models import UserProfile @@ -156,6 +159,8 @@ def summary(self) -> ImportTaskSummary: # pylint: disable=too-many-branches results = [child.result] for result in results: + if isinstance(result, dict): + continue summary.processed += 1 if result == CREATED: summary.created += 1 @@ -214,11 +219,14 @@ class Importer: owner: str import_type: str = 'default' index: bool | None = None # None: whether the import is small enough (should_index_import) + isolated_version: bool = False MIN_BATCH_SIZE: int = 50 IMPORT_CACHE: str = "import_cache/" # pylint: disable=too-many-arguments - def __init__(self, task_id, path, username, owner_type, owner, import_type='default', index=None): + def __init__( # pylint: disable=too-many-arguments + self, task_id, path, username, owner_type, owner, import_type='default', index=None, + isolated_version=False): super().__init__() self.task_id = task_id self.path = path @@ -227,6 +235,28 @@ def __init__(self, task_id, path, username, owner_type, owner, import_type='defa self.owner = owner self.import_type = import_type self.index = index + self.isolated_version = isolated_version + self.version_lines = [] + + def validate_versioning(self, resources, resource_types): + try: + for line in self.version_lines: + Source.resolve_versioning_mode(self.isolated_version, line.get('versioning_mode')) + if not self.isolated_version: + return resource_types + if sum(get(resources, 'Source Version', {}).values()) != 1: + raise ValidationError(ISOLATED_VERSION_NEEDS_ONE_SOURCE_VERSION) + IsolatedVersionCommit( + self.version_lines[0], UserProfile.objects.filter(username=self.username).first(), True).preflight() + except DjangoValidationError as ex: + raise ValidationError(ex.message_dict) from ex + return self.get_isolated_version_resource_types(resource_types) + + @staticmethod + def get_isolated_version_resource_types(resource_types): + resource_types = [_type for _type in resource_types if _type != 'Source Version'] + resource_types.insert(resource_types.index('Mapping') + 1, 'Source Version') + return resource_types def is_npm_import(self) -> bool: return self.import_type == 'npm' @@ -271,6 +301,7 @@ def run(self): # pylint: disable=too-many-locals dependencies = [] self.prepare_resources(self.path, resource_types, dependencies, [], resources) + resource_types = self.validate_versioning(resources, resource_types) self.index = should_index_import(self.index, sum(sum(files.values()) for files in resources.values())) tasks = self.prepare_tasks(resource_types, dependencies, resources) if tasks: @@ -443,15 +474,17 @@ def schedule_tasks(self, tasks): group_queue = [] for task in tasks: group_tasks = [] + previous_subtask_ids = list(subtask_ids) for group_task in task: # TODO: create 2 queues for new bulk import subtasks: bulk_import_subtask and bulk_import_subtask_root subtask_id = uuid() subtask_ids.append(subtask_id) - group_tasks.append(bulk_import_subtask.si(group_task['path'], group_task['username'], - group_task['owner_type'], group_task['owner'], - group_task['resource_type'], group_task['files'], - self.index) - .set(task_id=subtask_id)) + args = (group_task['path'], group_task['username'], group_task['owner_type'], group_task['owner'], + group_task['resource_type'], group_task['files'], self.index) + if self.isolated_version: + is_version = group_task['resource_type'] == 'Source Version' + args = (*args, True, is_version, previous_subtask_ids if is_version else None) + group_tasks.append(bulk_import_subtask.si(*args).set(task_id=subtask_id)) if len(group_tasks) == 1: # Prevent celery from converting group to a single task group_tasks.append(bulk_import_subtask_empty.si()) @@ -480,7 +513,9 @@ def categorize_resources(self, json_file, path, file_path, resource_types, resou try: parser = ijson.parse(json_file, multiple_values=True, allow_comments=True) + line = None # top-level scalar fields of the current resource, to keep its Source Version lines for prefix, event, value in parser: + line = self.collect_version_line(line, prefix, event, value) # Expect {"resourceType": ""}{"resourceType": ""} or {"type": ""}{"type": ""} if event == 'string' and (prefix in {'resourceType', 'type'}): # Categorize files based on resourceType @@ -498,6 +533,20 @@ def categorize_resources(self, json_file, path, file_path, resource_types, resou except JSONError as json_error: raise JSONError(f'Failed to process {path}/{file_path}') from json_error + def collect_version_line(self, line, prefix, event, value): + if not prefix: + if event == 'start_map': + return {} + if event == 'end_map' and get(line, 'type') == 'Source Version': + self.version_lines.append(line) + return line if event == 'map_key' else None + if line is not None and '.' not in prefix: + if event in ('string', 'number', 'boolean', 'null'): + line[prefix] = value + elif event in ('start_map', 'start_array'): + line[prefix] = {} if event == 'start_map' else [] + return line + class ImportRequest: path: str @@ -514,8 +563,10 @@ class ResourceImporter: converters = [CodeSystemConverter] result_type = [CREATED, UPDATED, FAILED, DELETED, NOT_FOUND, PERMISSION_DENIED, UNCHANGED] - def __init__(self, indexer=None): + def __init__(self, indexer=None, isolated=False): self.indexer = indexer # collects the concepts and mappings imported with _index=False + self.isolated = isolated + self.importer = None @staticmethod def get_resource_types(): @@ -543,7 +594,10 @@ def import_resource(self, resource, username, owner_type, owner): for resource_importer in self.resource_importers: if resource_importer.can_handle(resource): user_profile = UserProfile.objects.get(username=username) - importer = resource_importer(resource, user_profile, True) + is_content = resource_importer in (ConceptImporter, MappingImporter) + importer = resource_importer( + resource, user_profile, True, **({'isolated': self.isolated} if is_content else {})) + self.importer = importer result = importer.run() if self.indexer: self.indexer.add(get(importer, 'instance')) @@ -663,7 +717,9 @@ class ImporterSubtask: indexer: ImportIndexer # pylint: disable=too-many-arguments - def __init__(self, path, username, owner_type, owner, resource_type, files, index=True): + def __init__( + self, path, username, owner_type, owner, resource_type, files, index=True, isolated=False, + isolated_version=False, previous_subtask_ids=None): super().__init__() self.path = path self.username = username @@ -672,8 +728,13 @@ def __init__(self, path, username, owner_type, owner, resource_type, files, inde self.resource_type = resource_type self.files = files self.indexer = ImportIndexer(index) + self.isolated = isolated + self.isolated_version = isolated_version + self.previous_subtask_ids = previous_subtask_ids or [] + self.seen = {'concepts': set(), 'mappings': set()} + self.hierarchy_map = {} - def run(self): + def run(self): # pylint: disable=too-many-branches results = [] try: @@ -688,6 +749,8 @@ def run(self): ImporterUtils.fetch_to_file(remote_file, temp) is_zipped, is_tarred = ImporterUtils.is_zipped_or_tarred(temp) + if self.isolated_version: + self.collect_hierarchy_map(temp, is_zipped, is_tarred) if is_zipped: self.import_zip(temp, results) @@ -712,8 +775,46 @@ def run(self): results.extend([str(ex)] * (results_count - len(results))) self.indexer.finish() + if self.isolated and not self.isolated_version: + results.append({'seen': {key: sorted(value) for key, value in self.seen.items()}}) return results + def collect_hierarchy_map(self, temp, is_zipped, is_tarred): + def collect(json_file): + for item in ijson.items(json_file, '', multiple_values=True, allow_comments=True, use_float=True): + if isinstance(item, dict) and item.get('type') == 'Concept': + self.hierarchy_map.update(Concept.get_import_hierarchy_map([item], include_parentless=True)) + + temp.seek(0) + if is_zipped: + with ZipFile(temp) as package: + for info in package.infolist(): + if info.filename.endswith('.json') and info.file_size: + with package.open(info) as json_file: + collect(json_file) + elif is_tarred: + with tarfile.open(fileobj=temp, mode='r') as package: + for member in package.getmembers(): + if member.name.endswith('.json') and member.size: + with package.extractfile(member) as json_file: + collect(json_file) + else: + collect(temp) + temp.seek(0) + + def commit_isolated_version(self, version_line): + commit = IsolatedVersionCommit( + version_line, UserProfile.objects.get(username=self.username), True, hierarchy_map=self.hierarchy_map) + for results in self.iter_previous_results(): + commit.add_subtask_results(results) + return commit.run() + + def iter_previous_results(self): + """Each earlier subtask's results in turn, so only one is held at a time.""" + for task_id in self.previous_subtask_ids: + result = AsyncResult(task_id).result + yield result if isinstance(result, list) else [result] + def import_zip(self, temp, results): with ZipFile(temp) as package: for file in self.files: @@ -768,6 +869,11 @@ def import_files(self, temp, results): except Exception as ex: self.add_exception_to_results(file, start_index, end_index, count, ex, results) + def add_seen(self, importer, result): + versioned_object_id = importer.get_seen_versioned_object_id(result) if importer else None + if versioned_object_id: + self.seen['concepts' if isinstance(importer, ConceptImporter) else 'mappings'].add(versioned_object_id) + @staticmethod def add_exception_to_results(file, start_index, end_index, count, ex, results): error = f"Failed to process {file.get('filepath', '')}({start_index}:{end_index}) " \ @@ -784,7 +890,7 @@ def import_resource(self, json_file, filepath, start_index, end_index): # pylin count = None results = [] - for resource in ijson.items(parse, '', multiple_values=True, allow_comments=True): + for resource in ijson.items(parse, '', multiple_values=True, allow_comments=True, use_float=True): if resource.get('__action', None) == 'DELETE': continue @@ -799,10 +905,16 @@ def import_resource(self, json_file, filepath, start_index, end_index): # pylin resource['owner_type'] = self.owner_type if 'owner' not in resource: resource['owner'] = self.owner - result = ResourceImporter(self.indexer).import_resource( - resource, self.username, self.owner_type, self.owner) + if self.isolated_version and self.resource_type == 'Source Version': + resource_importer = None + result = self.commit_isolated_version(resource) + else: + resource_importer = ResourceImporter(self.indexer, self.isolated) + result = resource_importer.import_resource( + resource, self.username, self.owner_type, self.owner) if isinstance(result, int): results.append(result) + self.add_seen(get(resource_importer, 'importer'), result) else: results.append(f'{error} due to: {result}') except Exception as e: @@ -819,7 +931,7 @@ def move_to_start_index(self, json_file, start_index): json_file.seek(0) # position at the beginning of file if start_index != 0: index = 0 - parse_events = ijson.parse(json_file, multiple_values=True, allow_comments=True) + parse_events = ijson.parse(json_file, multiple_values=True, allow_comments=True, use_float=True) for prefix, event, value in parse_events: if event == 'string' and prefix in ('resourceType', 'type') and value == self.resource_type: index += 1 diff --git a/core/importers/models.py b/core/importers/models.py index 0f4575730..db551e4ab 100644 --- a/core/importers/models.py +++ b/core/importers/models.py @@ -7,6 +7,7 @@ from celery.utils.log import get_task_logger from django.conf import settings from django.core.exceptions import ValidationError +from django.db import transaction from django.db.models import F from ocldev.oclfleximporter import OclFlexImporter from pydash import compact, get @@ -16,11 +17,13 @@ from core.common import ERRBIT_LOGGER from core.common.constants import HEAD, ALL from core.common.tasks import bulk_import_parts_inline, delete_organization, batch_index_resources, \ - post_import_update_resource_counts, make_hierarchy, index_concepts_mapped_codes + post_import_update_resource_counts, index_concepts_mapped_codes from core.common.utils import drop_version, is_url_encoded_string, encode_string, to_parent_uri, chunks from core.concepts.models import Concept from core.mappings.models import Mapping from core.orgs.models import Organization +from core.sources.constants import ISOLATED_METADATA_FIELDS, ISOLATED_VERSION_CONTENT_FAILED, \ + ISOLATED_VERSION_LINE_INVALID, ISOLATED_VERSION_EXISTS, ISOLATED_VERSION_NO_EDIT_ACCESS from core.sources.models import Source from core.tasks.models import Task from core.users.models import UserProfile @@ -109,13 +112,14 @@ class BaseResourceImporter: mandatory_fields = set() allowed_fields = [] - def __init__(self, data, user, update_if_exists=False, cache=None): + def __init__(self, data, user, update_if_exists=False, cache=None, isolated=False): # pylint: disable=too-many-arguments self.user = user self.data = data self.update_if_exists = update_if_exists self.queryset = None self.index_resources = False self.cache = cache if cache is not None else {} + self.isolated = isolated @classmethod def can_handle(cls, obj): @@ -192,6 +196,17 @@ def run(self): def process(self): raise NotImplementedError() + def track_isolated_result(self, result): + if self.isolated and result in (UPDATED, UNCHANGED) and get(self, 'instance.versioned_object_id'): + if not self.instance.versioned_object.is_active: + self.instance.restore_to_head() + return result + + def get_seen_versioned_object_id(self, result): + if self.isolated and result in (CREATED, UPDATED, UNCHANGED, DELETED): + return get(self, 'instance.versioned_object_id') + return None + def retire_resource(self, resource): """ Retires a concept or mapping and keeps the version the retire saved as the instance, marked with the one it @@ -310,7 +325,10 @@ def delete(self): class SourceVersionImporter(BaseResourceImporter): mandatory_fields = {"id"} - allowed_fields = ["id", "external_id", "description", "released"] + allowed_fields = [ + "id", "released", "versioning_mode", "name", "default_locale", "supported_locales", "extras", + *ISOLATED_METADATA_FIELDS + ] @staticmethod def get_resource_type(): @@ -332,15 +350,129 @@ def parse(self): self.data['version'] = self.data.pop('id') self.data['mnemonic'] = source self.data[owner_type] = owner + self.requested_versioning_mode = self.data.pop('versioning_mode', None) + supported_locales = self.data.get('supported_locales') + if isinstance(supported_locales, str): + self.data['supported_locales'] = supported_locales.split(',') + + def __init__(self, data, user, update_if_exists=False, cache=None, isolated_version=False): # pylint: disable=too-many-arguments + super().__init__(data, user, update_if_exists, cache=cache) + self.isolated_version = isolated_version + self.requested_versioning_mode = None + + def get_versioning_mode_errors(self): + try: + self.data['versioning_mode'] = Source.resolve_versioning_mode( + self.isolated_version, self.requested_versioning_mode) + except ValidationError as ex: + return ex.message_dict + return None def process(self): + errors = self.get_versioning_mode_errors() + if errors: + return errors source = Source(**self.data) + source._metadata_fields = list(self.data.keys()) # pylint: disable=protected-access if source.has_parent_edit_access(self.user): errors = Source.persist_new_version(source, self.user) return errors or CREATED return PERMISSION_DENIED +class IsolatedVersionRollback(Exception): + def __init__(self, result): + super().__init__(result) + self.result = result + + +class IsolatedVersionCommit: + """ + The last step of an isolated version import, shared by every import path. HEAD's hierarchy replacement, the drop + of what the import didn't contain and the version share one transaction. preflight() runs the same checks before + any content imports. + """ + + def __init__( # pylint: disable=too-many-arguments + self, version_line, user, update_if_exists=False, hierarchy_map=None, seen_concept_ids=None, + seen_mapping_ids=None, content_failures=0): + self.version_line = version_line + self.user = user + self.update_if_exists = update_if_exists + self.hierarchy_map = hierarchy_map or {} + self.seen_concept_ids = set(seen_concept_ids or []) + self.seen_mapping_ids = set(seen_mapping_ids or []) + self.content_failures = content_failures + + def add_seen(self, seen): + self.seen_concept_ids.update(get(seen, 'concepts') or []) + self.seen_mapping_ids.update(get(seen, 'mappings') or []) + + def add_runner_result(self, result): + """Folds in one BulkImportInline part's result, so the parts are read once and not kept.""" + self.add_seen(get(result, 'seen')) + self.content_failures += sum(len(get(result, f'json.{key}') or []) for key in BulkImportInline.FAILURE_KEYS) + + def add_subtask_results(self, results): + """Folds in one archive subtask's results: a status per resource, and a dict of the ids it saw.""" + for result in results: + if isinstance(result, dict): + self.add_seen(get(result, 'seen')) + elif result not in (CREATED, UPDATED, UNCHANGED, DELETED): + self.content_failures += 1 + + def get_head(self): + owner_type = (self.version_line.get('owner_type', '') or '').lower() + owner_filter = 'user__username' if owner_type in ['user', 'users'] else 'organization__mnemonic' + return Source.objects.filter( + **{owner_filter: self.version_line.get('owner')}, mnemonic=self.version_line.get('source'), + version=HEAD).first() + + def get_rejection(self, importer, needs_head=True): + """(True, result) when this version would be rejected. A HEAD the import itself creates isn't there yet.""" + is_clean = importer.clean() + if not is_clean: + return True, is_clean + errors = importer.get_versioning_mode_errors() + if errors: + return True, errors + head = self.get_head() + if (needs_head and not head) or (head and not head.has_edit_access(self.user)): + return True, PERMISSION_DENIED + return False, None + + def preflight(self): + """Runs before any content imports, so a version that would be rejected leaves HEAD untouched.""" + importer = SourceVersionImporter({**self.version_line}, self.user, self.update_if_exists, isolated_version=True) + rejected, result = self.get_rejection(importer, needs_head=False) + if rejected: + raise ValidationError(result if isinstance(result, dict) else {'__all__': [{ + False: ISOLATED_VERSION_LINE_INVALID, None: ISOLATED_VERSION_EXISTS, + PERMISSION_DENIED: ISOLATED_VERSION_NO_EDIT_ACCESS}[result]]}) + + def run(self): + if self.content_failures: + return {'__all__': [ISOLATED_VERSION_CONTENT_FAILED]} + importer = SourceVersionImporter( + {**self.version_line}, self.user, self.update_if_exists, isolated_version=True) + rejected, result = self.get_rejection(importer) + if rejected: + return result + head = self.get_head() + try: + with transaction.atomic(): + Concept.reconcile_import_hierarchy(self.hierarchy_map, self.user, replace_head=True) + head.drop_unseen_children(self.seen_concept_ids, self.seen_mapping_ids) + result = importer.process() + if result != CREATED: + raise IsolatedVersionRollback(result) + except IsolatedVersionRollback as ex: + return ex.result + except ValidationError as ex: + return get(ex, 'message_dict') or {'__all__': ex.messages} + return CREATED + + class CollectionImporter(BaseResourceImporter): mandatory_fields = {'id', 'name', 'owner_type', 'owner'} allowed_fields = [ @@ -443,8 +575,9 @@ class ConceptImporter(BaseResourceImporter): def get_resource_type(): return 'Concept' - def __init__(self, data, user, update_if_exists, skip_hierarchy_tasks=False, cache=None): # pylint: disable=too-many-arguments - super().__init__(data, user, update_if_exists, cache=cache) + def __init__( # pylint: disable=too-many-arguments + self, data, user, update_if_exists, skip_hierarchy_tasks=False, cache=None, isolated=False): + super().__init__(data, user, update_if_exists, cache=cache, isolated=isolated) self.skip_hierarchy_tasks = skip_hierarchy_tasks self.version = False self.instance = None @@ -498,11 +631,11 @@ def process(self): self.instance._index = False # pylint: disable=protected-access errors = Concept.create_new_version_for( instance=self.instance, data=self.data, user=self.user, create_parent_version=False, - add_prev_version_children=False + add_prev_version_children=False, isolated=self.isolated ) if errors and Concept.is_standard_checksum_error(errors): - return UNCHANGED - return errors or UPDATED + return self.track_isolated_result(UNCHANGED) + return errors or self.track_isolated_result(UPDATED) if 'update_comment' in self.data: self.data['comment'] = self.data['update_comment'] @@ -514,7 +647,7 @@ def process(self): data=persist_data, user=self.user, create_parent_version=False) if self.instance.id: - return CREATED + return self.track_isolated_result(CREATED) return self.instance.errors or errors or FAILED return PERMISSION_DENIED @@ -528,7 +661,7 @@ def delete(self): try: if parent.has_edit_access(self.user): self.retire_resource(self.get_queryset().first()) - return DELETED + return self.track_isolated_result(DELETED) return PERMISSION_DENIED except Exception as ex: return {'errors': ex.args} @@ -548,8 +681,8 @@ class MappingImporter(BaseResourceImporter): def get_resource_type(): return 'Mapping' - def __init__(self, data, user, update_if_exists, cache=None): - super().__init__(data, user, update_if_exists, cache=cache) + def __init__(self, data, user, update_if_exists, cache=None, isolated=False): # pylint: disable=too-many-arguments + super().__init__(data, user, update_if_exists, cache=cache, isolated=isolated) self.version = False self.instance = None @@ -665,17 +798,18 @@ def process(self): self.instance = queryset.first().clone() self.instance._counted = None # pylint: disable=protected-access self.instance._index = False # pylint: disable=protected-access - errors = Mapping.create_new_version_for(self.instance, self.data, self.user, cache=self.cache) + errors = Mapping.create_new_version_for( + self.instance, self.data, self.user, cache=self.cache, isolated=self.isolated) if errors and Mapping.is_standard_checksum_error(errors): - return UNCHANGED - return errors or UPDATED + return self.track_isolated_result(UNCHANGED) + return errors or self.track_isolated_result(UPDATED) if 'update_comment' in self.data: self.data['comment'] = self.data['update_comment'] self.data.pop('update_comment') self.instance = Mapping.persist_new( {**self.data, '_counted': None, '_index': False}, self.user, cache=self.cache) if self.instance.id: - return CREATED + return self.track_isolated_result(CREATED) return self.instance.errors or errors or FAILED return PERMISSION_DENIED @@ -689,7 +823,7 @@ def delete(self): try: if parent.has_edit_access(self.user): self.retire_resource(self.get_queryset().first()) - return DELETED + return self.track_isolated_result(DELETED) return PERMISSION_DENIED except Exception as ex: return {'errors': ex.args} @@ -858,11 +992,15 @@ class BulkImportInline(BaseImporter): def __init__( # pylint: disable=too-many-arguments self, content, username, update_if_exists=False, input_list=None, user=None, set_user=True, - self_task_id=None, skip_hierarchy_tasks=False, index=None + self_task_id=None, skip_hierarchy_tasks=False, index=None, isolated_version=False, isolated=False ): super().__init__(content, username, update_if_exists, user, not bool(input_list), set_user) self.self_task_id = self_task_id self.skip_hierarchy_tasks = skip_hierarchy_tasks + self.isolated_version = isolated_version + self.isolated = isolated or isolated_version + self.seen_concept_ids = set() + self.seen_mapping_ids = set() # Lookup cache shared across this run's items (see ConceptImporter/MappingImporter). # It caches misses too (None/False), so it's only safe because a chunk is single-resource-type # in the production parallel path (BulkImportParallelRunner.make_parts splits concepts and @@ -893,10 +1031,31 @@ def __init__( # pylint: disable=too-many-arguments self.last_progress_notified_at = 0 self.elapsed_seconds = 0 self.index = should_index_import(index, self.total) + self.preflight_isolated_version(Source.validate_import_versioning_modes(self.input_list, self.isolated_version)) + + def preflight_isolated_version(self, version_line): + if version_line: + IsolatedVersionCommit(version_line, self.user, self.update_if_exists).preflight() def set_task(self): self.task = Task.objects.filter(id=self.self_task_id).first() + FAILURE_KEYS = ('failed', 'exception', 'permission_denied', 'invalid', 'unknown', 'others') + + @property + def content_failures_count(self): + return sum(len(getattr(self, key)) for key in self.FAILURE_KEYS) + + def import_source_version(self, item): + if not self.isolated_version: + return SourceVersionImporter(item, self.user, self.update_if_exists).run() + return IsolatedVersionCommit( + item, self.user, self.update_if_exists, + hierarchy_map=Concept.get_import_hierarchy_map(self.input_list, include_parentless=True), + seen_concept_ids=self.seen_concept_ids, seen_mapping_ids=self.seen_mapping_ids, + content_failures=self.content_failures_count + ).run() + def handle_item_import_result(self, result, item): # pylint: disable=too-many-return-statements if result is None: self.exists.append(item) @@ -983,9 +1142,7 @@ def run(self): # pylint: disable=too-many-branches,too-many-statements,too-many ) continue if item_type == 'source version': - self.handle_item_import_result( - SourceVersionImporter(item, self.user, self.update_if_exists).run(), original_item - ) + self.handle_item_import_result(self.import_source_version(item), original_item) continue if item_type == 'collection': collection_importer = CollectionImporter(item, self.user, self.update_if_exists) @@ -1003,10 +1160,11 @@ def run(self): # pylint: disable=too-many-branches,too-many-statements,too-many concept_importer = ConceptImporter( item, self.user, self.update_if_exists, skip_hierarchy_tasks=self.skip_hierarchy_tasks and bool(item.get('id')), - cache=self.cache + cache=self.cache, isolated=self.isolated ) _result = concept_importer.delete() if action == 'delete' else concept_importer.run() indexer.add(concept_importer.instance) + self.seen_concept_ids.update(compact([concept_importer.get_seen_versioned_object_id(_result)])) except Exception as ex: ERRBIT_LOGGER.log(ex) _result = {'__all__': str(ex)} @@ -1014,9 +1172,11 @@ def run(self): # pylint: disable=too-many-branches,too-many-statements,too-many continue if item_type == 'mapping': try: - mapping_importer = MappingImporter(item, self.user, self.update_if_exists, cache=self.cache) + mapping_importer = MappingImporter( + item, self.user, self.update_if_exists, cache=self.cache, isolated=self.isolated) _result = mapping_importer.delete() if action == 'delete' else mapping_importer.run() indexer.add(mapping_importer.instance) + self.seen_mapping_ids.update(compact([mapping_importer.get_seen_versioned_object_id(_result)])) except Exception as ex: ERRBIT_LOGGER.log(ex) _result = {'__all__': str(ex)} @@ -1078,13 +1238,18 @@ def make_result(self): 'detailed_summary': self.detailed_summary, 'report': self.report } + if self.isolated and not self.isolated_version: + self.result['seen'] = {'concepts': sorted(self.seen_concept_ids), 'mappings': sorted(self.seen_mapping_ids)} class BulkImportParallelRunner(BaseImporter): # pragma: no cover def __init__( - self, content, username, update_if_exists, parallel=None, self_task_id=None, index=None + self, content, username, update_if_exists, parallel=None, self_task_id=None, index=None, + isolated_version=False ): # pylint: disable=too-many-arguments super().__init__(content, username, update_if_exists, None, False) + self.isolated_version = isolated_version + self.isolated_version_results = {} self.start_time = time.time() self.self_task_id = self_task_id self.set_task() @@ -1106,6 +1271,9 @@ def __init__( self.populate_input_list() self.total = len(self.input_list) self.index = should_index_import(index, self.total) + version_line = Source.validate_import_versioning_modes(self.input_list, self.isolated_version) + if version_line: + IsolatedVersionCommit(version_line, self.user, self.update_if_exists).preflight() self.make_resource_distribution() self.make_parts() self.collect_concept_hierarchy_map() @@ -1157,22 +1325,8 @@ def make_parts(self): prev_line = line def collect_concept_hierarchy_map(self): - for line in self.input_list: - if line.get('type', '').lower() != 'concept': - continue - parent_urls = line.get('parent_concept_urls') or [] - concept_id = line.get('id') - owner = line.get('owner') - source = line.get('source') - if parent_urls and concept_id and owner and source: - owner_type = line.get('owner_type', '').lower() - owner_prefix = 'users' if owner_type in ['user', 'users'] else 'orgs' - # P2: normalize concept_id the same way ConceptImporter.parse() does, - # so the URI matches what was actually persisted in the database. - if not is_url_encoded_string(concept_id): - concept_id = encode_string(concept_id, safe='') - child_uri = f'/{owner_prefix}/{owner}/sources/{source}/concepts/{concept_id}/' - self.concept_hierarchy_map[child_uri] = parent_urls + self.concept_hierarchy_map = Concept.get_import_hierarchy_map( + self.input_list, include_parentless=self.isolated_version) @staticmethod def get_resource_id(resource): @@ -1256,6 +1410,40 @@ def get_overall_tasks_progress(self): def has_hierarchy_reconciliation_step(self): return bool(self.concept_hierarchy_map) + def reconcile_hierarchy(self): + # an isolated version's hierarchy is rebuilt only by its commit, inside the version's transaction + if self.isolated_version or self.hierarchy_reconciliation_done or not self.concept_hierarchy_map: + return + user = UserProfile.objects.filter(username=self.username).first() + Concept.reconcile_import_hierarchy(self.concept_hierarchy_map, user) + self.hierarchy_reconciliation_done = True + self.notify_progress() + + def commit_isolated_version(self, version_line): + commit = IsolatedVersionCommit( + version_line, UserProfile.objects.filter(username=self.username).first(), self.update_if_exists, + hierarchy_map=self.concept_hierarchy_map) + for task in self.tasks: + try: + task_result = task.result + except: # pylint: disable=bare-except + continue + if isinstance(task_result, dict): + commit.add_runner_result(task_result) + result = commit.run() + # the commit owns the hierarchy, so a rejected version leaves it untouched + self.hierarchy_reconciliation_done = True + self.isolated_version_results.setdefault(self.get_result_key(result), []).append( + {**version_line, 'errors': result} if isinstance(result, dict) else version_line) + self.notify_progress() + + @staticmethod + def get_result_key(result): + if isinstance(result, dict): + return 'failed' + return {None: 'exists', False: 'invalid', CREATED: 'created', PERMISSION_DENIED: 'permission_denied'}.get( + result, 'others') + def get_total_progress_target(self): return self.total + int(self.has_hierarchy_reconciliation_step()) @@ -1289,6 +1477,9 @@ def run(self): print("***************") while len(self.parts) > 0: part_list = self.parts.popleft() + if part_list and self.isolated_version and get(part_list, '0.type', '').lower() == 'source version': + self.commit_isolated_version(part_list[0]) + part_list = part_list[1:] if part_list: part_type = get(part_list, '0.type', '').lower() if part_type: @@ -1302,33 +1493,7 @@ def run(self): self.resource_wise_time[part_type] += round(time.time() - start_time, 4) self.notify_progress() - if self.concept_hierarchy_map: - # P1: restrict reconciliation to concepts the importing user can actually edit, - # mirroring the has_edit_access guard in ConceptImporter.process(). This excludes - # PERMISSION_DENIED rows and prevents hierarchy changes on sources the user does - # not own, even when those concepts already exist in the database. - user = UserProfile.objects.filter(username=self.username).first() - accessible_uris = set( - concept.uri - for concept in Concept.objects.filter( - uri__in=self.concept_hierarchy_map.keys(), id=F('versioned_object_id') - ).select_related('parent') - if concept.parent.has_edit_access(user) - ) - viewable_parent_uris = set(Concept.get_viewable_parent_uris( - list({uri for uris in self.concept_hierarchy_map.values() for uri in uris}), user)) - inverted = {} - for child_uri, parent_uris in self.concept_hierarchy_map.items(): - if child_uri not in accessible_uris: - continue - for parent_uri in [uri for uri in parent_uris if uri in viewable_parent_uris]: - if parent_uri not in inverted: - inverted[parent_uri] = [] - inverted[parent_uri].append(child_uri) - if inverted: - make_hierarchy(inverted) - self.hierarchy_reconciliation_done = True - self.notify_progress() + self.reconcile_hierarchy() post_import_update_resource_counts.apply_async(queue='default', permanent=False) @@ -1396,6 +1561,9 @@ def json_result(self): except: # pylint: disable=bare-except pass + for key, items in self.isolated_version_results.items(): + total_result[key] += items + total_result['processed'] += len(items) total_result['start_time'] = self.start_time_formatted total_result['elapsed_seconds'] = self.elapsed_seconds total_result['child_resource_time_distribution'] = self.resource_wise_time @@ -1424,7 +1592,7 @@ def queue_tasks(self, part_list, is_child): chunked_lists = [part_list] if has_delete_action else compact( self.chunker_list(part_list, self.parallel, is_child)) jobs = group( - bulk_import_parts_inline.s(_list, self.username, self.update_if_exists, self.index) + bulk_import_parts_inline.s(_list, self.username, self.update_if_exists, self.index, self.isolated_version) for _list in chunked_lists ) group_result = jobs.apply_async(queue='concurrent') diff --git a/core/importers/tests.py b/core/importers/tests.py index 4b032094b..35d8207b2 100644 --- a/core/importers/tests.py +++ b/core/importers/tests.py @@ -29,25 +29,29 @@ from core.collections.tests.factories import OrganizationCollectionFactory from core.common.constants import OPENMRS_VALIDATION_SCHEMA, DEPRECATED_API_HEADER, ACCESS_TYPE_NONE from core.common.tasks import post_import_update_resource_counts, bulk_import_parts_inline, bulk_import_inline, \ - bulk_import, batch_index_resources, bulk_import_parallel_inline, bulk_import_new, index_concepts_mapped_codes + bulk_import, batch_index_resources, bulk_import_parallel_inline, bulk_import_new, index_concepts_mapped_codes, \ + index_dropped_source_children from core.common.tests import OCLAPITestCase, OCLTestCase from core.common.utils import decode_string, startswith_temp_version, get_bulk_import_celery_once_lock_key from core.concepts.documents import ConceptDocument -from core.concepts.models import Concept +from core.concepts.models import Concept, HierarchicalConcepts from core.concepts.tests.factories import ConceptFactory from core.importers.importer import ImporterSubtask, ImportTask, ImportTaskSummary, Importer, ResourceImporter from core.importers.input_parsers import ImportContentParser from core.importers.models import BulkImport, BulkImportInline, BulkImportParallelRunner, \ CREATED, UPDATED, DELETED, PERMISSION_DENIED, UNCHANGED, FAILED, NOT_FOUND, \ BaseImporter, BaseResourceImporter, OrganizationImporter, SourceImporter, SourceVersionImporter, \ - CollectionImporter, CollectionVersionImporter, ConceptImporter, MappingImporter, ReferenceImporter, ImportIndexer + CollectionImporter, CollectionVersionImporter, ConceptImporter, MappingImporter, ReferenceImporter, ImportIndexer, \ + IsolatedVersionCommit from core.importers.views import csv_file_data_to_input_list, ImportRetrieveDestroyMixin from core.mappings.documents import MappingDocument from core.mappings.models import Mapping from core.mappings.tests.factories import MappingFactory from core.orgs.models import Organization from core.orgs.tests.factories import OrganizationFactory -from core.sources.constants import AUTO_ID_SEQUENTIAL, AUTO_ID_UUID +from core.sources.constants import AUTO_ID_SEQUENTIAL, AUTO_ID_UUID, ISOLATED_VERSION_CONTENT_FAILED, \ + ISOLATED_VERSION_EXISTS, ISOLATED_VERSION_LINE_INVALID, ISOLATED_VERSION_NO_EDIT_ACCESS, \ + VERSIONING_MODE_NEEDS_ISOLATED_VERSION, INVALID_VERSIONING_MODE from core.sources.models import Source from core.sources.tests.factories import OrganizationSourceFactory from core.tasks.models import Task @@ -2892,7 +2896,7 @@ def group_of(signatures): bulk_import_parts_inline_mock.reset_mock() importer = BulkImportParallelRunner(json.dumps(concept), 'ocladmin', True, None, 'runner-task-id', index) importer.queue_tasks([concept], True) - bulk_import_parts_inline_mock.s.assert_called_once_with([concept], 'ocladmin', True, index) + bulk_import_parts_inline_mock.s.assert_called_once_with([concept], 'ocladmin', True, index, False) def test_celery_once_lock_key_matches_the_task_key(self): """Revoking an import clears its celery_once lock under the key get_bulk_import_celery_once_lock_key rebuilds @@ -3186,9 +3190,9 @@ def test_collect_concept_hierarchy_map_ignores_non_concepts(self): @patch('core.importers.models.post_import_update_resource_counts.apply_async', Mock()) @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive', Mock()) @patch('core.importers.models.BulkImportParallelRunner.queue_tasks', Mock()) - @patch('core.importers.models.make_hierarchy') - def test_run_calls_make_hierarchy_with_inverted_map(self, make_hierarchy_mock): - """After all chunks complete, make_hierarchy receives the inverted {parent_uri: [child_uris]} map.""" + @patch('core.concepts.models.ImportHierarchy.apply') + def test_run_reconciles_the_hierarchy_once(self, apply_mock): + """After all chunks complete, the hierarchy is applied once, with the child linked to its parent.""" org = OrganizationFactory(mnemonic='TestOrg') source = OrganizationSourceFactory(organization=org, mnemonic='TestSource', version='HEAD') parent = ConceptFactory(parent=source, mnemonic='ParentConcept') @@ -3207,21 +3211,21 @@ def test_run_calls_make_hierarchy_with_inverted_map(self, make_hierarchy_mock): }), ]) - importer = BulkImportParallelRunner(content, 'ocladmin', True) - importer.run() + with patch('core.concepts.models.Concept.get_versioned_objects_by_uri', + wraps=Concept.get_versioned_objects_by_uri) as resolve_mock: + importer = BulkImportParallelRunner(content, 'ocladmin', True) + importer.run() - make_hierarchy_mock.assert_called_once() + apply_mock.assert_called_once_with(False) + resolve_mock.assert_called_once() self.assertTrue(importer.hierarchy_reconciliation_done) - inverted = make_hierarchy_mock.call_args[0][0] - self.assertIn(parent.uri, inverted) - self.assertIn(child.uri, inverted[parent.uri]) @patch('core.importers.models.post_import_update_resource_counts.apply_async', Mock()) @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive', Mock()) @patch('core.importers.models.BulkImportParallelRunner.queue_tasks', Mock()) - @patch('core.importers.models.make_hierarchy') - def test_run_skips_make_hierarchy_when_no_hierarchy(self, make_hierarchy_mock): - """make_hierarchy is not called when no concept in the import has parent_concept_urls.""" + @patch('core.concepts.models.ImportHierarchy.apply') + def test_run_skips_hierarchy_when_no_hierarchy(self, apply_mock): + """The hierarchy isn't applied when no concept in the import has parent_concept_urls.""" content = '\n'.join([ json.dumps({ "type": "Concept", "id": "StandaloneConcept", @@ -3231,7 +3235,7 @@ def test_run_skips_make_hierarchy_when_no_hierarchy(self, make_hierarchy_mock): importer = BulkImportParallelRunner(content, 'ocladmin', True) importer.run() - make_hierarchy_mock.assert_not_called() + apply_mock.assert_not_called() @patch('core.importers.models.post_import_update_resource_counts.apply_async', Mock()) @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive', Mock()) @@ -3301,10 +3305,9 @@ def test_run_hierarchy_parent_before_child(self): @patch('core.importers.models.post_import_update_resource_counts.apply_async', Mock()) @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive', Mock()) @patch('core.importers.models.BulkImportParallelRunner.queue_tasks', Mock()) - @patch('core.importers.models.make_hierarchy') - def test_run_excludes_inaccessible_concepts(self, make_hierarchy_mock): + def test_run_excludes_inaccessible_concepts(self): """ - Concepts from sources the importing user cannot edit must be excluded from make_hierarchy, + Concepts from sources the importing user cannot edit must be excluded from the hierarchy, even when they exist in the database. This prevents hierarchy changes on foreign sources that were denied during import (P1 security fix). """ @@ -3343,17 +3346,9 @@ def test_run_excludes_inaccessible_concepts(self, make_hierarchy_mock): importer = BulkImportParallelRunner(content, importing_user.username, True) importer.run() - make_hierarchy_mock.assert_called_once() - inverted = make_hierarchy_mock.call_args[0][0] - - # OrgA child must be linked to its parent - self.assertIn(parent_a.uri, inverted) - self.assertIn(child_a.uri, inverted[parent_a.uri]) - - # OrgB child must NOT appear — user has no access to SourceB - self.assertNotIn(parent_b.uri, inverted) - all_children = [uri for uris in inverted.values() for uri in uris] - self.assertNotIn(child_b.uri, all_children) + self.assertEqual( + set(child_a.parent_concepts.values_list('id', flat=True)), {parent_a.get_latest_version().id}) + self.assertFalse(child_b.parent_concepts.exists()) class BulkImportViewTest(OCLAPITestCase): @@ -4178,7 +4173,7 @@ def test_bulk_import_parts_inline(self, bulk_import_inline_mock): bulk_import_parts_inline([1, 2], 'username', True) # pylint: disable=no-value-for-parameter bulk_import_inline_mock.assert_called_once_with( content=None, username='username', update_if_exists=True, input_list=[1, 2], - self_task_id=ANY, skip_hierarchy_tasks=True, index=True + self_task_id=ANY, skip_hierarchy_tasks=True, index=True, isolated=False ) bulk_import_inline_mock().run.assert_called_once() @@ -4186,7 +4181,7 @@ def test_bulk_import_parts_inline(self, bulk_import_inline_mock): bulk_import_parts_inline([1, 2], 'username', True, False) # pylint: disable=no-value-for-parameter bulk_import_inline_mock.assert_called_once_with( content=None, username='username', update_if_exists=True, input_list=[1, 2], - self_task_id=ANY, skip_hierarchy_tasks=True, index=False + self_task_id=ANY, skip_hierarchy_tasks=True, index=False, isolated=False ) @patch('core.importers.models.BulkImportInline') @@ -4195,14 +4190,14 @@ def test_bulk_import_inline(self, bulk_import_inline_mock): bulk_import_inline([1, 2], 'username', True) bulk_import_inline_mock.assert_called_once_with( # the import's size decides - content=[1, 2], username='username', update_if_exists=True, index=None + content=[1, 2], username='username', update_if_exists=True, index=None, isolated_version=False ) bulk_import_inline_mock().run.assert_called_once() bulk_import_inline_mock.reset_mock() bulk_import_inline([1, 2], 'username', True, False) bulk_import_inline_mock.assert_called_once_with( - content=[1, 2], username='username', update_if_exists=True, index=False + content=[1, 2], username='username', update_if_exists=True, index=False, isolated_version=False ) @patch('core.importers.models.BulkImportParallelRunner') @@ -4210,7 +4205,7 @@ def test_bulk_import_parallel_inline(self, runner_mock): bulk_import_parallel_inline('content', 'username', True, 2, False) # pylint: disable=no-value-for-parameter runner_mock.assert_called_once_with( content='content', username='username', update_if_exists=True, parallel=2, self_task_id=ANY, - index=False + index=False, isolated_version=False ) runner_mock().run.assert_called_once() @@ -6050,3 +6045,528 @@ def test_deprecated_inline_route_limited_too(self, task_mock): response = self.post(self.preview_user, {'file': self.json_file(600)}, url='/importers/bulk-import-inline/') self.assertEqual(response.status_code, 403) self.assertEqual(response.data['error_code'], 'imports_file_size_limit_reached') + + +@patch('core.sources.models.index_source_mappings', Mock(__name__='index_source_mappings')) +@patch('core.sources.models.index_source_concepts', Mock(__name__='index_source_concepts')) +@patch('core.importers.models.index_concepts_mapped_codes', Mock()) +@patch('core.importers.models.batch_index_resources', Mock(__name__='batch_index_resources')) +class IsolatedVersionImportTest(OCLTestCase): + def setUp(self): + super().setUp() + self.org = OrganizationFactory(mnemonic='Regen') + self.source_line = { + 'type': 'Source', 'id': 'LOINC', 'name': 'LOINC', 'owner_type': 'Organization', 'owner': 'Regen', + 'default_locale': 'en', 'supported_locales': 'en' + } + + @staticmethod + def concept(mnemonic, parents=None, **kwargs): + return { + 'type': 'Concept', 'id': mnemonic, 'concept_class': 'LOINC', 'datatype': 'N/A', 'owner': 'Regen', + 'owner_type': 'Organization', 'source': 'LOINC', + 'names': [{'name': f'{mnemonic} name', 'locale': 'en', 'locale_preferred': True}], + 'parent_concept_urls': [f'/orgs/Regen/sources/LOINC/concepts/{parent}/' for parent in parents or []], + **kwargs + } + + @staticmethod + def mapping(mnemonic, from_concept, to_concept): + return { + 'type': 'Mapping', 'id': mnemonic, 'map_type': 'SAME-AS', 'owner': 'Regen', 'owner_type': 'Organization', + 'source': 'LOINC', 'from_concept_url': f'/orgs/Regen/sources/LOINC/concepts/{from_concept}/', + 'to_concept_url': f'/orgs/Regen/sources/LOINC/concepts/{to_concept}/' + } + + @staticmethod + def version(version, **kwargs): + return { + 'type': 'Source Version', 'id': version, 'source': 'LOINC', 'owner': 'Regen', 'owner_type': 'Organization', + 'description': f'LOINC {version}', 'released': True, **kwargs + } + + def run_import(self, lines, isolated_version=True): + importer = BulkImportInline( + '\n'.join(json.dumps(line) for line in [self.source_line, *lines]), 'ocladmin', True, + isolated_version=isolated_version) + with self.captureOnCommitCallbacks(execute=True): + importer.run() + return importer + + @staticmethod + def members(version): + return set(Source.objects.get(mnemonic='LOINC', version=version).concepts.values_list('mnemonic', flat=True)) + + def test_isolated_import_drops_unseen_concepts(self): + self.run_import([self.concept('A'), self.concept('B', extras={'x': 1}), self.version('v1')]) + importer = self.run_import([self.concept('B'), self.concept('C'), self.version('v2')]) + + self.assertEqual(len(importer.failed), 0) + self.assertEqual(self.members('v1'), {'A', 'B'}) + self.assertEqual(self.members('v2'), {'B', 'C'}) + self.assertEqual(Source.objects.get(mnemonic='LOINC', version='v2').versioning_mode, 'isolated') + head = Source.objects.get(mnemonic='LOINC', version='HEAD') + self.assertEqual(head.versioning_mode, 'carry_forward') + self.assertEqual(set(head.concepts.filter( + id=F('versioned_object_id'), is_active=True).values_list('mnemonic', flat=True)), {'B', 'C'}) + concept_a = Concept.objects.get(mnemonic='A', id=F('versioned_object_id')) + self.assertFalse(concept_a.is_active) + self.assertTrue(concept_a.get_latest_version().is_latest_version) + self.assertTrue(concept_a.get_latest_version().sources.filter(version='v1').exists()) + self.assertEqual(Concept.objects.get(mnemonic='B', is_latest_version=True).extras, {}) + self.assertEqual(head.active_concepts, 2) + + def test_isolated_import_drops_and_revives_mappings(self): + concepts = [self.concept('A'), self.concept('B')] + self.run_import([*concepts, self.mapping('M1', 'A', 'B'), self.mapping('M2', 'B', 'A'), self.version('v1')]) + self.run_import([*concepts, self.mapping('M1', 'A', 'B'), self.version('v2')]) + self.run_import([*concepts, self.mapping('M1', 'A', 'B'), self.mapping('M2', 'B', 'A'), self.version('v3')]) + + def mapping_members(version): + return set(Source.objects.get(mnemonic='LOINC', version=version).mappings.values_list( + 'mnemonic', flat=True)) + + self.assertEqual(mapping_members('v1'), {'M1', 'M2'}) + self.assertEqual(mapping_members('v2'), {'M1'}) + self.assertEqual(mapping_members('v3'), {'M1', 'M2'}) + self.assertTrue(Mapping.objects.get(mnemonic='M2', id=F('versioned_object_id')).is_active) + + def test_isolated_import_revives_dropped_concept(self): + self.run_import([self.concept('A'), self.concept('B'), self.version('v1')]) + concept_a_v1 = Concept.objects.get(mnemonic='A', is_latest_version=True) + self.run_import([self.concept('B'), self.version('v2')]) + self.run_import([self.concept('A'), self.concept('B'), self.version('v3')]) + + self.assertEqual(self.members('v3'), {'A', 'B'}) + self.assertEqual(self.members('v2'), {'B'}) + concept_a = Concept.objects.get(mnemonic='A', id=F('versioned_object_id')) + self.assertTrue(concept_a.is_active) + self.assertEqual(concept_a.id, concept_a_v1.versioned_object_id) + self.assertTrue(Source.objects.get(mnemonic='LOINC', version='v1').concepts.filter( + id=concept_a_v1.id).exists()) + + def test_isolated_import_moves_child_without_leaking_old_parent(self): + self.run_import([ + self.concept('P1'), self.concept('P2'), self.concept('C', parents=['P1']), self.version('v1')]) + self.run_import([ + self.concept('P1'), self.concept('P2'), self.concept('C', parents=['P2']), self.version('v2')]) + + concept = Concept.objects.get(mnemonic='C', id=F('versioned_object_id')) + v1 = Source.objects.get(mnemonic='LOINC', version='v1') + v2 = Source.objects.get(mnemonic='LOINC', version='v2') + self.assertEqual( + set(concept.get_version_hierarchy_queryset('parent_concepts', v1).values_list('mnemonic', flat=True)), + {'P1'}) + self.assertEqual( + set(concept.get_version_hierarchy_queryset('parent_concepts', v2).values_list('mnemonic', flat=True)), + {'P2'}) + parent_one = Concept.objects.get(mnemonic='P1', id=F('versioned_object_id')) + self.assertFalse(parent_one.get_version_hierarchy_queryset('child_concepts', v2).exists()) + self.assertEqual( + set(parent_one.get_version_hierarchy_queryset('child_concepts', v1).values_list('mnemonic', flat=True)), + {'C'}) + self.assertEqual(concept.parent_concept_urls, ['/orgs/Regen/sources/LOINC/concepts/P2/']) + + def test_isolated_import_fails_when_hierarchy_root_is_dropped(self): + self.run_import([self.concept('ROOT'), self.concept('A', parents=['ROOT']), self.version('v1')]) + head = Source.objects.get(mnemonic='LOINC', version='HEAD') + head.hierarchy_root = Concept.objects.get(mnemonic='ROOT', id=F('versioned_object_id')) + head.save() + + importer = self.run_import([self.concept('A'), self.version('v2')]) + + self.assertEqual(len(importer.failed), 1) + self.assertIn('hierarchy_root', importer.failed[0]['errors']) + self.assertFalse(Source.objects.filter(mnemonic='LOINC', version='v2').exists()) + self.assertTrue(Concept.objects.get(mnemonic='ROOT', id=F('versioned_object_id')).is_active) + + def test_isolated_import_skips_version_when_content_fails(self): + self.run_import([self.concept('A'), self.concept('B'), self.version('v1')]) + invalid = self.concept('C') + invalid.pop('concept_class') + + importer = self.run_import([self.concept('A'), invalid, self.version('v2')]) + + self.assertEqual(len(importer.invalid), 1) + self.assertEqual(importer.failed[0]['errors'], {'__all__': [ISOLATED_VERSION_CONTENT_FAILED]}) + self.assertFalse(Source.objects.filter(mnemonic='LOINC', version='v2').exists()) + self.assertTrue(Concept.objects.get(mnemonic='B', id=F('versioned_object_id')).is_active) + + def test_isolated_import_needs_one_source_version_after_content(self): + for lines in ( + [self.concept('A')], + [self.concept('A'), self.version('v1'), self.version('v2')], + [self.version('v1'), self.concept('A')], + ): + with self.assertRaises(DjangoValidationError): + self.run_import(lines) + + def test_carry_forward_import_keeps_unseen_concepts(self): + self.run_import([self.concept('A'), self.concept('B', extras={'x': 1}), self.version('v1')], False) + self.run_import([self.concept('B'), self.version('v2')], False) + + self.assertEqual(self.members('v2'), {'A', 'B'}) + self.assertEqual(Concept.objects.get(mnemonic='B', is_latest_version=True).extras, {'x': 1}) + self.assertEqual(Source.objects.get(mnemonic='LOINC', version='v2').versioning_mode, 'carry_forward') + + @staticmethod + def head_concepts(): + return set(Source.objects.get(mnemonic='LOINC', version='HEAD').concepts.filter( + id=F('versioned_object_id'), is_active=True).values_list('mnemonic', flat=True)) + + @staticmethod + def head_parents(mnemonic): + return set(HierarchicalConcepts.objects.filter( + child__mnemonic=mnemonic, child_id=F('child__versioned_object_id')).values_list( + 'parent__mnemonic', flat=True)) + + @patch('core.sources.models.Source.batch_index') + def test_rejected_version_leaves_head_unchanged(self, batch_index_mock): + self.run_import([ + self.concept('P1'), self.concept('P2'), self.concept('C', parents=['P1']), self.concept('A'), + self.version('v1')]) + batch_index_mock.reset_mock() + + with patch('core.sources.models.Source.persist_new_version', return_value={'version': ['Not saved']}): + importer = self.run_import([ + self.concept('P1'), self.concept('P2'), self.concept('C', parents=['P2']), self.version('v2')]) + + self.assertEqual(importer.failed[0]['errors'], {'version': ['Not saved']}) + self.assertFalse(Source.objects.filter(mnemonic='LOINC', version='v2').exists()) + self.assertEqual(self.head_concepts(), {'P1', 'P2', 'C', 'A'}) + batch_index_mock.assert_not_called() + + def test_existing_version_changes_nothing(self): + self.run_import([self.concept('A'), self.concept('B'), self.version('v1')]) + + with self.assertRaises(DjangoValidationError) as context: + self.run_import([self.concept('B'), self.concept('NEW'), self.version('v1')]) + + self.assertEqual(context.exception.message_dict, {'__all__': [ISOLATED_VERSION_EXISTS]}) + self.assertFalse(Concept.objects.filter(mnemonic='NEW').exists()) + self.assertEqual(self.head_concepts(), {'A', 'B'}) + + def test_preflight_rejects_a_version_before_content_imports(self): + self.run_import([self.concept('A'), self.version('v1')]) + user = UserProfileFactory() + for line, message in ( + ({'source': 'LOINC', 'owner': 'Regen', 'owner_type': 'Organization'}, ISOLATED_VERSION_LINE_INVALID), + (self.version('v2'), ISOLATED_VERSION_NO_EDIT_ACCESS), + ): + with self.assertRaises(DjangoValidationError) as context: + IsolatedVersionCommit(line, user).preflight() + self.assertEqual(context.exception.message_dict, {'__all__': [message]}) + self.assertIsNone(IsolatedVersionCommit( + {**self.version('v2'), 'source': 'NEW'}, UserProfile.objects.get(username='ocladmin')).preflight()) + + @patch('core.sources.models.Source.batch_index', side_effect=Exception('ES down')) + def test_failed_search_update_does_not_undo_a_committed_version(self, batch_index_mock): + self.run_import([self.concept('A'), self.concept('B'), self.version('v1')]) + + importer = self.run_import([self.concept('B'), self.version('v2')]) + + self.assertTrue(batch_index_mock.called) + self.assertEqual(len(importer.created), 1) + self.assertEqual(self.members('v2'), {'B'}) + self.assertEqual(self.head_concepts(), {'B'}) + + @patch('core.sources.models.index_dropped_source_children.apply_async') + def test_dropped_codes_are_repaired_by_a_queued_task(self, apply_async_mock): + self.run_import([self.concept('A'), self.concept('B'), self.version('v1')]) + head = Source.objects.get(mnemonic='LOINC', version='HEAD') + dropped = Concept.objects.get(mnemonic='A', id=F('versioned_object_id')) + kept = Concept.objects.get(mnemonic='B', id=F('versioned_object_id')) + + with override_settings(TEST_MODE=False), self.captureOnCommitCallbacks(execute=True): + head.drop_unseen_children([kept.id], []) + + task = Task.objects.get(name='index_dropped_source_children') + apply_async_mock.assert_called_once_with( + (head.id, [dropped.id, dropped.get_latest_version().id], []), task_id=task.id, queue='indexing') + + def test_dropped_codes_repair_task_retries_and_reindexes(self): + self.assertEqual(index_dropped_source_children.max_retries, 5) + self.assertIn(Exception, index_dropped_source_children.autoretry_for) + self.run_import([self.concept('A'), self.version('v1')]) + head = Source.objects.get(mnemonic='LOINC', version='HEAD') + + with patch('core.sources.models.Source.index_dropped_children') as index_mock: + index_dropped_source_children(head.id, [1], [2]) + index_dropped_source_children(0, [1], [2]) + + index_mock.assert_called_once_with([1], [2]) + + def test_hierarchy_root_drop_rolls_back_the_hierarchy(self): + self.run_import([ + self.concept('ROOT'), self.concept('P1'), self.concept('C', parents=['P1']), self.version('v1')]) + head = Source.objects.get(mnemonic='LOINC', version='HEAD') + head.hierarchy_root = Concept.objects.get(mnemonic='ROOT', id=F('versioned_object_id')) + head.save() + + importer = self.run_import([self.concept('P1'), self.concept('C'), self.version('v2')]) + + self.assertIn('hierarchy_root', importer.failed[0]['errors']) + self.assertEqual(self.head_concepts(), {'ROOT', 'P1', 'C'}) + + def test_commit_rolls_back_the_hierarchy_with_the_version(self): + self.run_import([ + self.concept('P1'), self.concept('P2'), self.concept('C', parents=['P1']), self.version('v1')]) + version_line = self.version('v2') + version_line.pop('type') + prefix = '/orgs/Regen/sources/LOINC/concepts/' + + def commit(): + with self.captureOnCommitCallbacks(execute=True): + return IsolatedVersionCommit( + version_line, UserProfile.objects.get(username='ocladmin'), + hierarchy_map={f'{prefix}C/': [f'{prefix}P2/'], f'{prefix}P1/': [], f'{prefix}P2/': []}, + seen_concept_ids=Concept.objects.filter( + mnemonic__in=['P1', 'P2', 'C'], id=F('versioned_object_id')).values_list('id', flat=True) + ).run() + + with patch('core.sources.models.Source.persist_new_version', return_value={'version': ['Not saved']}): + self.assertEqual(commit(), {'version': ['Not saved']}) + self.assertEqual(self.head_parents('C'), {'P1'}) + + self.assertEqual(commit(), CREATED) + self.assertEqual(self.head_parents('C'), {'P2'}) + + def test_versioning_mode_is_resolved_before_anything_imports(self): + for lines, isolated_version, message in ( + ([self.concept('A'), self.version('v1', versioning_mode='carry_forward')], True, + VERSIONING_MODE_NEEDS_ISOLATED_VERSION), + ([self.concept('A'), self.version('v1', versioning_mode='isolated')], False, + VERSIONING_MODE_NEEDS_ISOLATED_VERSION), + ([self.concept('A'), self.version('v1', versioning_mode='other')], False, INVALID_VERSIONING_MODE), + ): + with self.assertRaises(DjangoValidationError) as context: + self.run_import(lines, isolated_version) + self.assertEqual(context.exception.message_dict, {'versioning_mode': [message]}) + self.assertFalse(Concept.objects.filter(mnemonic='A').exists()) + + def test_version_line_mode_matching_the_import_is_accepted(self): + self.run_import([self.concept('A'), self.version('v1', versioning_mode='isolated')]) + self.run_import([self.concept('B'), self.version('v2', versioning_mode='carry_forward')], False) + + self.assertEqual(Source.objects.get(mnemonic='LOINC', version='v1').versioning_mode, 'isolated') + self.assertEqual(Source.objects.get(mnemonic='LOINC', version='v2').versioning_mode, 'carry_forward') + self.assertEqual(self.members('v2'), {'A', 'B'}) + + def test_plain_version_importer_rejects_isolated_mode(self): + self.run_import([self.concept('A'), self.version('v1')]) + version_line = self.version('v2', versioning_mode='isolated') + version_line.pop('type') + + result = SourceVersionImporter(version_line, UserProfile.objects.get(username='ocladmin')).run() + + self.assertEqual(result, {'versioning_mode': [VERSIONING_MODE_NEEDS_ISOLATED_VERSION]}) + self.assertFalse(Source.objects.filter(mnemonic='LOINC', version='v2').exists()) + + def test_commit_needs_edit_access_to_head(self): + self.run_import([self.concept('A'), self.concept('B'), self.version('v1')]) + version_line = self.version('v2') + version_line.pop('type') + + result = IsolatedVersionCommit( + version_line, UserProfileFactory(), seen_concept_ids=[ + Concept.objects.get(mnemonic='B', id=F('versioned_object_id')).id]).run() + + self.assertEqual(result, PERMISSION_DENIED) + self.assertEqual(self.head_concepts(), {'A', 'B'}) + + def test_isolated_version_merges_metadata_into_head(self): + self.run_import([self.concept('A'), self.version('v1')]) + head = Source.objects.get(mnemonic='LOINC', version='HEAD') + head.website = 'https://loinc.org' + head.publisher = 'Old' + head.save() + + self.run_import([self.concept('A'), self.version('v2', publisher='Regenstrief', description='2.82')]) + + head.refresh_from_db() + version = Source.objects.get(mnemonic='LOINC', version='v2') + self.assertEqual(head.publisher, 'Regenstrief') + self.assertEqual(head.description, '2.82') + self.assertEqual(head.website, 'https://loinc.org') + self.assertEqual(version.publisher, 'Regenstrief') + self.assertEqual(version.website, 'https://loinc.org') + self.assertEqual(head.name, 'LOINC') + + +@patch('core.sources.models.index_source_mappings', Mock(__name__='index_source_mappings')) +@patch('core.sources.models.index_source_concepts', Mock(__name__='index_source_concepts')) +@patch('core.importers.models.index_concepts_mapped_codes', Mock()) +@patch('core.importers.models.batch_index_resources', Mock(__name__='batch_index_resources')) +class IsolatedVersionAPITest(OCLAPITestCase): + def setUp(self): + super().setUp() + self.token = UserProfile.objects.get(username='ocladmin').get_token() + OrganizationFactory(mnemonic='Regen') + + def run_import(self, lines): + with self.captureOnCommitCallbacks(execute=True): + BulkImportInline( + '\n'.join(json.dumps(line) for line in [ + {'type': 'Source', 'id': 'LOINC', 'name': 'LOINC', 'owner_type': 'Organization', 'owner': 'Regen', + 'default_locale': 'en', 'supported_locales': 'en'}, *lines]), + 'ocladmin', True, isolated_version=True).run() + + def test_version_hierarchy_endpoints(self): + concept = IsolatedVersionImportTest.concept + version = IsolatedVersionImportTest.version + self.run_import([concept('P1'), concept('P2'), concept('C', parents=['P1']), version('v1')]) + self.run_import([concept('P1'), concept('P2'), concept('C', parents=['P2']), version('v2')]) + + for repo_version, parent in (('v1', 'P1'), ('v2', 'P2')): + response = self.client.get( + f'/orgs/Regen/sources/LOINC/{repo_version}/concepts/C/parents/', + HTTP_AUTHORIZATION='Token ' + self.token) + self.assertEqual(response.status_code, 200) + self.assertEqual([item['id'] for item in response.data], [parent]) + response = self.client.get( + '/orgs/Regen/sources/LOINC/v2/concepts/P1/children/', HTTP_AUTHORIZATION='Token ' + self.token) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data, []) + response = self.client.get( + '/orgs/Regen/sources/LOINC/concepts/C/parents/', HTTP_AUTHORIZATION='Token ' + self.token) + self.assertEqual([item['id'] for item in response.data], ['P2']) + + def test_version_create_cannot_set_versioning_mode(self): + self.run_import([IsolatedVersionImportTest.concept('A'), IsolatedVersionImportTest.version('v1')]) + response = self.client.post( + '/orgs/Regen/sources/LOINC/versions/', + {'id': 'v2', 'description': 'v2', 'versioning_mode': 'isolated'}, + HTTP_AUTHORIZATION='Token ' + self.token, format='json') + self.assertEqual(response.status_code, 201) + self.assertEqual(response.data['versioning_mode'], 'carry_forward') + + @patch('core.importers.views.queue_bulk_import') + def test_import_passes_isolated_version(self, queue_bulk_import_mock): + queue_bulk_import_mock.return_value = Task.new(queue='default', username='ocladmin', name='bulk_import') + response = self.client.post( + '/importers/bulk-import/?update_if_exists=true&isolated_version=true', {'data': '{"type": "Concept"}'}, + HTTP_AUTHORIZATION='Token ' + self.token, format='json') + self.assertEqual(response.status_code, 202) + self.assertTrue(queue_bulk_import_mock.call_args.kwargs['isolated_version']) + + response = self.client.post( + '/importers/bulk-import/?isolated_version=maybe', {'data': '{"type": "Concept"}'}, + HTTP_AUTHORIZATION='Token ' + self.token, format='json') + self.assertEqual(response.status_code, 400) + + response = self.client.post( + '/importers/bulk-import/?isolated_version=true', {'data': '{}', 'import_type': 'npm'}, + HTTP_AUTHORIZATION='Token ' + self.token, format='json') + self.assertEqual(response.status_code, 400) + + +@patch('core.sources.models.index_source_mappings', Mock(__name__='index_source_mappings')) +@patch('core.sources.models.index_source_concepts', Mock(__name__='index_source_concepts')) +@patch('core.importers.models.index_concepts_mapped_codes', Mock()) +@patch('core.importers.models.batch_index_resources', Mock(__name__='batch_index_resources')) +class IsolatedVersionZipImportTest(OCLTestCase): + def setUp(self): + super().setUp() + OrganizationFactory(mnemonic='Regen') + BulkImportInline(json.dumps({ + 'type': 'Source', 'id': 'LOINC', 'name': 'LOINC', 'owner_type': 'Organization', 'owner': 'Regen', + 'default_locale': 'en', 'supported_locales': 'en'}), 'ocladmin', True).run() + + def import_zip(self, concepts, mappings, version): + concept = IsolatedVersionImportTest.concept + directory = tempfile.mkdtemp() + path = os.path.join(directory, 'loinc.zip') + with ZipFile(path, 'w') as package: + package.writestr('concepts.json', '\n'.join(json.dumps(concept(*args)) for args in concepts)) + package.writestr('mappings.json', '\n'.join( + json.dumps(IsolatedVersionImportTest.mapping(*args)) for args in mappings)) + package.writestr('version.json', json.dumps(IsolatedVersionImportTest.version(version))) + results = [] + for resource_type, filepath, count in ( + ('Concept', 'concepts.json', len(concepts)), ('Mapping', 'mappings.json', len(mappings))): + if count: + results += ImporterSubtask( + path, 'ocladmin', 'Organization', 'Regen', resource_type, + [{'filepath': filepath, 'start_index': 0, 'end_index': count}], False, True).run() + with patch.object(ImporterSubtask, 'iter_previous_results', return_value=[results]), \ + self.captureOnCommitCallbacks(execute=True): + return ImporterSubtask( + path, 'ocladmin', 'Organization', 'Regen', 'Source Version', + [{'filepath': 'version.json', 'start_index': 0, 'end_index': 1}], False, True, True).run() + + def test_isolated_version_zip_import(self): + self.assertEqual(self.import_zip([('A',), ('B',)], [('M1', 'A', 'B')], 'v1'), [CREATED]) + self.assertEqual(self.import_zip([('B',), ('C', ['B'])], [], 'v2'), [CREATED]) + + self.assertEqual(IsolatedVersionImportTest.members('v1'), {'A', 'B'}) + self.assertEqual(IsolatedVersionImportTest.members('v2'), {'B', 'C'}) + self.assertEqual(Source.objects.get(mnemonic='LOINC', version='v2').mappings.count(), 0) + self.assertEqual(Source.objects.get(mnemonic='LOINC', version='v2').versioning_mode, 'isolated') + self.assertEqual( + Concept.objects.get(mnemonic='C', id=F('versioned_object_id')).parent_concept_urls, + ['/orgs/Regen/sources/LOINC/concepts/B/']) + + def test_isolated_version_zip_import_skips_version_when_content_fails(self): + self.import_zip([('A',), ('B',)], [], 'v1') + with patch.object(ImporterSubtask, 'iter_previous_results', return_value=[[CREATED], ['Failed to import']]): + result = ImporterSubtask( + '/dev/null', 'ocladmin', 'Organization', 'Regen', 'Source Version', [], False, True, True + ).commit_isolated_version(IsolatedVersionImportTest.version('v2')) + self.assertEqual(result, {'__all__': [ISOLATED_VERSION_CONTENT_FAILED]}) + + def test_zip_import_with_numeric_extras(self): + directory = tempfile.mkdtemp() + path = os.path.join(directory, 'loinc.zip') + with ZipFile(path, 'w') as package: + package.writestr('concepts.json', '\n'.join( + json.dumps(IsolatedVersionImportTest.concept(mnemonic, extras={'rank': 1.5, 'order': 2})) + for mnemonic in ('N1', 'N2', 'N3'))) + results = [] + for start_index, end_index in ((0, 1), (1, 3)): + results += ImporterSubtask( + path, 'ocladmin', 'Organization', 'Regen', 'Concept', + [{'filepath': 'concepts.json', 'start_index': start_index, 'end_index': end_index}], False).run() + + self.assertEqual(results, [CREATED, CREATED, CREATED]) + self.assertEqual( + Concept.objects.get(mnemonic='N3', id=F('versioned_object_id')).extras, {'rank': 1.5, 'order': 2}) + + def test_categorize_resources_keeps_only_source_version_lines(self): + importer = Importer('task-id', '/tmp/loinc.zip', 'ocladmin', 'Organization', 'Regen', isolated_version=True) + resources = {} + lines = [ + {'type': 'Source', 'id': 'LOINC', 'versioning_mode': 'carry_forward', 'extras': {'type': 'x'}}, + {**IsolatedVersionImportTest.version('v1'), 'versioning_mode': 'isolated', 'extras': {'a': [1]}}, + ] + importer.categorize_resources( + io.BytesIO('\n'.join(json.dumps(line) for line in lines).encode()), + '/tmp/loinc.zip', 'lines.json', ['Source', 'Source Version'], resources) + + self.assertEqual(importer.version_lines, [{ + **IsolatedVersionImportTest.version('v1'), 'versioning_mode': 'isolated', 'extras': {}}]) + self.assertEqual( + importer.validate_versioning(resources, ['Source', 'Source Version', 'Concept', 'Mapping']), + ['Source', 'Concept', 'Mapping', 'Source Version']) + + def test_validate_versioning_rejects_before_content_imports(self): + self.import_zip([('A',)], [], 'v1') + resources = {'Source Version': {'/tmp/loinc.zip/version.json': 1}} + for line, isolated_version, errors in ( + ({**IsolatedVersionImportTest.version('v2'), 'versioning_mode': 1}, True, + {'versioning_mode': [INVALID_VERSIONING_MODE]}), + ({**IsolatedVersionImportTest.version('v2'), 'versioning_mode': {}}, False, + {'versioning_mode': [INVALID_VERSIONING_MODE]}), + (IsolatedVersionImportTest.version('v1'), True, {'__all__': [ISOLATED_VERSION_EXISTS]}), + ): + importer = Importer( + 'task-id', '/tmp/loinc.zip', 'ocladmin', 'Organization', 'Regen', isolated_version=isolated_version) + importer.version_lines = [line] + with self.assertRaises(ValidationError) as context: + importer.validate_versioning(resources, ['Source Version', 'Concept', 'Mapping']) + self.assertEqual(context.exception.detail, errors) + + def test_isolated_version_resource_types(self): + self.assertEqual( + Importer.get_isolated_version_resource_types( + ['Source', 'Source Version', 'Concept', 'Mapping', 'Reference']), + ['Source', 'Concept', 'Mapping', 'Source Version', 'Reference']) diff --git a/core/importers/views.py b/core/importers/views.py index c535b721e..54ed79d04 100644 --- a/core/importers/views.py +++ b/core/importers/views.py @@ -24,10 +24,11 @@ from core.common.views import BaseAPIView from core.common.tasks import bulk_import_new from core.common.swagger_parameters import update_if_exists_param, task_param, result_param, username_param, \ - file_upload_param, file_url_param, parallel_threads_param, verbose_param, index_param + file_upload_param, file_url_param, parallel_threads_param, verbose_param, index_param, isolated_version_param from core.common.utils import queue_bulk_import, is_csv_file, get_truthy_values, get_queue_task_names, \ get_export_service -from core.importers.constants import ALREADY_QUEUED, INVALID_UPDATE_IF_EXISTS, NO_CONTENT_TO_IMPORT, INVALID_INDEX +from core.importers.constants import ALREADY_QUEUED, INVALID_UPDATE_IF_EXISTS, NO_CONTENT_TO_IMPORT, INVALID_INDEX, \ + INVALID_ISOLATED_VERSION, ISOLATED_VERSION_IMPORT_TYPE from core.importers.importer import Importer, ResourceImporter from core.importers.input_parsers import ImportContentParser from core.importers.limits import ( @@ -58,6 +59,13 @@ def get_index_param(request): return index == 'true', None +def get_isolated_version_param(request): + isolated_version = request.GET.get('isolated_version', 'false') + if isolated_version not in ['true', 'false']: + return None, Response({'exception': INVALID_ISOLATED_VERSION}, status=status.HTTP_400_BAD_REQUEST) + return isolated_version == 'true', None + + def import_response(request, import_queue, data, threads=None, inline=False, deprecated=False): # pylint: disable=too-many-arguments if not data: return Response({'exception': NO_CONTENT_TO_IMPORT}, status=status.HTTP_400_BAD_REQUEST) @@ -73,13 +81,18 @@ def import_response(request, import_queue, data, threads=None, inline=False, dep ) update_if_exists = update_if_exists == 'true' index, error_response = get_index_param(request) + if error_response: + return error_response + isolated_version, error_response = get_isolated_version_param(request) if error_response: return error_response data = data.decode('utf-8') if isinstance(data, bytes) else data task = None try: - task = queue_bulk_import(data, import_queue, username, update_if_exists, threads, inline, index=index) + task = queue_bulk_import( + data, import_queue, username, update_if_exists, threads, inline, index=index, + isolated_version=isolated_version) task.refresh_from_db() except AlreadyQueued: if task: @@ -180,7 +193,8 @@ def get_parsers(self): @swagger_auto_schema( manual_parameters=[ - update_if_exists_param, index_param, file_url_param, file_upload_param, parallel_threads_param + update_if_exists_param, index_param, isolated_version_param, file_url_param, file_upload_param, + parallel_threads_param ], deprecated=True ) @@ -259,15 +273,21 @@ def store_upload(file): @swagger_auto_schema( manual_parameters=[ - update_if_exists_param, index_param, file_url_param, file_upload_param, parallel_threads_param + update_if_exists_param, index_param, isolated_version_param, file_url_param, file_upload_param, + parallel_threads_param ], ) - def post(self, request, import_queue=None): # pylint: disable=too-many-locals + def post(self, request, import_queue=None): # pylint: disable=too-many-locals,too-many-return-statements,too-many-branches check_request_body_size(request) if 'import_type' in request.data: index, error_response = get_index_param(request) if error_response: return error_response + isolated_version, error_response = get_isolated_version_param(request) + if error_response: + return error_response + if isolated_version and request.data.get('import_type', 'npm') == 'npm': + return Response({'exception': ISOLATED_VERSION_IMPORT_TYPE}, status=status.HTTP_400_BAD_REQUEST) check_import_payload(request) owner_type = request.data.get('owner_type', 'user') owner = request.data.get('owner', self.request.user.username) @@ -299,7 +319,8 @@ def post(self, request, import_queue=None): # pylint: disable=too-many-locals task = get_queue_task_names(sanitize_import_queue(request.user, import_queue), self.request.user.username) new_task = bulk_import_new.apply_async( (file_url, self.request.user.username, owner_type, owner, - request.data.get('import_type', 'npm'), index), task_id=task.id, queue=task.queue) + request.data.get('import_type', 'npm'), index, *([True] if isolated_version else [])), + task_id=task.id, queue=task.queue) return Response({ 'task': new_task.id, 'state': new_task.state diff --git a/core/integration_tests/tests_sources.py b/core/integration_tests/tests_sources.py index 21d4ae445..617fb7eb3 100644 --- a/core/integration_tests/tests_sources.py +++ b/core/integration_tests/tests_sources.py @@ -198,7 +198,8 @@ def test_post_201(self): 'canonical_url', 'identifier', 'publisher', 'contact', 'meta', 'jurisdiction', 'purpose', 'copyright', 'content_type', 'revision_date', 'logo_url', 'text', 'experimental', 'case_sensitive', 'collection_reference', 'hierarchy_meaning', 'compositional', - 'version_needed', 'hierarchy_root_url', 'autoid_concept_mnemonic', 'autoid_mapping_mnemonic', + 'version_needed', 'versioning_mode', 'hierarchy_root_url', 'autoid_concept_mnemonic', + 'autoid_mapping_mnemonic', 'autoid_concept_external_id', 'autoid_mapping_external_id', 'autoid_concept_name_external_id', 'autoid_concept_description_external_id', 'autoid_concept_mnemonic_start_from', 'autoid_concept_external_id_start_from', @@ -320,7 +321,8 @@ def test_put_200(self): 'canonical_url', 'identifier', 'publisher', 'contact', 'meta', 'jurisdiction', 'purpose', 'copyright', 'content_type', 'revision_date', 'logo_url', 'text', 'experimental', 'case_sensitive', 'collection_reference', 'hierarchy_meaning', 'compositional', - 'version_needed', 'hierarchy_root_url', 'autoid_concept_mnemonic', 'autoid_mapping_mnemonic', + 'version_needed', 'versioning_mode', 'hierarchy_root_url', 'autoid_concept_mnemonic', + 'autoid_mapping_mnemonic', 'autoid_concept_external_id', 'autoid_mapping_external_id', 'autoid_concept_name_external_id', 'autoid_concept_description_external_id', 'autoid_concept_mnemonic_start_from', 'autoid_concept_external_id_start_from', @@ -2918,6 +2920,29 @@ def test_markdown_output_for_all_verbosity_levels(self): self.assertIn('mapping-target-v1', markdown_output) self.assertIn('mapping-target-v2', markdown_output) + @override_settings(CHANGELOG_ENRICHMENT_MAX_RESOURCES=1) + def test_large_diff_skips_enrichment(self): + data = self._build_changelog_output_fixture() + token = data['source'].created_by.get_token() + + response = self._post_changelog(data['source_v1'], data['source_v2'], token, 4, output='json') + + self.assertEqual(response.status_code, 200) + self.assertTrue(response.data['meta']['enrichment_skipped']) + changed_concept = response.data['concepts']['changed_major']['concept-detailed'] + self.assertEqual(changed_concept['id'], 'concept-detailed') + self.assertIn('display_name', changed_concept) + self.assertNotIn('names', changed_concept) + self.assertNotIn('prev_names', changed_concept) + changed_mapping = response.data['mappings']['changed_major']['mapping-detailed'] + self.assertEqual(changed_mapping['to_concept'], 'mapping-target-v2') + self.assertNotIn('prev_to_concept', changed_mapping) + + response = self._post_changelog(data['source_v1'], data['source_v2'], token, 4, output='markdown') + + self.assertEqual(response.status_code, 200) + self.assertIn('without enrichment', response.data['markdown']) + class SourceVersionsComparisonViewTest(OCLAPITestCase): def test_post_200(self): # pylint: disable=too-many-locals,too-many-statements @@ -3326,6 +3351,10 @@ def test_post_200(self): # pylint: disable=too-many-statements,too-many-locals } ) + @patch('core.common.checksums.ChecksumChangelog.CONCEPTS_BATCH_SIZE', 1) + def test_post_200_in_concept_batches_of_one(self): + self.test_post_200() + class SourceReleaseFlagsTest(OCLAPITestCase): """ diff --git a/core/mappings/models.py b/core/mappings/models.py index e183d8370..658476b67 100644 --- a/core/mappings/models.py +++ b/core/mappings/models.py @@ -421,16 +421,13 @@ def is_existing_in_parent(self): return self.parent.mappings_set.filter(mnemonic__exact=self.mnemonic).exists() @classmethod - def create_new_version_for(cls, instance, data, user, cache=None): + def create_new_version_for(cls, instance, data, user, cache=None, isolated=False): # pylint: disable=too-many-arguments instance.populate_fields_from_relations(data, cache=cache, user=user) - instance.extras = data.get('extras', instance.extras) - instance.external_id = data.get('external_id', instance.external_id) instance.mnemonic = data.get('mnemonic', instance.mnemonic) instance.map_type = data.get('map_type', instance.map_type) - instance.sort_weight = data.get('sort_weight', instance.sort_weight) instance.comment = data.get('update_comment') or data.get('comment') - instance.retire_reason = data.get('retire_reason', instance.retire_reason) - instance.retired = data.get('retired', instance.retired) + instance.set_version_fields( + data, ('extras', 'external_id', 'sort_weight', 'retire_reason', 'retired'), isolated) return instance.save_as_new_version(user) diff --git a/core/settings.py b/core/settings.py index f3e8de165..b2a403015 100644 --- a/core/settings.py +++ b/core/settings.py @@ -732,6 +732,8 @@ def get_set_from_env(name): LIST_QUERY_TIMEOUT_MS = int(os.environ.get('LIST_QUERY_TIMEOUT_MS', 30000)) # Largest page a concept, mapping or reference list serves, so one page stays a short request LIST_MAX_LIMIT = int(os.environ.get('LIST_MAX_LIMIT', 100)) +# Above this many new, removed or changed concepts and mappings, a changelog skips names, descriptions and prev_* +CHANGELOG_ENRICHMENT_MAX_RESOURCES = int(os.environ.get('CHANGELOG_ENRICHMENT_MAX_RESOURCES', 50000)) ANALYTICS_API = os.environ.get('ANALYTICS_API', 'http://host.docker.internal:8002') if ANALYTICS_API: diff --git a/core/sources/constants.py b/core/sources/constants.py index 6702e5300..7ed4463cc 100644 --- a/core/sources/constants.py +++ b/core/sources/constants.py @@ -5,6 +5,28 @@ DELETE_SUCCESS = 'Successfully deleted Source.' VERSION_ALREADY_EXISTS = "Source version '{}' already exist." HIERARCHY_ROOT_MUST_BELONG_TO_SAME_SOURCE = 'Hierarchy Root must belong to the same Source.' +VERSIONING_MODE_CARRY_FORWARD = 'carry_forward' +VERSIONING_MODE_ISOLATED = 'isolated' +VERSIONING_MODES = ( + (VERSIONING_MODE_CARRY_FORWARD, 'Carry Forward'), + (VERSIONING_MODE_ISOLATED, 'Isolated'), +) +ISOLATED_VERSION_NEEDS_ONE_SOURCE_VERSION = \ + 'An isolated_version import needs exactly one Source Version line, after its concepts and mappings.' +ISOLATED_VERSION_CONTENT_FAILED = 'Not created: some of the version content failed to import.' +ISOLATED_VERSION_LINE_INVALID = 'The Source Version line needs an id.' +ISOLATED_VERSION_EXISTS = 'This Source Version already exists.' +ISOLATED_VERSION_NO_EDIT_ACCESS = 'You need edit access to the source to import a version into it.' +HIERARCHY_ROOT_CANNOT_BE_DROPPED = 'Hierarchy Root must be present in an isolated version.' +INVALID_VERSIONING_MODE = 'versioning_mode must be carry_forward or isolated.' +VERSIONING_MODE_NEEDS_ISOLATED_VERSION = \ + 'versioning_mode isolated needs isolated_version=true, and isolated_version=true makes an isolated version.' +ISOLATED_METADATA_FIELDS = ( + 'description', 'full_name', 'website', 'external_id', 'canonical_url', 'identifier', 'contact', 'jurisdiction', + 'publisher', 'purpose', 'copyright', 'revision_date', 'text', 'experimental', 'source_type', 'content_type', + 'collection_reference', 'hierarchy_meaning', 'case_sensitive', 'compositional', 'version_needed', +) +ISOLATED_REQUIRED_METADATA_FIELDS = ('name', 'default_locale', 'supported_locales') HIERARCHY_MEANINGS = ( ('grouped-by', 'grouped-by'), ('is-a', 'is-a'), diff --git a/core/sources/migrations/0047_source_versioning_mode.py b/core/sources/migrations/0047_source_versioning_mode.py new file mode 100644 index 000000000..a2c581375 --- /dev/null +++ b/core/sources/migrations/0047_source_versioning_mode.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.17 on 2026-11-05 13:45 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('sources', '0046_retire_public_edit_access'), + ] + + operations = [ + migrations.AddField( + model_name='source', + name='versioning_mode', + field=models.CharField(blank=True, choices=[('carry_forward', 'Carry Forward'), ('isolated', 'Isolated')], default='carry_forward', max_length=20, null=True), + ), + ] diff --git a/core/sources/models.py b/core/sources/models.py index 38ea9163b..a3f7c3f84 100644 --- a/core/sources/models.py +++ b/core/sources/models.py @@ -12,16 +12,19 @@ from pydash import get, compact from core.common.checksums import VersionCompareMixin -from core.common.constants import HEAD +from core.common.constants import HEAD, VERSION_UNCOPYABLE_EXTRAS from core.common.models import ConceptContainerModel from core.common.tasks import update_mappings_source, index_source_concepts, index_source_mappings, \ - sync_source_concept_vectors, index_source_public_access -from core.common.utils import to_camel_case + sync_source_concept_vectors, index_source_public_access, index_dropped_source_children +from core.common.utils import to_camel_case, chunks from core.common.validators import validate_non_negative -from core.concepts.models import ConceptName, Concept +from core.concepts.models import ConceptName, Concept, HierarchicalConcepts from core.services.storages.postgres import PostgresQL from core.sources.constants import SOURCE_TYPE, SOURCE_VERSION_TYPE, HIERARCHY_ROOT_MUST_BELONG_TO_SAME_SOURCE, \ - HIERARCHY_MEANINGS, AUTO_ID_CHOICES, AUTO_ID_SEQUENTIAL, AUTO_ID_UUID, LOCALE_EXTERNAL_AUTO_ID_CHOICES + HIERARCHY_MEANINGS, AUTO_ID_CHOICES, AUTO_ID_SEQUENTIAL, AUTO_ID_UUID, LOCALE_EXTERNAL_AUTO_ID_CHOICES, \ + VERSIONING_MODES, VERSIONING_MODE_CARRY_FORWARD, VERSIONING_MODE_ISOLATED, ISOLATED_METADATA_FIELDS, \ + ISOLATED_REQUIRED_METADATA_FIELDS, HIERARCHY_ROOT_CANNOT_BE_DROPPED, ISOLATED_VERSION_NEEDS_ONE_SOURCE_VERSION, \ + INVALID_VERSIONING_MODE, VERSIONING_MODE_NEEDS_ISOLATED_VERSION from core.tasks.models import Task @@ -107,6 +110,8 @@ class Meta: compositional = models.BooleanField(null=True, blank=True, default=None) version_needed = models.BooleanField(null=True, blank=True, default=None) hierarchy_root = models.ForeignKey('concepts.Concept', null=True, blank=True, on_delete=models.SET_NULL) + versioning_mode = models.CharField( + null=True, blank=True, max_length=20, choices=VERSIONING_MODES, default=VERSIONING_MODE_CARRY_FORWARD) # auto-id autoid_concept_mnemonic = models.CharField(null=True, blank=True, choices=AUTO_ID_CHOICES, max_length=10) autoid_concept_external_id = models.CharField(null=True, blank=True, choices=AUTO_ID_CHOICES, max_length=10) @@ -229,6 +234,98 @@ def update_version_data(self, head): self.compositional = head.compositional self.version_needed = head.version_needed + @property + def is_isolated(self): + return self.versioning_mode == VERSIONING_MODE_ISOLATED + + def apply_version_metadata_to_head(self, head): + self.versioning_mode = self.versioning_mode or VERSIONING_MODE_CARRY_FORWARD + if not self.is_isolated: + return + fields = [field for field in get(self, '_metadata_fields') or [] if field in ISOLATED_METADATA_FIELDS] + for field in fields: + setattr(head, field, getattr(self, field)) + for field in ISOLATED_REQUIRED_METADATA_FIELDS: + if field in (get(self, '_metadata_fields') or []) and getattr(self, field): + setattr(head, field, getattr(self, field)) + if 'extras' in (get(self, '_metadata_fields') or []): + head.extras = { + **{key: value for key, value in (head.extras or {}).items() if key in VERSION_UNCOPYABLE_EXTRAS}, + **(self.extras or {}) + } + head.save() + + def queue_index_dropped_children(self, concept_ids, mapping_ids): + if get(settings, 'TEST_MODE'): + self.index_dropped_children(concept_ids, mapping_ids) + return + task = Task.new(queue='indexing', user=self.updated_by, name=index_dropped_source_children.__name__) + index_dropped_source_children.apply_async( + (self.id, concept_ids, mapping_ids), task_id=task.id, queue='indexing') + + def index_dropped_children(self, concept_ids, mapping_ids): + from core.concepts.documents import ConceptDocument + from core.mappings.documents import MappingDocument + from core.mappings.models import Mapping + for model, document, ids in ((Concept, ConceptDocument, concept_ids), (Mapping, MappingDocument, mapping_ids)): + for chunk in chunks(ids, 5000): + self.batch_index(model.objects.filter(id__in=chunk), document) + self.update_children_counts(sync=True) + + def drop_unseen_children(self, seen_concept_ids, seen_mapping_ids): + from core.mappings.models import Mapping + to_index = {Concept: [], Mapping: []} + with transaction.atomic(): + Source.objects.select_for_update().get(id=self.id) + concept_ids = set(self.concepts_set.filter( + id=F('versioned_object_id'), is_active=True).values_list('id', flat=True)) - set(seen_concept_ids) + mapping_ids = set(self.mappings_set.filter( + id=F('versioned_object_id'), is_active=True).values_list('id', flat=True)) - set(seen_mapping_ids) + if self.hierarchy_root_id and Concept.objects.filter( + id=self.hierarchy_root_id, versioned_object_id__in=concept_ids).exists(): + raise ValidationError({'hierarchy_root': [HIERARCHY_ROOT_CANNOT_BE_DROPPED]}) + for model, ids in ((Concept, concept_ids), (Mapping, mapping_ids)): + for chunk in chunks(sorted(ids), 5000): + latest_ids = list(model.objects.filter( + versioned_object_id__in=chunk, is_latest_version=True).values_list('id', flat=True)) + model.objects.filter(id__in=chunk).update(is_active=False) + model.sources.through.objects.filter( + source_id=self.id, **{f'{model.__name__.lower()}_id__in': [*chunk, *latest_ids]}).delete() + if model is Concept: + HierarchicalConcepts.objects.filter(child_id__in=chunk).delete() + to_index[model] += [*chunk, *latest_ids] + + if to_index[Concept] or to_index[Mapping]: + # registered robustly so a failure to queue can't stop the version's seeding + transaction.on_commit( + lambda: self.queue_index_dropped_children(to_index[Concept], to_index[Mapping]), robust=True) + return {'concepts': len(concept_ids), 'mappings': len(mapping_ids)} + + @staticmethod + def resolve_versioning_mode(isolated_version, requested_mode=None): + mode = VERSIONING_MODE_ISOLATED if isolated_version else VERSIONING_MODE_CARRY_FORWARD + if requested_mode in (None, ''): + return mode + if not isinstance(requested_mode, str) or requested_mode not in dict(VERSIONING_MODES): + raise ValidationError({'versioning_mode': [INVALID_VERSIONING_MODE]}) + if requested_mode != mode: + raise ValidationError({'versioning_mode': [VERSIONING_MODE_NEEDS_ISOLATED_VERSION]}) + return mode + + @classmethod + def validate_import_versioning_modes(cls, input_list, isolated_version): + """Checks the whole import before anything runs. An isolated_version import gets its one version line back.""" + types = [(line.get('type', '') or '').lower() for line in input_list] + version_indexes = [index for index, _type in enumerate(types) if _type == 'source version'] + for index in version_indexes: + cls.resolve_versioning_mode(isolated_version, input_list[index].get('versioning_mode')) + if not isolated_version: + return None + content_indexes = [index for index, _type in enumerate(types) if _type in ('concept', 'mapping')] + if len(version_indexes) != 1 or (content_indexes and content_indexes[-1] > version_indexes[0]): + raise ValidationError(ISOLATED_VERSION_NEEDS_ONE_SOURCE_VERSION) + return input_list[version_indexes[0]] + def get_concept_name_locales(self): return ConceptName.objects.filter(concept__in=self.get_active_concepts()) diff --git a/core/sources/serializers.py b/core/sources/serializers.py index 30a9f3eba..472352c4c 100644 --- a/core/sources/serializers.py +++ b/core/sources/serializers.py @@ -263,6 +263,7 @@ class SourceCreateSerializer(SourceCreateOrUpdateSerializer): hierarchy_meaning = CharField(required=False, allow_null=True, allow_blank=True) collection_reference = CharField(required=False, allow_null=True, allow_blank=True) hierarchy_root_url = CharField(allow_null=True, allow_blank=True, required=False) + versioning_mode = CharField(read_only=True) def validate(self, attrs): # create/create_version consume the raw payload (not validated_data), so DRF's boolean @@ -427,7 +428,7 @@ class Meta: 'canonical_url', 'identifier', 'publisher', 'contact', 'jurisdiction', 'purpose', 'copyright', 'content_type', 'revision_date', 'logo_url', 'summary', 'text', 'client_configs', 'experimental', 'case_sensitive', 'collection_reference', 'hierarchy_meaning', 'compositional', - 'version_needed', 'hierarchy_root_url', 'hierarchy_root', 'meta', + 'version_needed', 'hierarchy_root_url', 'hierarchy_root', 'meta', 'versioning_mode', 'autoid_concept_mnemonic', 'autoid_concept_external_id', 'autoid_concept_name_external_id', 'autoid_concept_description_external_id', 'autoid_mapping_mnemonic', 'autoid_mapping_external_id', @@ -523,6 +524,7 @@ class Meta: 'content_type', 'revision_date', 'summary', 'text', 'meta', 'experimental', 'case_sensitive', 'collection_reference', 'hierarchy_meaning', 'compositional', 'version_needed', 'hierarchy_root_url', 'checksums', 'states', 'tasks', 'properties', 'filters', + 'versioning_mode', 'match_algorithms', 'external_exports' ) + AbstractRepoResourcesSerializer.Meta.fields