OpenConceptLab/ocl_online#397 | changing source public_access re-index resources via task

core/common/mixins.py

@@ -535,6 +535,8 @@ def update(self, request, *args, **kwargs):  # pylint: disable=unused-argument
             if serializer.is_valid():
                 self.object.get_checksums(recalculate=True)
                 serializer = self.get_detail_serializer(self.object)
+                if get(self.object, '_public_access_task'):
+                    success_status_code = status.HTTP_202_ACCEPTED
                 return Response(serializer.data, status=success_status_code)
 
         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

core/common/models.py

@@ -543,7 +543,6 @@ def get_actions(ids):
                     '_index': index_name,
                     '_id': object_id,
                     'doc': partial_doc,
-                    'doc_as_upsert': True,
                 }
 
         return BaseModel.batch_index_partial_by_ids(queryset, document, get_actions, single_batch, parallel)

core/common/tasks.py

@@ -903,6 +903,35 @@ def index_source_mappings(
             source.clear_mappings_cache()
 
 
+@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
+)
+def index_source_public_access(self, source_id):
+    """
+    Sets public_can_view on a source's concept and mapping docs from its current public_access, with retries; the
+    source shows as processing meanwhile. Not queued once, so each public_access change gets its own run.
+    """
+    from core.sources.models import Source
+    from core.concepts.documents import ConceptDocument
+    from core.mappings.documents import MappingDocument
+    source = Source.objects.filter(id=source_id).first()
+    if not source:
+        return
+    task_id = self.request.id
+    source.add_processing(task_id)
+    try:
+        partial_doc = {'public_can_view': source.public_can_view}
+        batch_index_with_summary(source.batch_index, source.concepts_set, ConceptDocument, partial_doc=partial_doc)
+        batch_index_with_summary(source.batch_index, source.mappings_set, MappingDocument, partial_doc=partial_doc)
+    except Exception:
+        if self.request.retries < self.max_retries:
+            raise
+        source.remove_processing(task_id)
+        raise
+    source.remove_processing(task_id)
+
+
 @app.task(base=QueueOnceCustomTask)
 def update_source_active_concepts_count(source_id):
     from core.sources.models import Source

core/common/tests.py

@@ -1719,18 +1719,43 @@ def get_batch(batch_slice):
                 '_index': doc_instance._index._name,  # pylint: disable=protected-access
                 '_id': 10,
                 'doc': {'flag': True},
-                'doc_as_upsert': True
             },
             {
                 '_op_type': 'update',
                 '_index': doc_instance._index._name,  # pylint: disable=protected-access
                 '_id': 11,
                 'doc': {'flag': True},
-                'doc_as_upsert': True
             }
         ])
         self.assertEqual([call.kwargs for call in bulk_calls], [{'raise_on_error': False}])
 
+    @override_settings(TEST_MODE=False)
+    @patch('core.common.models.parallel_bulk', return_value=[(False, {'update': {'_id': 11, 'status': 404}})])
+    def test_batch_index_partial_full_indexes_missing_docs_instead_of_upserting(self, _):
+        ids_queryset = MagicMock()
+        ids_queryset.__getitem__.side_effect = lambda s: [10, 11] if s == slice(0, 500, None) else []
+
+        queryset = Mock()
+        ordered_queryset = Mock()
+        ordered_queryset.values_list.return_value = ids_queryset
+        queryset.order_by.return_value = ordered_queryset
+        missing_queryset = Mock()
+        queryset.filter.return_value = missing_queryset
+
+        doc_instance = Mock()
+        doc_instance.django.auto_refresh = False
+        doc_instance.django.queryset_pagination = None
+        document = Mock(return_value=doc_instance)
+
+        with patch.object(BaseModel, 'batch_index_full') as batch_index_full_mock:
+            BaseModel.batch_index_partial(queryset, document, False, {'public_can_view': False})
+
+        queryset.filter.assert_called_once_with(id__in={11})
+        batch_index_full_mock.assert_called_once_with(
+            single_batch=False, queryset=missing_queryset, document=document, prefetch=[], select_related=[],
+            refresh=False
+        )
+
     @override_settings(TEST_MODE=False)
     @patch('core.common.models.parallel_bulk', return_value=[])
     def test_batch_index_source_version_append_streams_batches_without_parallel_bulk(self, parallel_bulk_mock):

core/sources/models.py

@@ -15,7 +15,7 @@
 from core.common.constants import HEAD
 from core.common.models import ConceptContainerModel
 from core.common.tasks import update_mappings_source, index_source_concepts, index_source_mappings, \
-    sync_source_concept_vectors, resolve_url_registry_entries
+    sync_source_concept_vectors, resolve_url_registry_entries, index_source_public_access
 from core.common.utils import to_camel_case
 from core.common.validators import validate_non_negative
 from core.concepts.models import ConceptName, Concept
