Create KnowledgeBaseBuilder and extract settings from knowledge base class

examples/basic.py

@@ -1,6 +1,6 @@
 from shared import get_example_logger
 
-from src.knowledge_base import KnowledgeBase
+from src.kb.knowledge_base import KnowledgeBase
 
 EXAMPLE_NAME = "basic"
 logger = get_example_logger(EXAMPLE_NAME)

examples/binding_models.py

@@ -3,14 +3,14 @@
 from rdflib import URIRef
 from shared import get_example_logger
 
+from src.kb.knowledge_base import KnowledgeBase
 from src.ke.models import (
     BindingModel,
     BindingSet,
     KnowledgeInteractionInfo,
     Literal,
     Uri,
 )
-from src.knowledge_base import KnowledgeBase
 
 EXAMPLE_NAME = "binding-models"
 logger = get_example_logger(EXAMPLE_NAME)

examples/custom-settings/custom_settings.py

@@ -33,6 +33,9 @@
 logger = get_example_logger(EXAMPLE_NAME)
 
 
+# Subclass KnowledgeBaseSettings to add application-specific settings alongside the
+# standard KB configuration fields. The model_config points to the YAML file for this
+# example and enables CLI argument parsing.
 class AppSettings(KnowledgeBaseSettings):
     model_config = SettingsConfigDict(
         yaml_file="custom-settings/settings.yaml",
@@ -45,6 +48,9 @@ class AppSettings(KnowledgeBaseSettings):
     db_port: int = 5432
     debug: bool = False
 
+    # Override settings_customise_sources to prepend CliSettingsSource so that CLI
+    # arguments take the highest priority, followed by the sources defined in the base
+    # class (env vars, config file, defaults).
     @classmethod
     def settings_customise_sources(cls, settings_cls, **kwargs):  # type: ignore
         return (
@@ -53,27 +59,38 @@ def settings_customise_sources(cls, settings_cls, **kwargs):  # type: ignore
         )
 
 
+# Instantiate AppSettings to load configuration from all sources at once (CLI args,
+# env vars, YAML file, and field defaults), in priority order.
 settings = AppSettings()  # type: ignore
-kb = KnowledgeBase.from_settings(settings)
-kb.ki_from_settings_with_default_handler("ask-from-settings")
-kb.ki_from_settings_with_default_handler("post-from-settings")
 
 
-@kb.ki_from_settings("answer-from-settings")
 def example_answer_from_settings(
     binding_set: BindingSet, info: KnowledgeInteractionInfo
 ) -> BindingSet:
     return binding_set
 
 
-@kb.ki_from_settings("react-from-settings")
 def example_react_from_settings(
     binding_set: BindingSet, info: KnowledgeInteractionInfo
 ) -> BindingSet:
     return binding_set
 
 
+# Use KnowledgeBase.from_settings to build the KB from the settings object instead of
+# passing explicit constructor arguments. KIs defined in the settings YAML are
+# registered automatically; use .handler() to attach a handler function to each of
+# the ANSWER/REACT KIs by name. This is required, otherwise .build() will fail.
+kb = (
+    KnowledgeBase.from_settings(settings)
+    .handler("answer-from-settings", example_answer_from_settings)
+    .handler("react-from-settings", example_react_from_settings)
+    .build()
+)
+
+
 if __name__ == "__main__":
+    # After building, we can see that KI contexts are accessible, and so
+    # are the settings from configuration sources.
     ask_ctx = kb.ki_registry["ask-from-settings"]
     post_ctx = kb.ki_registry["post-from-settings"]
 

examples/post_measurement.py

@@ -5,12 +5,12 @@
 from rdflib import URIRef
 from shared import get_example_logger
 
+from src.kb.knowledge_base import KnowledgeBase
 from src.ke.models import (
     BindingModel,
     Literal,
     Uri,
 )
-from src.knowledge_base import KnowledgeBase
 
 EXAMPLE_NAME = "post-measurement"
 logger = get_example_logger(EXAMPLE_NAME)

src/__init__.py

@@ -1,6 +1,7 @@
 import logging
 
-from .knowledge_base import KnowledgeBase
+from .kb.builder import KnowledgeBaseBuilder
+from .kb.knowledge_base import KnowledgeBase
 from .settings import KnowledgeBaseSettings
 
 __version__ = "0.1.0a0"

src/kb/builder.py

@@ -0,0 +1,102 @@
+from typing import Self
+
+from ..ke.models import KiTypes
+from ..knowledge_interaction import Handler, KnowledgeInteractionContext
+from ..settings import KnowledgeBaseSettings
+from .knowledge_base import KnowledgeBase
+
+
+class KnowledgeBaseBuilder:
+    """Builds a :class:`KnowledgeBase` from a :class:`~.settings.KnowledgeBaseSettings`
+    instance.
+
+    Returned by :meth:`KnowledgeBase.from_settings`. Attach handlers for incoming
+    (ANSWER/REACT) knowledge interactions via :meth:`handler`, then call :meth:`build`
+    to obtain the configured :class:`KnowledgeBase`. ASK and POST KIs defined in
+    settings are registered automatically — no explicit call needed.
+
+    Example::
+
+        builder = KnowledgeBase.from_settings(settings)
+
+        @builder.handler("my-answer-ki")
+        def my_handler(binding_set, info):
+            return binding_set
+
+        kb = builder.build()
+        kb.connect()
+        kb.register()
+        kb.start_handling_loop()
+    """
+
+    def __init__(self, settings: KnowledgeBaseSettings) -> None:
+        self._settings = settings
+        self._kb = KnowledgeBase(
+            id=settings.knowledge_base.id,
+            name=settings.knowledge_base.name,
+            description=settings.knowledge_base.description,
+            ke_url=settings.knowledge_engine_endpoint,
+        )
+        self._unhandled_incoming: set[str] = {
+            ki.name
+            for ki in settings.knowledge_interactions
+            if ki.type in (KiTypes.ANSWER, KiTypes.REACT)
+        }
+
+    def handler(self, ki_name: str, func: Handler) -> Self:
+        """Attach *func* as the handler for the ANSWER or REACT KI named *ki_name*.
+
+        Args:
+            ki_name: Name of the KI as declared in settings.
+            func: Handler callable; receives a binding set and
+                :class:`~.ke.models.KnowledgeInteractionInfo` and returns a binding set.
+
+        Raises:
+            ValueError: If *ki_name* is not declared in settings, or if the KI is of
+                type ASK or POST (outgoing KIs do not take handlers; they are registered
+                automatically).
+        """
+        try:
+            info = self._settings.get_configured_interaction(ki_name)
+        except ValueError as err:
+            raise ValueError(f"KI named '{ki_name}' not found in settings.") from err
+
+        if info.type not in (KiTypes.ANSWER, KiTypes.REACT):
+            raise ValueError(
+                f"KI '{ki_name}' is of type {info.type}. Only ANSWER and REACT KIs "
+                f"accept a handler; ASK and POST KIs are registered automatically."
+            )
+
+        self._kb._register_ki_decorator(info=info, defer_ke_registration=True)(func)
+        self._unhandled_incoming.discard(ki_name)
+        return self
+
+    def build(self) -> KnowledgeBase:
+        """Return the configured :class:`KnowledgeBase`.
+
+        All ASK and POST KIs from settings are registered at this point. ANSWER and
+        REACT KIs must have had their handlers attached via :meth:`handler` before
+        calling ``build()``.
+
+        Raises:
+            ValueError: If any ANSWER or REACT KI from settings has no handler.
+        """
+        if self._unhandled_incoming:
+            names = ", ".join(sorted(self._unhandled_incoming))
+            raise ValueError(
+                f"The following ANSWER/REACT KIs from settings have no handler "
+                f"attached: {names}. Call builder.handler(ki_name, func) for each "
+                f"before building."
+            )
+
+        for ki in self._settings.knowledge_interactions:
+            if ki.type in (KiTypes.ASK, KiTypes.POST):
+                self._kb.register_ki(
+                    KnowledgeInteractionContext(
+                        info=ki,
+                        handler=None,
+                    ),
+                    defer_ke_registration=True,
+                )
+
+        return self._kb

src/knowledge_base.py → src/kb/knowledge_base.py

@@ -1,13 +1,15 @@
+from __future__ import annotations
+
 import logging
 from collections.abc import Callable, Sequence
 from enum import StrEnum
 from functools import wraps
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
-from .ke import Client
-from .ke.client import ClientProtocol, PollResult
-from .ke.errors import KnowledgeEngineNotAvailableError
-from .ke.models import (
+from ..ke import Client
+from ..ke.client import ClientProtocol, PollResult
+from ..ke.errors import KnowledgeEngineNotAvailableError
+from ..ke.models import (
     AskAnswerInteractionInfo,
     BindingModel,
     BindingSet,
@@ -16,12 +18,15 @@
     KnowledgeInteractionInfo,
     PostReactInteractionInfo,
 )
-from .knowledge_interaction import (
+from ..knowledge_interaction import (
     Handler,
     KnowledgeInteractionContext,
     KnowledgeInteractionStatus,
 )
-from .settings import KnowledgeBaseSettings
+
+if TYPE_CHECKING:
+    from ..settings import KnowledgeBaseSettings
+    from .builder import KnowledgeBaseBuilder
 
 logger = logging.getLogger(__name__)
 
@@ -46,21 +51,20 @@ def __init__(self, id: str, name: str, description: str, ke_url: str):
             name=name,
             description=description,
         )
-        self._build_settings: KnowledgeBaseSettings | None = None
 
     @classmethod
-    def from_settings(cls, settings: KnowledgeBaseSettings) -> "KnowledgeBase":
-        """Create a :class:`KnowledgeBase` from a
+    def from_settings(cls, settings: KnowledgeBaseSettings) -> KnowledgeBaseBuilder:
+        """Create a :class:`~.knowledge_base_builder.KnowledgeBaseBuilder` from a
         :class:`~.settings.KnowledgeBaseSettings` instance (or a subclass thereof).
+
+        Attach handlers for incoming KIs via the builder's
+        :meth:`~.knowledge_base_builder.KnowledgeBaseBuilder.handler` method, then call
+        :meth:`~.knowledge_base_builder.KnowledgeBaseBuilder.build` to obtain the
+        configured :class:`KnowledgeBase`.
         """
-        kb = cls(
-            id=settings.knowledge_base.id,
-            name=settings.knowledge_base.name,
-            description=settings.knowledge_base.description,
-            ke_url=settings.knowledge_engine_endpoint,
-        )
-        kb._build_settings = settings
-        return kb
+        from .builder import KnowledgeBaseBuilder
+
+        return KnowledgeBaseBuilder(settings)
 
     def connect(self) -> None:
         """Checks whether the KE runtime is available and raises an exception if not.
@@ -375,91 +379,6 @@ def react_ki(
             defer_ke_registration=defer_ke_registration,
         )
 
-    def ki_from_settings(
-        self, ki_name: str, defer_ke_registration: bool = True
-    ) -> Callable[[Handler], Handler]:
-        """Return a decorator that registers the decorated function as a KI
-        handler with info from the KB settings.
-
-        Raises:
-            ValueError: If no settings are found or ``ki_name`` is not found in the
-            settings, or if registration constraints are violated.
-            SmartConnectorNotFoundError: Propagated from ``register_ki`` when contacting
-              the KE runtime.
-            UnexpectedHttpResponseError: Propagated from ``register_ki`` when contacting
-              the KE runtime.
-        """
-        if not self._build_settings:
-            raise ValueError(
-                "Cannot register KI from settings because the KB was not built from "
-                "settings. Please build the KB using KnowledgeBase.from_settings() "
-                "with a KnowledgeBaseSettings that includes the desired KI info."
-            )
-        try:
-            info = self._build_settings.get_configured_interaction(ki_name)
-        except ValueError as err:
-            raise ValueError(
-                f"KI named '{ki_name}' not found in KB settings. Please ensure the "
-                f"settings include a KI with this name."
-            ) from err
-
-        return self._register_ki_decorator(
-            info=info,
-            defer_ke_registration=defer_ke_registration,
-        )
-
-    def ki_from_settings_with_default_handler(
-        self, ki_name: str, defer_ke_registration: bool = True
-    ) -> None:
-        """Register a KI that was defined in the settings of a KB. Only applicable to
-        KIs of type ASK or POST, which will be registered with the default ASK and POST
-        handlers, respectively.
-
-        .. warning::
-            The default ASK and POST handlers are not yet implemented. The KI will be
-            registered successfully, but invoking it will raise
-            :exc:`NotImplementedError`. Use :meth:`ki_from_settings` with a custom
-            handler instead.
-
-        Raises:
-            ValueError: If no settings are found or ``ki_name`` is not found in the
-            settings, if the KI type is not ASK or POST, or if registration constraints
-            are violated.
-            SmartConnectorNotFoundError: Propagated from ``register_ki`` when contacting
-              the KE runtime.
-            UnexpectedHttpResponseError: Propagated from ``register_ki`` when contacting
-              the KE runtime.
-        """
-        if not self._build_settings:
-            raise ValueError(
-                "Cannot register KI from settings because the KB was not built from "
-                "settings. Please build the KB using KnowledgeBase.from_settings() with"
-                " a KnowledgeBaseSettings that includes the desired KI info."
-            )
-
-        try:
-            info = self._build_settings.get_configured_interaction(ki_name)
-        except ValueError as err:
-            raise ValueError(
-                f"KI named '{ki_name}' not found in KB settings. Please ensure the "
-                f"settings include a KI with this name."
-            ) from err
-
-        if not (info.type == KiTypes.ASK or info.type == KiTypes.POST):
-            raise ValueError(
-                f"KI named '{ki_name}' in settings must be of type ASK or POST to use "
-                f"the default handler registration method."
-            )
-
-        self.register_ki(
-            KnowledgeInteractionContext(
-                info=info,
-                handler=None,
-            ),
-            defer_ke_registration=defer_ke_registration,
-        )
-        return
-
     def call(self, binding_set: BindingSet, ki_name: str) -> BindingSet:
         """Invoke the handler of a registered KI by its name.
 

src/legacy/__main__.py

@@ -7,7 +7,7 @@
 import time
 import signal
 import requests.exceptions
-from src.knowledge_base import KnowledgeBaseUnregistered
+from src.kb.knowledge_base import KnowledgeBaseUnregistered
 
 from src.knowledge_mapper import KnowledgeMapper
 from src.auth.sql_auth import SqlAuth

src/legacy/data_source.py

@@ -1,4 +1,4 @@
-from src.knowledge_base import KnowledgeBase
+from src.kb.knowledge_base import KnowledgeBase
 
 class DataSource:
     def test(self):