Add POST/ASK interactions to knowledge base and client

examples/ask_interaction.py

@@ -0,0 +1,48 @@
+from shared import get_example_logger
+
+from src import KnowledgeBase
+from src.ke.models import BindingModel, Literal, Uri
+
+EXAMPLE_NAME = "ask-interaction"
+logger = get_example_logger(EXAMPLE_NAME)
+
+kb = KnowledgeBase(
+    id="http://example.org/knowledge-mapper/ask-interaction#kb",
+    name="ask-interaction-kb",
+    description="An example KB that demonstrates handling an ASK KI.",
+    ke_url="http://localhost:8280/rest",
+)
+
+
+class PersonBinding(BindingModel):
+    person: Uri
+    name: Literal[str]
+    age: Literal[int]
+
+
+kb.ask_ki(
+    name="ask-ki",
+    graph_pattern="""
+        ?person a ex:Person ;
+            ex:hasName ?name ;
+            ex:hasAge ?age .
+    """,
+    binding_model=PersonBinding,
+    prefixes={"ex": "http://example.org/knowledge-mapper/ask-interaction#"},
+)
+
+if __name__ == "__main__":
+    kb.register()
+    logger.info("KB registered.")
+    result = kb.ask(
+        [
+            {
+                "person": "http://example.org/knowledge-mapper/ask-interaction#person1",
+            }
+        ],
+        "ask-ki",
+    )
+    logger.info(f"Received result from ASK KI: {result}")
+
+    kb.unregister()
+    logger.info("KB unregistered.")

examples/compose.yaml

@@ -1,9 +1,21 @@
 services:
+  knowledge-directory:
+    image: ghcr.io/tno/knowledge-engine/knowledge-directory:1.3.2
+
   ker-examples:
     image: ghcr.io/tno/knowledge-engine/smart-connector:1.3.2
     ports:
-      - 8081:8081
       - 8280:8280
     environment: 
       KE_RUNTIME_EXPOSED_URL: http://ker-examples:8081
       KE_RUNTIME_PORT: 8081
+      KD_URL: http://knowledge-directory:8282
+
+  ker-testing:
+    image: ghcr.io/tno/knowledge-engine/smart-connector:1.3.2
+    ports:
+      - 8281:8280
+    environment: 
+      KE_RUNTIME_EXPOSED_URL: http://ker-testing:8081
+      KE_RUNTIME_PORT: 8081
+      KD_URL: http://knowledge-directory:8282

examples/post_measurement.py

@@ -0,0 +1,81 @@
+import time
+from datetime import datetime
+from uuid import uuid4
+
+from rdflib import URIRef
+from shared import get_example_logger
+
+from src.ke.models import (
+    BindingModel,
+    Literal,
+    Uri,
+)
+from src.knowledge_base import KnowledgeBase
+
+EXAMPLE_NAME = "post-measurement"
+logger = get_example_logger(EXAMPLE_NAME)
+
+kb = KnowledgeBase(
+    id="http://example.org/knowledge-mapper/post-measurement#kb",
+    name="post-measurement-kb",
+    description="An example KB that demonstrates handling a POST KI for posting a new "
+    "measurement.",
+    ke_url="http://localhost:8280/rest",
+)
+
+
+class MeasurementBinding(BindingModel):
+    measurement: Uri
+    value: Literal[float]
+    unit: Uri
+    time: Literal[datetime]
+
+
+class ResultBinding(BindingModel):
+    measurement: Uri
+    kb: Uri
+
+
+kb.post_ki(
+    name="post-measurement-ki",
+    argument_graph_pattern="""
+        ?measurement a ex:Measurement ;
+            ex:hasValue ?value ;
+            ex:hasUnit ?unit ;
+            ex:hasTime ?time .
+    """,
+    result_graph_pattern="""
+        ?measurement a ex:Measurement ;
+            ex:storedBy ?kb ;
+    """,
+    prefixes={"ex": "http://example.org/knowledge-mapper/post-measurement#"},
+    result_binding_model=ResultBinding,
+    argument_binding_model=MeasurementBinding,
+)
+
+
+if __name__ == "__main__":
+    kb.register()
+    logger.info("KB registered.")
+    time.sleep(
+        5
+    )  # Sleep for a bit to allow time for testing the POST KI with an external client
+    logger.info("Posting...")
+    result_bindings = kb.post(
+        [
+            MeasurementBinding(
+                measurement=URIRef(
+                    f"http://example.org/knowledge-mapper/post-measurement#measurement-{uuid4()}"
+                ),
+                value=99.9,
+                unit=URIRef(
+                    "http://example.org/knowledge-mapper/post-measurement#Percent"
+                ),
+                time=datetime.now(),
+            )
+        ],
+        "post-measurement-ki",
+    )
+    logger.info(f"Received result bindings: {result_bindings}")
+    kb.unregister()
+    logger.info("KB unregistered.")

