Merge pull request #208 from OP-TED/feature/TED-586

Feature/ted 586

Author: CaptainOfHacks

ted_sws/__init__.py

@@ -155,9 +155,21 @@ class TedAPIConfig:
     def TED_API_URL(self) -> str:
         return EnvConfigResolver().config_resolve()
 
+class FusekiConfig:
+    @property
+    def FUSEKI_ADMIN_USER(self) -> str:
+        return EnvConfigResolver().config_resolve()
+
+    @property
+    def FUSEKI_ADMIN_PASSWORD(self) -> str:
+        return EnvConfigResolver().config_resolve()
+
+    @property
+    def FUSEKI_ADMIN_HOST(self) -> str:
+        return EnvConfigResolver().config_resolve()
 
 class TedConfigResolver(MongoDBConfig, RMLMapperConfig, XMLProcessorConfig, ELKConfig, LoggingConfig,
-                        GitHubArtefacts, API, AllegroConfig, TedAPIConfig):
+                        GitHubArtefacts, API, AllegroConfig, TedAPIConfig, FusekiConfig):
     """
         This class resolve the secrets of the ted-sws project.
     """

ted_sws/data_manager/adapters/sparql_endpoint.py

@@ -104,6 +104,9 @@ def fetch_rdf(self) -> rdflib.Graph:
         :return:
         """
 
+    def add_data_to_repository(self, file_content, repository_name, mime_type):
+        pass
+
 
 class SPARQLTripleStoreEndpoint(TripleStoreEndpointABC):
 

ted_sws/data_manager/adapters/triple_store.py

@@ -1,18 +1,50 @@
 import abc
+from json import loads
+from pathlib import Path
+from typing import List, Union
+from urllib.parse import urljoin
 
+import rdflib.util
+import requests
 from franz.openrdf.repository import Repository
 from franz.openrdf.sail import AllegroGraphServer
+from requests.auth import HTTPBasicAuth
 
+from ted_sws import config
 from ted_sws.data_manager.adapters.sparql_endpoint import TripleStoreEndpointABC, SPARQLTripleStoreEndpoint
 
 
 class TripleStoreABC:
     @abc.abstractmethod
-    def add_data_to_repository(self, file_content: str, repository_name: str):
+    def create_repository(self, repository_name: str):
+        """
+        Method to create a repository
+        :param repository_name:
+        :return:
+        """
+
+    @abc.abstractmethod
+    def delete_repository(self, repository_name: str):
+        """
+        Method to delete a repository
+        :param repository_name:
+        :return:
+        """
+
+    @abc.abstractmethod
+    def list_repositories(self) -> List[str]:
+        """
+        Method to list all repositories
+        :return:
+        """
+
+    @abc.abstractmethod
+    def add_data_to_repository(self, file_content: Union[str, bytes, bytearray], mime_type: str, repository_name: str):
         """
         Method to add triples from a string
         :param file_content:
         :param repository_name:
+        :param mime_type:
         :return:
         """
 
@@ -31,7 +63,7 @@ def get_sparql_triple_store_endpoint(self, repository_name: str = None) -> Tripl
         """
 
 
-class AllegroGraphTripleStore(abc.ABC):
+class AllegroGraphTripleStore(TripleStoreABC):
     """
         This class is handling interactions with Allegro Graph triple store
         Note: If catalog name is not set, every operation will be executed at root level in the triple store
@@ -84,13 +116,15 @@ def list_repositories(self):
         """
         return self.allegro.openCatalog(name=self.catalog_name).listRepositories()
 
-    def add_data_to_repository(self, file_content: str, repository_name: str = None):
+    def add_data_to_repository(self, file_content: Union[str, bytes, bytearray], mime_type: str, repository_name: str = None):
         """
         Method to add triples from a string
         :param file_content:
         :param repository_name:
+        :param mime_type:
         :return:
         """
+
         repository = self._get_repository(repository_name=repository_name)
         repository.getConnection().addData(data=file_content)
 