@@ -544,6 +544,20 @@ def index_concepts_async(self, user, partial_doc=None, locales=None, exclude_loc
         except AlreadyQueued:
             pass
 
+    def index_public_access_async(self, user=None):
+        """
+        Queues the public_can_view re-index of this source's docs once the change is committed, marking the source
+        processing before the response goes out, and returns its Task. In TEST_MODE it runs inline: returns None.
+        """
+        if get(settings, 'TEST_MODE', False):
+            index_source_public_access(self.id)  # pylint: disable=no-value-for-parameter
+            return None
+        task = Task.new(queue='indexing', user=user or self.updated_by, name=index_source_public_access.__name__)
+        self.add_processing(task.id)
+        transaction.on_commit(lambda: index_source_public_access.apply_async(
+            (self.id, ), queue='indexing', persist_args=True, task_id=task.id))
+        return task
+
     def sync_concept_vectors_async(self, user=None, recheck=True, countdown=None):
         """
         Queues this version's vector sync (sync_source_concept_vectors) on the indexing queue, with its arguments

core/sources/signals.py

@@ -20,10 +20,5 @@ def propagate_parent_attributes(sender, instance=None, created=False, **kwargs):
             updated_mappings = instance.mappings_set.exclude(
                 public_access=instance.public_access).update(public_access=instance.public_access)
 
-            partial_doc = {'public_can_view': instance.public_can_view}
-            if updated_concepts:
-                from core.concepts.documents import ConceptDocument
-                instance.batch_index(instance.concepts_set, ConceptDocument, partial_doc=partial_doc)
-            if updated_mappings:
-                from core.mappings.documents import MappingDocument
-                instance.batch_index(instance.mappings_set, MappingDocument, partial_doc=partial_doc)
+            if updated_concepts or updated_mappings:
+                instance._public_access_task = instance.index_public_access_async()  # pylint: disable=protected-access

core/sources/tests/tests.py

@@ -24,6 +24,7 @@
 from core.common.tasks import update_source_active_concepts_count
 from core.common.tasks import update_source_active_mappings_count
 from core.common.tasks import update_validation_schema
+from core.common.tasks import index_source_public_access
 from core.common.tests import OCLTestCase, OCLAPITestCase
 from core.concepts.documents import ConceptDocument
 from core.concepts.models import Concept
@@ -130,6 +131,34 @@ def test_version_update_sets_external_id_from_version_external_id(self):
         version.refresh_from_db()
         self.assertEqual(version.external_id, 'ext-1')
 
+    @patch('core.sources.models.Source.index_public_access_async')
+    def test_put_public_access_change_returns_202_with_reindex_queued(self, index_public_access_async_mock):
+        source = OrganizationSourceFactory(
+            created_by=self.admin, updated_by=self.admin, public_access=ACCESS_TYPE_VIEW)
+        concept = ConceptFactory(parent=source, public_access=ACCESS_TYPE_VIEW)
+
+        response = self.client.put(
+            source.uri, {'public_access': ACCESS_TYPE_NONE}, format='json',
+            HTTP_AUTHORIZATION=f"Token {self.admin_token}"
+        )
+
+        self.assertEqual(response.status_code, 202)
+        self.assertEqual(response.data['public_access'], ACCESS_TYPE_NONE)
+        index_public_access_async_mock.assert_called_once_with()
+        concept.refresh_from_db()
+        self.assertEqual(concept.public_access, ACCESS_TYPE_NONE)
+
+    @patch('core.sources.models.Source.index_public_access_async')
+    def test_put_without_public_access_change_returns_200(self, index_public_access_async_mock):
+        source = OrganizationSourceFactory(created_by=self.admin, updated_by=self.admin)
+        ConceptFactory(parent=source)
+
+        response = self.client.put(
+            source.uri, {'name': 'renamed'}, format='json', HTTP_AUTHORIZATION=f"Token {self.admin_token}")
+
+        self.assertEqual(response.status_code, 200)
+        index_public_access_async_mock.assert_not_called()
+
     def test_version_delete_validation_error(self):
         source = OrganizationSourceFactory(created_by=self.admin, updated_by=self.admin)
         version = OrganizationSourceFactory(
@@ -2078,6 +2107,35 @@ def test_index_mappings_async_swallows_already_queued(self, mock_task):
         source = OrganizationSourceFactory()
         source.index_mappings_async(source.created_by)
 
+    @override_settings(TEST_MODE=False)
+    @patch('core.sources.models.index_source_public_access')
+    def test_index_public_access_async_marks_processing_and_queues_on_commit(self, mock_task):
+        mock_task.__name__ = 'index_source_public_access'
+        source = OrganizationSourceFactory()
+
+        with self.captureOnCommitCallbacks(execute=False) as callbacks:
+            task = source.index_public_access_async(source.created_by)
+
+        self.assertEqual(task.name, 'index_source_public_access')
+        self.assertEqual(task.queue, 'indexing')
+        source.refresh_from_db()
+        self.assertEqual(source._background_process_ids, [task.id])  # pylint: disable=protected-access
+        mock_task.apply_async.assert_not_called()
+
+        for callback in callbacks:
+            callback()
+        mock_task.apply_async.assert_called_once_with(
+            (source.id, ), queue='indexing', persist_args=True, task_id=task.id)
+
+    @patch('core.sources.models.index_source_public_access')
+    def test_index_public_access_async_runs_inline_in_test_mode(self, mock_task):
+        source = OrganizationSourceFactory()
+
+        self.assertIsNone(source.index_public_access_async())
+
+        mock_task.assert_called_once_with(source.id)
+        mock_task.apply_async.assert_not_called()
+
     @patch('core.sources.models.index_source_concepts')
     def test_index_concepts_async_swallows_already_queued(self, mock_task):
         mock_task.__name__ = 'index_source_concepts'
@@ -2310,6 +2368,31 @@ def test_propagate_parent_attributes_updates_mapping_public_access(self):
         mapping.refresh_from_db()
         self.assertEqual(mapping.public_access, ACCESS_TYPE_VIEW)
 
+    @patch('core.sources.models.Source.index_public_access_async')
+    def test_propagate_parent_attributes_queues_public_access_reindex(self, index_public_access_async_mock):
+        source = OrganizationSourceFactory(public_access=ACCESS_TYPE_VIEW)
+        ConceptFactory(parent=source, public_access=ACCESS_TYPE_VIEW)
+
+        source.public_access = ACCESS_TYPE_NONE
+        source._should_update_public_access = True  # pylint: disable=protected-access
+        source.save()
+
+        index_public_access_async_mock.assert_called_once_with()
+        self.assertEqual(
+            source._public_access_task, index_public_access_async_mock.return_value)  # pylint: disable=protected-access
+
+    @patch('core.sources.models.Source.index_public_access_async')
+    def test_propagate_parent_attributes_skips_reindex_without_children_to_update(
+            self, index_public_access_async_mock):
+        source = OrganizationSourceFactory(public_access=ACCESS_TYPE_VIEW)
+
+        source.public_access = ACCESS_TYPE_NONE
+        source._should_update_public_access = True  # pylint: disable=protected-access
+        source.save()
+
+        index_public_access_async_mock.assert_not_called()
+        self.assertFalse(hasattr(source, '_public_access_task'))
+
 
 class SourceCloneAPITest(OCLAPITestCase):
     def test_clone_api_returns_structured_errors_and_rolls_back(self):
@@ -2425,6 +2508,59 @@ def test_clean_filters_extra_keys(self):
             source.clean_filters()
 
 
+class IndexSourcePublicAccessTaskTest(OCLTestCase):
+    @staticmethod
+    def run_task(source_id, task_id='public-access-task', retries=0):
+        """Runs the task body as a worker would, with this task id and retry count."""
+        index_source_public_access.push_request(id=task_id, retries=retries)
+        try:
+            return index_source_public_access.run(source_id)
+        finally:
+            index_source_public_access.pop_request()
+
+    @patch('core.sources.models.Source.batch_index')
+    def test_sets_public_can_view_on_concepts_and_mappings_while_processing(self, batch_index_mock):
+        source = OrganizationSourceFactory(public_access=ACCESS_TYPE_NONE)
+        processing_ids = []
+        batch_index_mock.side_effect = lambda *args, **kwargs: processing_ids.append(
+            Source.objects.get(id=source.id)._background_process_ids)  # pylint: disable=protected-access
+
+        self.run_task(source.id)
+
+        self.assertEqual(batch_index_mock.call_args_list, [
+            call(ANY, ConceptDocument, partial_doc={'public_can_view': False}),
+            call(ANY, MappingDocument, partial_doc={'public_can_view': False}),
+        ])
+        self.assertEqual(processing_ids, [['public-access-task'], ['public-access-task']])
+        source.refresh_from_db()
+        self.assertEqual(source._background_process_ids, [])  # pylint: disable=protected-access
+
+    @patch('core.sources.models.Source.batch_index', side_effect=BatchIndexingError('ES down', {}))
+    def test_stays_processing_when_it_will_be_retried(self, _):
+        source = OrganizationSourceFactory()
+
+        with self.assertRaises(BatchIndexingError):
+            self.run_task(source.id, retries=0)
+
+        source.refresh_from_db()
+        self.assertEqual(source._background_process_ids, ['public-access-task'])  # pylint: disable=protected-access
+
+    @patch('core.sources.models.Source.batch_index', side_effect=BatchIndexingError('ES down', {}))
+    def test_stops_processing_when_out_of_retries(self, _):
+        source = OrganizationSourceFactory()
+
+        with self.assertRaises(BatchIndexingError):
+            self.run_task(source.id, retries=index_source_public_access.max_retries)
+
+        source.refresh_from_db()
+        self.assertEqual(source._background_process_ids, [])  # pylint: disable=protected-access
+
+    @patch('core.sources.models.Source.batch_index')
+    def test_does_nothing_for_missing_source(self, batch_index_mock):
+        self.assertIsNone(self.run_task(0))
+        batch_index_mock.assert_not_called()
+
+
 class TasksTest(OCLTestCase):
     @patch('core.sources.models.Source.index_children')
     @patch('core.common.tasks.export_source')