examples/testing/kb.py

@@ -0,0 +1,113 @@
+import sys
+from pathlib import Path
+from time import sleep
+from typing import cast
+
+from rdflib import URIRef
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from shared import get_example_logger
+
+from src import KnowledgeBase
+from src.ke.models import BindingModel, Literal, Uri
+
+EXAMPLE_NAME = "testing"
+logger = get_example_logger(EXAMPLE_NAME)
+
+kb = KnowledgeBase(
+    id="http://example.org/knowledge-mapper/testing#kb",
+    name="testing-kb",
+    description="An example KB that demonstrates testing the KB.",
+    ke_url="http://localhost:8280/rest",
+)
+
+kb.ask_ki(
+    name="ask-ki-no-binding-model",
+    graph_pattern="""
+        ?s a ex:TestSubject ;
+            ex:hasValue ?value .
+    """,
+    prefixes={"ex": "http://example.org/knowledge-mapper/testing#"},
+)
+
+
+class TestBinding(BindingModel):
+    s: Uri
+    value: Literal[str]
+
+
+kb.ask_ki(
+    name="ask-ki-with-binding-model",
+    graph_pattern="""
+        ?s a ex:TestSubject ;
+            ex:hasValue ?value .
+    """,
+    binding_model=TestBinding,
+    prefixes={"ex": "http://example.org/knowledge-mapper/testing#"},
+)
+
+
+def ask_for_values_of_subject(subject_name: str) -> list[str]:
+    result_binding_set: list[TestBinding] = kb.ask(
+        [
+            TestBinding(
+                s=URIRef(f"http://example.org/knowledge-mapper/testing#{subject_name}"),
+                value=None,
+            )
+        ],
+        "ask-ki-with-binding-model",
+    )  # pyright: ignore[reportAssignmentType]
+    return (
+        [str(binding.value) for binding in result_binding_set]
+        if result_binding_set
+        else []
+    )
+
+
+class ResultBinding(BindingModel):
+    s: Uri
+    other: Uri
+
+
+kb.post_ki(
+    name="post-ki",
+    argument_graph_pattern="""
+        ?s a ex:TestSubject ;
+            ex:hasValue ?value .
+    """,
+    result_graph_pattern="""
+        ?s a ex:TestSubject ;
+            ex:storedBy ?other .
+    """,
+    argument_binding_model=TestBinding,
+    result_binding_model=ResultBinding,
+    prefixes={"ex": "http://example.org/knowledge-mapper/testing#"},
+)
+
+
+def repeat_value_post(value: str, iterations: int) -> list[URIRef]:
+    result_binding_set: list[ResultBinding] = []
+    for i in range(iterations):
+        result_binding_set.extend(
+            kb.post(
+                [
+                    TestBinding(
+                        s=URIRef(
+                            f"http://example.org/knowledge-mapper/testing#Subject-{i}"
+                        ),
+                        value=value,
+                    )
+                ],
+                "post-ki",
+            )  # type: ignore
+        )
+        sleep(1)
+    return [cast(URIRef, binding.other) for binding in result_binding_set]
+
+
+if __name__ == "__main__":
+    logger.info(
+        "This KB demonstrates testing, and is not meant to be run as a standalone "
+        "example."
+    )