@@ -111,3 +145,121 @@ def get_sparql_triple_store_endpoint(self, repository_name: str = None) -> Tripl
         endpoint_url = f"{self.host}/repositories/{repository_name}"
         sparql_endpoint = SPARQLTripleStoreEndpoint(endpoint_url=endpoint_url)
         return sparql_endpoint
+
+
+class FusekiException(Exception):
+    """
+        An exception when Fuseki server interaction has failed.
+    """
+
+
+RDF_MIME_TYPES = {
+    "turtle": "text/turtle",
+    "xml": "application/rdf+xml",
+    "json-ld": "application/ld+json",
+    "n3": "text/n3",
+    "nt": "application/n-triples",
+    "nquads": "application/n-quads",
+    "trig": "application/trig",
+}
+
+
+class FusekiAdapter(TripleStoreABC):
+
+    def __init__(self, host: str = config.FUSEKI_ADMIN_HOST,
+                 user: str = config.FUSEKI_ADMIN_USER,
+                 password: str = config.FUSEKI_ADMIN_PASSWORD):
+
+        self.host = host
+        self.user = user
+        self.password = password
+
+    def create_repository(self, repository_name: str):
+        """
+            Create the dataset for the Fuseki store
+        :param repository_name: The repository identifier. This should be short alphanumeric string uniquely
+        identifying the repository
+        :return: true if repository was created
+        """
+        if not repository_name:
+            raise ValueError('Repository name cannot be empty.')
+
+        data = {
+            'dbType': 'tdb',  # assuming that all repository are created persistent across restart
+            'dbName': repository_name
+        }
+
+        response = requests.post(urljoin(self.host, f"/$/datasets"),
+                                 auth=HTTPBasicAuth(self.user,
+                                                    self.password),
+                                 data=data)
+
+        if response.status_code == 409:
+            raise FusekiException('A repository with this name already exists.')
+
+    def add_data_to_repository(self, file_content: Union[str, bytes, bytearray], mime_type: str, repository_name: str):
+        url = urljoin(self.host, f"{repository_name}/data")
+        headers = {
+            "Content-Type": mime_type,
+        }
+        response = requests.post(url,
+                                 auth=HTTPBasicAuth(self.user, self.password),
+                                 data=file_content, headers=headers)
+        if response.status_code != 200:
+            raise FusekiException(f'Server refused to load the content with code {response.status_code}.')
+
+    def add_file_to_repository(self, file_path: Path, repository_name: str):
+        file_content = file_path.open('rb').read()
+        file_format = rdflib.util.guess_format(str(file_path))
+        mime_type = RDF_MIME_TYPES[file_format] if file_format in RDF_MIME_TYPES else "text/n3"
+        self.add_data_to_repository(file_content=file_content, mime_type=mime_type, repository_name=repository_name)
+
+    def delete_repository(self, repository_name: str):
+        """
+            Delete the repository from the Fuseki store
+        :param repository_name: The repository identifier. This should be short alphanumeric string uniquely
+        identifying the repository
+        :return: true if repository was deleted
+        """
+        response = requests.delete(urljoin(self.host, f"/$/datasets/{repository_name}"),
+                                   auth=HTTPBasicAuth(self.user,
+                                                      self.password))
+
+        if response.status_code == 404:
+            raise FusekiException('The repository doesn\'t exist.')
+
+    def list_repositories(self) -> list:
+        """
+            Get the list of the repository names from the Fuseki store.
+        :return: the list of the repository names
+        :rtype: list
+        """
+        response = requests.get(urljoin(self.host, "/$/datasets"),
+                                auth=HTTPBasicAuth(self.user,
+                                                   self.password))
+
+        if response.status_code != 200:
+            raise FusekiException(f"Fuseki server request ({response.url}) returned response {response.status_code}")
+
+        return self._select_repository_names_from_fuseki_response(response_text=response.text)
+
+    @staticmethod
+    def _select_repository_names_from_fuseki_response(response_text) -> list:
+        """
+            Helper method for digging for the list of repository.
+        :param response_text:
+        :return: list of the repository names
+        """
+        result = loads(response_text)
+        return [d_item['ds.name'][1:] for d_item in result['datasets']]
+
+    def get_sparql_triple_store_endpoint(self, repository_name: str = None) -> TripleStoreEndpointABC:
+        """
+            Helper method to create the url for querying.
+        :param repository_name: The dataset identifier. This should be short alphanumeric string uniquely
+        identifying the repository
+        :return: the query url
+        """
+        endpoint_url = urljoin(self.host, repository_name + "/sparql")
+        sparql_endpoint = SPARQLTripleStoreEndpoint(endpoint_url=endpoint_url)
+        return sparql_endpoint

