IndexJob model

backend/addcorpus/exceptions.py

@@ -0,0 +1,33 @@
+from addcorpus.models import Corpus
+
+
+class PythonDefinitionRequired(Exception):
+    '''
+    Exception that can be raised when attempting to use functionality only applicable for
+    Python corpora, on a corpus that does not have a Python definition.
+    '''
+
+    def __init__(self, corpus: Corpus, message: str, *args):
+        self.corpus = corpus
+        self.message = message
+        super().__init__(*args)
+
+
+    def __str__(self):
+        return f'{self.message} (corpus: {self.corpus})'
+
+class NoPythonDefinitionAllowed(Exception):
+    '''
+    Exception that can be raised when attempting to use functionality only applicable for
+    database-only corpora, on a corpus with a Python definition.
+    '''
+
+    def __init__(self, corpus: Corpus, message: str, *args):
+        self.corpus = corpus
+        self.message = message
+        super().__init__(*args)
+
+
+    def __str__(self):
+        return f'{self.message} (corpus: {self.corpus})'
+

backend/addcorpus/python_corpora/corpus.py

@@ -516,16 +516,26 @@ def f(metadata):
     return f
 
 
-def consolidate_start_end_years(start, end, min_date, max_date):
+def consolidate_start_end_years(
+        start: Union[datetime, date, int],
+        end: Union[datetime, date, int],
+        min_date: datetime,
+        max_date: datetime
+):
     ''' given a start and end date provided by the user, make sure
     - that start is not before end
     - that start is not before min_date (corpus variable)
     - that end is not after max_date (corpus variable)
     '''
     if isinstance(start, int):
         start = datetime(year=start, month=1, day=1)
+    elif isinstance(start, date):
+        start = datetime(year=start.year, month=start.month, day=start.day)
     if isinstance(end, int):
         end = datetime(year=end, month=12, day=31)
+    elif isinstance(end, date):
+        end = datetime(year=end.year, month=end.month, day=end.day)
+
     if start > end:
         tmp = start
         start = end

backend/addcorpus/python_corpora/filters.py

@@ -3,7 +3,7 @@
 passed through to ElasticSearch.
 '''
 
-from datetime import datetime
+from datetime import datetime, date
 from addcorpus.constants import MappingType
 
 class Filter(object):
@@ -24,7 +24,7 @@ def serialize(self):
         search_dict = {'name': name}
         for key, value in self.__dict__.items():
             if key == 'search_filter' or key != 'field':
-                if type(value) == datetime:
+                if isinstance(value, datetime) or isinstance(value, date):
                     search_dict[key] = value.isoformat()
                 else:
                     search_dict[key] = value

backend/conftest.py

@@ -9,11 +9,12 @@
 
 from ianalyzer.elasticsearch import client_from_config
 from addcorpus.python_corpora.save_corpus import load_and_save_all_corpora
-from es import es_index as index
+from es import es_index as index, sync
 from django.conf import settings
 from django.contrib.auth.models import Group
 from addcorpus.models import Corpus
 from addcorpus.serializers import CorpusJSONDefinitionSerializer
+from es.models import Server
 
 @pytest.fixture(autouse=True)
 def media_dir(tmpdir, settings):
@@ -112,6 +113,13 @@ def es_client():
 
     return client
 
+
+@pytest.fixture()
+def es_server(db, settings) -> Server:
+    sync.update_server_table_from_settings()
+    return Server.objects.get(name='default')
+
+
 @pytest.fixture()
 def basic_mock_corpus() -> str:
     return 'mock-csv-corpus'
@@ -157,8 +165,9 @@ def _index_test_corpus(es_client: Elasticsearch, corpus_name: str):
 
     if not es_client.indices.exists(index=corpus.configuration.es_index):
         with warnings.catch_warnings():
-            warnings.filterwarnings('ignore', message="Corpus has no 'id' field")
-            index.perform_indexing(corpus)
+            job = index.create_indexing_job(corpus)
+            index.perform_indexing(job)
+
         # ES is "near real time", so give it a second before we start searching the index
         sleep(2)
 

backend/corpora/jewishmigration/test_jewishmigration.py

@@ -154,7 +154,8 @@ def jm_client(es_client, jm_corpus):
     Returns an elastic search client for the mock corpus.
     """
     # add data from mock corpus
-    es_index.create(es_client, jm_corpus, False, True, False)
+    job = es_index.create_indexing_job(jm_corpus, mappings_only=True, clear=True)
+    es_index.perform_indexing(job)
 
     # ES is "near real time", so give it a second before we start searching the index
     sleep(1)

backend/es/conftest.py

@@ -7,7 +7,8 @@
 from addcorpus.python_corpora.load_corpus import load_corpus_definition
 from addcorpus.models import Corpus
 from es import es_index
-
+from es.models import Index
+from indexing.models import IndexJob, CreateIndexTask
 
 @pytest.fixture(scope='session')
 def mock_corpus():
@@ -58,14 +59,26 @@ def es_index_client(es_client, mock_corpus):
         es_client.indices.delete(index=index)
 
 @pytest.fixture()
-def es_alias_client(es_client, mock_corpus):
+def es_alias_client(db, es_server, es_client, mock_corpus):
     """
     Create multiple indices with version numbers for the mock corpus in elasticsearch.
     Returns an elastic search client for the mock corpus.
     """
     # add data from mock corpus
     corpus = Corpus.objects.get(name=mock_corpus)
-    es_index.create(es_client, corpus, add=False, clear=True, prod=True) # create ianalyzer-times-1 index
+    index = Index.objects.create(
+        name='times-test-1',
+        server=es_server,
+    )
+    job = IndexJob.objects.create(
+        corpus=corpus,
+    )
+    create_task = CreateIndexTask.objects.create(
+        job=job,
+        index=index,
+        delete_existing=True,
+    )
+    es_index.create(create_task)
     es_client.indices.create(index='times-test-2')
     es_client.indices.create(index='times-test-bla-3')
 

backend/es/es_alias.py

@@ -1,19 +1,30 @@
 #!/usr/bin/env python3
 import re
+from django.db import transaction
+from elasticsearch import Elasticsearch
+from typing import Generator
 
 from addcorpus.models import Corpus, CorpusConfiguration
-from ianalyzer.elasticsearch import elasticsearch
+from ianalyzer.elasticsearch import elasticsearch, server_for_corpus, client_from_config
+from es.models import Server, Index
+from es.sync import update_server_table_from_settings
+from indexing.models import IndexJob, DeleteIndexTask, RemoveAliasTask, AddAliasTask
 
 import logging
 logger = logging.getLogger('indexing')
 
-
-def alias(corpus: Corpus, clean=False):
+@transaction.atomic
+def create_alias_job(corpus: Corpus, clean=False) -> IndexJob:
     '''