examples/testing/test_kb.py

@@ -0,0 +1,118 @@
+import pytest
+from rdflib import URIRef
+
+from src.ke.testing import TestClient
+
+# Import the Knowledge Base that you would like to test, along with any relevant binding
+# models.
+from .kb import TestBinding, ask_for_values_of_subject, kb, repeat_value_post
+
+# In your tests you likely want to use the TestClient to mock results from the KE.
+# A Knowledge Base instance is initialized with a real Client that makes HTTP requests
+# to the KE, so its important to replace it with a TestClient
+test_client = TestClient(fake_url="http://fake-ke")
+kb.client = test_client
+# Here the KB and its interactions are registered with the TestClient, which always
+# succeeds. This registration is necessary for the KB to be able to execute
+# interactions in the tests.
+kb.register()
+
+
+@pytest.fixture()
+def client():
+    return test_client
+
+
+# In a test you can do any ASK interaction that is registered.
+# The TestClient will return an empty result binding set by default, disregarding the
+# input.
+def test_ask_ki_no_resuls():
+    result_binding_set = kb.ask([], "ask-ki-no-binding-model")
+    assert result_binding_set == []
+
+
+# You likely want to mock result binding sets, which can be done using the TestClient as
+# in this test. The mocked result is returned when the ASK interaction is executed,
+# disregarding the input.
+def test_ask_ki_with_result(client: TestClient):
+    client.mock_result_binding_set(
+        "ask-ki-no-binding-model",
+        [
+            {
+                "s": "<http://example.org/knowledge-mapper/testing#Subject>",
+                "value": "test value",
+            }
+        ],
+    )
+    result_binding_set = kb.ask([], "ask-ki-no-binding-model")
+    assert result_binding_set == [
+        {
+            "s": "<http://example.org/knowledge-mapper/testing#Subject>",
+            "value": "test value",
+        }
+    ]
+
+
+# This is a little more useful when you have a binding model, testing the correctness of
+# the binding model according to the graph pattern. One test per interaction like this
+# per interaction is probably a good idea, to isolate issues with the binding model from
+# other issues.
+def test_ask_ki_with_binding_model(client: TestClient):
+    client.mock_result_binding_set(
+        "ask-ki-with-binding-model",
+        [
+            {
+                "s": "<http://example.org/knowledge-mapper/testing#Subject>",
+                "value": "test value",
+            }
+        ],
+    )
+
+    result_binding_set = kb.ask(
+        [
+            TestBinding(
+                s=URIRef("http://example.org/knowledge-mapper/testing#Subject"),
+                value=None,
+            )
+        ],
+        "ask-ki-with-binding-model",
+    )
+    assert result_binding_set == [
+        TestBinding(
+            s=URIRef("http://example.org/knowledge-mapper/testing#Subject"),
+            value="test value",
+        )
+    ]
+
+
+# However, most likely you will want to test the logic around interactions, where you
+# might want to mock different results for different inputs.
+def test_function_containing_ask(client: TestClient):
+    client.mock_result_binding_set(
+        ki_name="ask-ki-with-binding-model",
+        binding_set=[
+            TestBinding(
+                s=URIRef("http://example.org/knowledge-mapper/testing#Subject"),
+                value="test value",
+            ).model_dump(),
+        ],
+    )
+
+    result = ask_for_values_of_subject("Subject")
+    assert result == ["test value"]
+
+
+# Similar approaches can be taken for POST interactions.
+def test_function_containing_post(client: TestClient):
+    client.mock_result_binding_set(
+        ki_name="post-ki",
+        binding_set=[
+            {
+                "s": "<http://example.org/knowledge-mapper/testing#Subject>",
+                "other": "<http://example.org/knowledge-mapper/testing#Other>",
+            }
+        ],
+    )
+
+    result = repeat_value_post("test value", 1)
+    assert result == [URIRef("http://example.org/knowledge-mapper/testing#Other")]