ted_sws/notice_publisher_triple_store/services/load_transformed_notice_into_triple_store.py

@@ -2,7 +2,7 @@
     This module implements functionality to load a given notice into a triple store.
 """
 from ted_sws.data_manager.adapters.repository_abc import NoticeRepositoryABC
-from ted_sws.data_manager.adapters.triple_store import TripleStoreEndpointABC
+from ted_sws.data_manager.adapters.triple_store import TripleStoreEndpointABC, RDF_MIME_TYPES
 
 DEFAULT_NOTICE_REPOSITORY_NAME = "notices"
 
@@ -16,6 +16,6 @@ def load_notice_into_triple_store(notice_id: str, notice_repository: NoticeRepos
     notice = notice_repository.get(reference=notice_id)
     if notice is None:
         raise ValueError('Notice, with "%s" notice_id, was not found' % notice_id)
-
+    mime_type = RDF_MIME_TYPES
     rdf_manifestation_string = notice.rdf_manifestation.object_data
-    triple_store_repository.add_data_to_repository(file_content=rdf_manifestation_string, repository_name=repository_name)
+    triple_store_repository.add_data_to_repository(file_content=rdf_manifestation_string, repository_name=repository_name, mime_type=mime_type)

tests/e2e/conftest.py

@@ -2,7 +2,7 @@
 from pymongo import MongoClient
 
 from ted_sws import config
-from ted_sws.data_manager.adapters.triple_store import AllegroGraphTripleStore
+from ted_sws.data_manager.adapters.triple_store import AllegroGraphTripleStore, FusekiAdapter
 from tests import TEST_DATA_PATH
 
 
@@ -34,3 +34,8 @@ def path_ttl_file():
 @pytest.fixture
 def fake_mapping_suite_id() -> str:
     return "test_package"
+
+
+@pytest.fixture
+def fuseki_triple_store():
+    return FusekiAdapter(host=config.FUSEKI_ADMIN_HOST, user=config.FUSEKI_ADMIN_USER, password=config.FUSEKI_ADMIN_PASSWORD)

tests/e2e/data_manager/test_fuseki_triple_store.py

@@ -0,0 +1,48 @@
+from pathlib import Path
+
+import pytest
+from ted_sws import config
+from ted_sws.data_manager.adapters.triple_store import FusekiAdapter, FusekiException
+from tests import TEST_DATA_PATH
+from tests.fakes.fake_repository import FakeNoticeRepository
+
+REPOSITORY_NAME = "unknown_repository_123456677"
+SPARQL_QUERY_TRIPLES = "select * {?s ?p ?o} limit 10"
+
+
+def test_fuseki_triple_store_connection():
+    triple_store = FusekiAdapter()
+    if REPOSITORY_NAME in triple_store.list_repositories():
+        triple_store.delete_repository(repository_name=REPOSITORY_NAME)
+
+    triple_store.create_repository(repository_name=REPOSITORY_NAME)
+
+    assert REPOSITORY_NAME in triple_store.list_repositories()
+
+    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")
+
+
+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")
+    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)
+
+    with pytest.raises(FusekiException):
+        triple_store.delete_repository(repository_name=REPOSITORY_NAME)
+
+    assert REPOSITORY_NAME not in triple_store.list_repositories()
+

tests/e2e/mapping_suite_processor/conftest.py

@@ -2,8 +2,10 @@
 import pymongo
 import pytest
 
+from ted_sws.data_manager.adapters.triple_store import RDF_MIME_TYPES
 from tests import TEST_DATA_PATH
 
+
 @pytest.fixture
 @mongomock.patch(servers=(('server.example.com', 27017),))
 def fake_mongodb_client():
@@ -14,6 +16,7 @@ def fake_mongodb_client():
 def file_system_repository_path():
     return TEST_DATA_PATH / "notice_transformer" / "mapping_suite_processor_repository"
 
+
 @pytest.fixture
 def yarrrml_file_content():
     return """prefixes:
@@ -29,6 +32,7 @@ def yarrrml_file_content():
       - [ex:name, $(firstname)]
 """
 
+
 @pytest.fixture
 def rml_file_result():
     return """@prefix rr: <http://www.w3.org/ns/r2rml#>.
@@ -84,3 +88,6 @@ def package_folder_path():
     return TEST_DATA_PATH / "notice_validator" / "test_repository" / "test_package"
 
 
+@pytest.fixture
+def mime_type():
+    return RDF_MIME_TYPES

tests/e2e/mapping_suite_processor/test_allegro_triple_store.py

@@ -1,10 +1,10 @@
 
 
-def test_allegro_allegro_triple_store(ttl_file, path_ttl_file, allegro_triple_store):
+def test_allegro_triple_store(ttl_file, path_ttl_file, allegro_triple_store, mime_type):
     allegro_triple_store.create_repository(repository_name="testing")
     assert isinstance(allegro_triple_store.list_repositories(), list)
     assert "testing" in allegro_triple_store.list_repositories()
-    allegro_triple_store.add_data_to_repository(file_content=ttl_file, repository_name="testing")
+    allegro_triple_store.add_data_to_repository(file_content=ttl_file,mime_type=mime_type,repository_name="testing")
     allegro_triple_store.add_file_to_repository(file_path=path_ttl_file, repository_name="testing")
     assert allegro_triple_store._get_repository(repository_name="testing").getConnection().size() == 484
     allegro_triple_store.delete_repository(repository_name="testing")

tests/e2e/notice_publisher_triple_store/test_load_transformed_notice_into_triple_store.py

@@ -2,7 +2,6 @@
     load_notice_into_triple_store, DEFAULT_NOTICE_REPOSITORY_NAME
 from tests.fakes.fake_repository import FakeNoticeRepository
 
-
 SPARQL_QUERY_TRIPLES = "select * {?s ?p ?o}"
 SPARQL_QUERY_GRAPH = "SELECT ?g {  GRAPH ?g { ?s ?p ?o  } }"
 SPARQL_QUERY_FIXED_URI = "select * { <http://data.europa.eu/a4g/resource/ReviewerOrganisationIdentifier/2018-S-175-396207/de2507f9-ae25-37c8-809c-0109efe10669> ?p ?o .} "
@@ -19,19 +18,12 @@ def test_load_notice_into_triple_store(transformed_complete_notice, allegro_trip
 
     df_query_result = sparql_endpoint.with_query(sparql_query=SPARQL_QUERY_TRIPLES).fetch_tabular()
     assert df_query_result is not None
-    # assert that the graph in the triple store has more than 1 triple inside
-    if len(df_query_result) > 0:
-        assert True
+    assert len(df_query_result) > 0
 
-     # assert that at least one graph exists in the triple store
-    df_query_result = sparql_endpoint.with_query(sparql_query=SPARQL_QUERY_GRAPH).fetch_tabular()
+    df_query_result = sparql_endpoint.with_query(sparql_query=SPARQL_QUERY_GRAPH).fetch_tree()
     assert df_query_result is not None
-    if len(df_query_result) > 0:
-        assert True
+    assert len(df_query_result) > 0
 
-     # assert that the there is an epo:SPARQL_QUERY_FIXED_URI object that has the value equal to the notice ID
     df_query_result = sparql_endpoint.with_query(sparql_query=SPARQL_QUERY_FIXED_URI).fetch_tabular()
     assert df_query_result is not None
-    print(df_query_result)
-    if len(df_query_result) > 0:
-        assert True
+    assert len(df_query_result) > 0