-    Script to create, update and remove aliases from ES
+    Create a job to move the alias of a corpus to the index with the highest version
     '''
+
+    job = IndexJob.objects.create(corpus=corpus)
+
     corpus_config = corpus.configuration
     corpus_name = corpus.name
+    update_server_table_from_settings()
+    server = Server.objects.get(name=server_for_corpus(corpus_name))
     index_name = corpus_config.es_index
     index_alias = corpus_config.es_alias
     client = elasticsearch(corpus_name)
@@ -22,33 +33,53 @@ def alias(corpus: Corpus, clean=False):
     indices = client.indices.get(index='{}-*'.format(index_name))
     highest_version = get_highest_version_number(indices, alias)
 
-    actions = []
-
     for index_name, properties in indices.items():
         is_aliased = alias in properties['aliases'].keys()
         is_highest_version = extract_version(index_name, alias) == highest_version
+        index, _ = Index.objects.get_or_create(server=server, name=index_name)
 
         if not is_highest_version and clean:
-            logger.info('Removing index `{}`'.format(index_name))
-            # note that removing an index automatically removes alias
-            actions.append({'remove_index': {'index': index_name}})
+            DeleteIndexTask.objects.create(job=job, index=index)
 
         if not is_highest_version and is_aliased and not clean:
-            logger.info('Removing alias `{}` for index `{}`'.format(alias, index_name))
-            actions.append(
-                {'remove': {'index': index_name, 'alias': alias}})
+            RemoveAliasTask.objects.create(job=job, index=index, alias=alias)
 
         if is_highest_version and not is_aliased:
-            logger.info('Adding alias `{}` for index `{}`'.format(alias, index_name))
-            actions.append(
-                {'add': {'index': index_name, 'alias': alias}})
-        elif is_highest_version and is_aliased:
-            logger.info('Alias `{}` already exists for `{}`, skipping alias creation'.format(
-                alias, index_name))
+            AddAliasTask.objects.create(job=job, index=index, alias=alias)
+
+    return job
+
+
+def add_alias(task: AddAliasTask):
+    '''
+    Add an alias to an Elasticsearch index, as defined by an AddAliasTask
+    '''
+    client = task.client()
+    client.indices.put_alias(
+        index=task.index.name,
+        name=task.alias
+    )
 
-    if len(actions) > 0:
-        client.indices.update_aliases(actions=actions)
-    logger.info('Done updating aliases')
+
+def remove_alias(task: RemoveAliasTask):
+    '''
+    Remove an alias from an Elasticsearch index, as defined by a RemoveAliasTask
+    '''
+    client = task.client()
+    client.indices.delete_alias(
+        index=task.index.name,
+        name=task.alias
+    )
+
+
+def delete_index(task: DeleteIndexTask):
+    '''
+    Delete an Elasticsearch index, as defined by a DeleteIndexTask
+    '''
+    client = task.client()
+    client.indices.delete(
+        index=task.index.name,
+    )
 
 
 def get_current_index_name(corpus: CorpusConfiguration, client) -> str:
@@ -109,3 +140,14 @@ def get_highest_version_number(indices, current_index=None):
         return max([v for v in versions if v is not None])
     except:
         return 0
+
+
+def indices_with_alias(server: Server, alias: str) -> Generator[Index, None, None]:
+    client = client_from_config(server.configuration)
+    if client.indices.exists_alias(name=alias):
+        for index_name in client.indices.get_alias(name=alias):
+            aliased_index, _ = Index.objects.get_or_create(
+                server=server,
+                name=index_name,
+            )
+            yield aliased_index