Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions core/collections/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,3 +37,7 @@
UNRESOLVED_RESOURCE_VERSION_TYPE = 'reference.resource_version'
INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE = 'INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE'
EXPANSION_LOCKED_REPO_VERSION_TOGGLE = 'EXPANSION_LOCKED_REPO_VERSION_TOGGLE'
VERSION_EXPANSION_LOCK = 'lock'
VERSION_EXPANSION_LATEST = 'latest'
VERSION_EXPANSION_NONE = 'none'
VERSION_EXPANSION_CHOICES = (VERSION_EXPANSION_LOCK, VERSION_EXPANSION_LATEST, VERSION_EXPANSION_NONE)
112 changes: 67 additions & 45 deletions core/collections/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,8 @@
REFERENCE_TYPE_CHOICES, CONCEPT_REFERENCE_TYPE, MAPPING_REFERENCE_TYPE, SOURCE_MAPPINGS, SOURCE_TO_CONCEPTS,
TRANSFORM_TO_RESOURCE_VERSIONS, COLLECTION_REFERENCE_TYPE, TRANSFORM_TO_EXTENSIONAL,
RESOURCE_VERSION_REPO_RESOLUTION_TOGGLE, UNRESOLVED_RESOURCE_VERSION_TYPE,
INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE, EXPANSION_LOCKED_REPO_VERSION_TOGGLE)
INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE, EXPANSION_LOCKED_REPO_VERSION_TOGGLE, VERSION_EXPANSION_LOCK,
VERSION_EXPANSION_LATEST)
from core.collections.parsers import CollectionReferenceParser
from core.collections.translators import CollectionReferenceTranslator
from core.collections.utils import is_concept, is_mapping
Expand Down Expand Up @@ -401,36 +402,36 @@ def get_locked_repo_versions(self, relation):
expansion = self.expansion if CollectionReference.is_expansion_locked_repo_version_on() else None
return list(getattr(expansion, relation).filter(is_active=True).order_by('-id')) if expansion else []

def can_seed_from_head_expansion(self):
return bool(
not self.is_head and self.should_auto_expand and CollectionReference.is_expansion_locked_repo_version_on()
and get(self.head, 'expansion')
)
def get_expansion_choice(self, expansion=None):
"""How a new version's expansion is built with the lock toggle on: lock or latest, else None (as before)"""
if self.is_head or not self.should_auto_expand or not CollectionReference.is_expansion_locked_repo_version_on():
return None
if expansion in (VERSION_EXPANSION_LOCK, VERSION_EXPANSION_LATEST):
return expansion
return VERSION_EXPANSION_LOCK if get(self.head, 'expansion') else VERSION_EXPANSION_LATEST

def seed_from_head_expansion(self, index=True):
"""False when HEAD has no expansion to copy, so the version seeds as without one"""
def seed_from_head_expansion(self, lock=True, index=True):
"""References from HEAD, evaluated again with HEAD's expansion parameters (and its locks, with lock)"""
is_outermost = not transaction.get_connection().in_atomic_block
attempts = SERIALIZATION_RETRY_ATTEMPTS if is_outermost else 1
for attempt in range(1, attempts + 1):
try:
expansion = self.seed_from_head_expansion_in_snapshot(is_outermost)
expansion = self.seed_from_head_expansion_in_snapshot(is_outermost, lock)
break
except OperationalError as ex:
if attempt == attempts or not self.is_serialization_failure(ex):
raise
self.refresh_from_db()
self.__dict__.pop('locked_source_versions', None)
self.__dict__.pop('locked_collection_versions', None)
if not expansion:
return False
if index:
expansion.index_resources(
list(expansion.concepts.values_list('id', flat=True)),
list(expansion.mappings.values_list('id', flat=True)))
return True

