Feature/ted 886

dags/operators/DagBatchPipelineOperator.py

@@ -19,7 +19,7 @@
 DEFAULT_NUBER_OF_CELERY_WORKERS = 144
 NOTICE_PROCESS_WORKFLOW_DAG_NAME = "notice_process_workflow"
 DEFAULT_START_WITH_TASK_ID = "notice_normalisation_pipeline"
-
+DEFAULT_PIPELINE_NAME_FOR_LOGS  = "unknown_pipeline_name"
 
 class BatchPipelineCallable(Protocol):
 
@@ -59,7 +59,11 @@ def execute(self, context: Any):
         mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
         notice_repository = NoticeRepository(mongodb_client=mongodb_client)
         processed_notice_ids = []
-        pipeline_name = self.notice_pipeline_callable.__name__
+        pipeline_name = DEFAULT_PIPELINE_NAME_FOR_LOGS
+        if self.notice_pipeline_callable:
+            pipeline_name = self.notice_pipeline_callable.__name__
+        elif self.batch_pipeline_callable:
+            pipeline_name = self.batch_pipeline_callable.__name__
         number_of_notices = len(notice_ids)
         batch_event_message = EventMessage(
             message=f"Batch processing for pipeline = [{pipeline_name}] with {number_of_notices} notices.",

infra/airflow-cluster/Dockerfile

@@ -21,5 +21,9 @@ RUN wget -c https://kumisystems.dl.sourceforge.net/project/saxon/Saxon-HE/10/Jav
 RUN cd .saxon && unzip SaxonHE10-6J.zip && rm -rf SaxonHE10-6J.zip
 
 
+RUN mkdir -p ./.limes
+RUN wget -c https://github.com/dice-group/LIMES/releases/download/1.7.9/limes.jar -P ./.limes
+
+
 RUN pip install --upgrade pip
 RUN pip install --no-cache-dir -r requirements.txt --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.4.1/constraints-no-providers-3.8.txt"

infra/airflow-cluster/docker-compose-worker.yaml

@@ -77,6 +77,7 @@ x-airflow-common:
     AIRFLOW_HOME: /opt/airflow
     RML_MAPPER_PATH: /opt/airflow/.rmlmapper/rmlmapper.jar
     XML_PROCESSOR_PATH: /opt/airflow/.saxon/saxon-he-10.6.jar
+    LIMES_ALIGNMENT_PATH: /opt/airflow/.limes/limes.jar
     DAG_LOGGER_CONFIG_HANDLERS: ${DAG_LOGGER_CONFIG_HANDLERS}
   volumes:
 #    - ./config/airflow.cfg:/opt/airflow/airflow.cfg

infra/airflow-cluster/docker-compose.yaml

@@ -79,6 +79,7 @@ x-airflow-common:
     AIRFLOW_HOME: /opt/airflow
     RML_MAPPER_PATH: /opt/airflow/.rmlmapper/rmlmapper.jar
     XML_PROCESSOR_PATH: /opt/airflow/.saxon/saxon-he-10.6.jar
+    LIMES_ALIGNMENT_PATH: /opt/airflow/.limes/limes.jar
     DAG_LOGGER_CONFIG_HANDLERS: ${DAG_LOGGER_CONFIG_HANDLERS}
   volumes:
 #    - ./config/airflow.cfg:/opt/airflow/airflow.cfg

infra/airflow/Dockerfile

@@ -20,6 +20,9 @@ RUN wget -c https://api.bitbucket.org/2.0/repositories/Dragos0000/rml-mapper/src
 RUN wget -c https://kumisystems.dl.sourceforge.net/project/saxon/Saxon-HE/10/Java/SaxonHE10-6J.zip -P .saxon/
 RUN cd .saxon && unzip SaxonHE10-6J.zip && rm -rf SaxonHE10-6J.zip
 
+RUN mkdir -p ./.limes
+RUN wget -c https://github.com/dice-group/LIMES/releases/download/1.7.9/limes.jar -P ./.limes
+
 
 RUN pip install --upgrade pip
 RUN pip install --no-cache-dir -r requirements.txt --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.4.1/constraints-no-providers-3.8.txt"

infra/airflow/docker-compose.yaml

@@ -69,6 +69,7 @@ x-airflow-common:
     AIRFLOW_HOME: /opt/airflow
     RML_MAPPER_PATH: /opt/airflow/.rmlmapper/rmlmapper.jar
     XML_PROCESSOR_PATH: /opt/airflow/.saxon/saxon-he-10.6.jar
+    LIMES_ALIGNMENT_PATH: /opt/airflow/.limes/limes.jar
     DAG_LOGGER_CONFIG_HANDLERS: ${DAG_LOGGER_CONFIG_HANDLERS}
   volumes:
 #    - ./config/airflow.cfg:/opt/airflow/airflow.cfg

tests/e2e/data_manager/test_fuseki_triple_store.py

@@ -1,4 +1,3 @@
-
 import pytest
 from ted_sws.data_manager.adapters.triple_store import FusekiAdapter, FusekiException
 from tests import TEST_DATA_PATH
@@ -19,27 +18,26 @@ def test_fuseki_triple_store_connection():
     with pytest.raises(FusekiException):
         triple_store.create_repository(repository_name=REPOSITORY_NAME)
 
-
-def test_fuseki_triple_store_add_file_to_repository():
-    triple_store = FusekiAdapter()
-    rdf_file_path = TEST_DATA_PATH / "example.ttl"
-    assert rdf_file_path.exists()
-    triple_store.add_file_to_repository(rdf_file_path,
-                                        repository_name="unknown_repository_123456677")
+    triple_store.delete_repository(repository_name=REPOSITORY_NAME)
 
 
 def test_fuseki_triple_store_get_sparql_endpoint(fuseki_triple_store):
-    sparql_endpoint = fuseki_triple_store.get_sparql_triple_store_endpoint(repository_name="unknown_repository_123456677")
+    if REPOSITORY_NAME not in fuseki_triple_store.list_repositories():
+        fuseki_triple_store.create_repository(repository_name=REPOSITORY_NAME)
+    rdf_file_path = TEST_DATA_PATH / "example.ttl"
+    assert rdf_file_path.exists()
+    fuseki_triple_store.add_file_to_repository(rdf_file_path,
+                                               repository_name=REPOSITORY_NAME)
+    sparql_endpoint = fuseki_triple_store.get_sparql_triple_store_endpoint(
+        repository_name=REPOSITORY_NAME)
     assert sparql_endpoint is not None
     df_query_result = sparql_endpoint.with_query(sparql_query=SPARQL_QUERY_TRIPLES).fetch_tabular()
     assert df_query_result is not None
     assert len(df_query_result) > 0
 
-    triple_store = FusekiAdapter()
-    triple_store.delete_repository(repository_name=REPOSITORY_NAME)
+    fuseki_triple_store.delete_repository(repository_name=REPOSITORY_NAME)
 
     with pytest.raises(FusekiException):
-        triple_store.delete_repository(repository_name=REPOSITORY_NAME)
-
-    assert REPOSITORY_NAME not in triple_store.list_repositories()
+        fuseki_triple_store.delete_repository(repository_name=REPOSITORY_NAME)
 
+    assert REPOSITORY_NAME not in fuseki_triple_store.list_repositories()