def seed_from_head_expansion_in_snapshot(self, is_outermost):
"""References and expansion copied from HEAD in one transaction, so both come from the same HEAD state"""
def seed_from_head_expansion_in_snapshot(self, is_outermost, lock=True):
"""References, parameters and locks read from HEAD in one transaction, so all come from the same HEAD state"""
with transaction.atomic():
if is_outermost:
with transaction.get_connection().cursor() as cursor:
Expand All @@ -441,30 +442,28 @@ def seed_from_head_expansion_in_snapshot(self, is_outermost):
return expansion
head = self.head
head_expansion = Expansion.objects.filter(uri=head.expansion_uri).first() if head.expansion_uri else None
if not head_expansion:
return None
self.seed_references()
parameters = dict(get(head_expansion, 'parameters') or {})
system_versions = head_expansion.get_locked_system_version_parameter() if lock and head_expansion else None
if system_versions:
parameters[ExpansionParameters.INCLUDE_SYSTEM] = system_versions
expansion = Expansion(
mnemonic=mnemonic, collection_version=self, parameters=head_expansion.parameters,
mnemonic=mnemonic, collection_version=self, parameters=parameters,
created_by_id=self.created_by_id, updated_by_id=self.created_by_id)
expansion.clean()
expansion.full_clean()
expansion.save()
for relation in Expansion.REPO_VERSION_RELATIONS:
getattr(expansion, relation).set(getattr(head_expansion, relation).all())
if lock and head_expansion:
# system-version doesn't reach value sets, so they start from HEAD's value set lock
expansion.evaluated_collection_versions.set(
head_expansion.evaluated_collection_versions.filter(is_active=True))
if not self.expansion_uri:
self.expansion_uri = expansion.uri
self.save()
if head_expansion.is_processing or head_expansion.has_pending_readd:
# HEAD's expansion is mid-build, so evaluate this version's references against HEAD's locks
references = list(self.references.all())
for reference in references:
reference._async = True # pylint: disable=protected-access
expansion.add_references(references, index=False, force_reevaluate=True)
else:
expansion.copy_members_from(head_expansion)
expansion.unresolved_repo_versions = head_expansion.unresolved_repo_versions
expansion.save(update_fields=['unresolved_repo_versions'])
references = list(self.references.all())
for reference in references:
reference._async = True # pylint: disable=protected-access
expansion.add_references(references, index=False, force_reevaluate=True)
self.replace_head_resources(expansion)
return expansion

Expand Down Expand Up @@ -1831,7 +1830,7 @@ def add_references( # pylint: disable=too-many-locals,too-many-statements,too-m
for system_version in compact(system_versions.split(',')):
_system = system_version.strip()
_cache_key = f"NONE-{_system}-NONE"
version, _ = ConceptContainerModel.resolve_reference_expression(_system)
version = self.resolve_system_version_entry(_system)
_system_version_cache[_cache_key] = version
if version.id:
include_system_versions.append(version)
Expand Down Expand Up @@ -2036,20 +2035,29 @@ def index_resources(self, concepts, mappings):
if mappings:
self.index_mappings(mappings)

@property
def has_pending_readd(self):
"""A reference removal leaves this expansion short until its re-add succeeds"""
return Task.unfinished_readd_references_tasks().filter(args__0=self.id).exists()

def copy_members_from(self, expansion):
for relation in ('concepts', 'mappings'):
through = getattr(Expansion, relation).through
field = f'{relation[:-1]}_id'
with transaction.get_connection().cursor() as cursor:
cursor.execute(
f'INSERT INTO {through._meta.db_table} (expansion_id, {field}) ' # pylint: disable=protected-access
f'SELECT %s, {field} FROM {through._meta.db_table} WHERE expansion_id = %s', # pylint: disable=protected-access
[self.id, expansion.id])
@staticmethod
def resolve_system_version_entry(entry):
version, _ = ConceptContainerModel.resolve_reference_expression(entry.strip())
return version

def get_locked_system_version_parameter(self):
"""This expansion's system-version, plus each source it locked that the parameter doesn't name yet"""
entries = [entry.strip() for entry in compact(
str(get(self.parameters, ExpansionParameters.INCLUDE_SYSTEM) or '').split(','))]
named = set()
for entry in entries:
resolved = self.resolve_system_version_entry(entry)
if resolved.id:
named.add(resolved.versioned_object_url)
for version in self.evaluated_source_versions.filter(is_active=True).order_by('-id'):
if version.versioned_object_url in named:
continue
named.add(version.versioned_object_url)
entry = f"{version.canonical_url}|{version.version}" if version.canonical_url else None
if not entry or get(self.resolve_system_version_entry(entry), 'id') != version.id:
entry = f"{version.versioned_object_url}|{version.version}"
entries.append(entry)
return ','.join(entries)

@staticmethod
def replace_head_rows(rows, owner, field, klass):
Expand All @@ -2066,12 +2074,26 @@ def replace_head_rows(rows, owner, field, klass):
rows.model.objects.filter(id__in=head_rows.values('id')).update(**{f'{field}_id': Subquery(latest)})

def seed_children(self, index=True, force_reevaluate=False):
version = self.collection_version
# a version's default expansion holds resource versions, so a rebuild swaps HEAD rows as its creation did
swap_head = force_reevaluate and not version.is_head and self.is_default and \
CollectionReference.is_expansion_locked_repo_version_on()
previous_concept_ids = set(self.concepts.values_list('id', flat=True)) if swap_head and index else set()
previous_mapping_ids = set(self.mappings.values_list('id', flat=True)) if swap_head and index else set()
self.add_references(
references=self.collection_version.references,
index=index,
references=version.references,
index=index and not swap_head,
is_adding_all=True,
force_reevaluate=force_reevaluate
)
if swap_head:
version.replace_head_resources(self)
if index:
concept_ids = set(self.concepts.values_list('id', flat=True))
mapping_ids = set(self.mappings.values_list('id', flat=True))
self.index_resources(list(concept_ids), list(mapping_ids))
self.full_index_resources(
list(previous_concept_ids - concept_ids), list(previous_mapping_ids - mapping_ids))

def wait_until_processed(self): # pragma: no cover
processing = self.is_processing
Expand Down
11 changes: 9 additions & 2 deletions core/collections/serializers.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,12 @@
from django.core.validators import RegexValidator
from pydash import get
from rest_framework.fields import CharField, ListField, IntegerField, DateTimeField, JSONField, \
BooleanField, SerializerMethodField
BooleanField, SerializerMethodField, ChoiceField
from rest_framework.relations import PrimaryKeyRelatedField
from rest_framework.serializers import ModelSerializer, Serializer, ValidationError

from core.client_configs.serializers import ClientConfigSerializer
from core.collections.constants import VERSION_EXPANSION_CHOICES, VERSION_EXPANSION_NONE
from core.collections.models import Collection, CollectionReference, Expansion
from core.common.constants import HEAD, DEFAULT_ACCESS_TYPE, NAMESPACE_REGEX, INCLUDE_SUMMARY, \
INCLUDE_CLIENT_CONFIGS, INVALID_EXPANSION_URL, INCLUDE_STATES, INCLUDE_TASKS, INCLUDE_RESOLVED_REPO_VERSIONS, \
Expand Down Expand Up @@ -264,6 +265,7 @@ class CollectionCreateSerializer(CollectionCreateOrUpdateSerializer):
locked_date = DateTimeField(required=False, allow_null=True)
autoexpand_head = BooleanField(required=False, default=True)
autoexpand = BooleanField(required=False, default=True)
expansion = ChoiceField(choices=VERSION_EXPANSION_CHOICES, required=False, write_only=True)

def validate(self, attrs):
# create/create_version consume the raw payload (not validated_data), so DRF's boolean
Expand All @@ -287,7 +289,12 @@ def create(self, validated_data):
def create_version(self, validated_data):
collection = self.prepare_object(validated_data)
user = self.context['request'].user
errors = Collection.persist_new_version(collection, user)
expansion = validated_data.get('expansion')
if expansion == VERSION_EXPANSION_NONE:
collection.autoexpand = False
elif expansion:
collection.autoexpand = True
errors = Collection.persist_new_version(collection, user, expansion=expansion)
self._errors.update(errors)
return collection

Expand Down
Loading
Loading