refactored effector to work similar to the processing pipeline, accepts a list of deref handlers, made public key overwrite warning more specific, gave effector deref handlers access to handler context, readded post init injection to effector

Author: lukvmil

src/koi_net/assembler.py

@@ -41,7 +41,7 @@ class NodeContainer(Protocol):
     """Dummy 'shape' for node containers built by assembler."""
     entrypoint = EntryPoint
 
-class NodeAssembler(metaclass=BuildOrderer):    
+class NodeAssembler(metaclass=BuildOrderer):
     def __new__(self) -> NodeContainer:
         """Returns assembled node container."""
         return self._build()

src/koi_net/config/core.py

@@ -101,7 +101,9 @@ def generate_rid_cascade(self):
         log.debug(f"Node RID set to {self.koi_net.node_rid}")
 
         if self.koi_net.node_profile.public_key != pub_key.to_der():
+            if self.koi_net.node_profile.public_key:
+                log.warning("New private key overwriting old public key!")
+            
             self.koi_net.node_profile.public_key = pub_key.to_der()
-            log.warning("New private key overwrote old public key!")
 
         return self

src/koi_net/core.py

@@ -50,6 +50,7 @@ class BaseNode(NodeAssembler):
         basic_network_output_filter,
         forget_edge_on_node_deletion
     ]
+    deref_handlers = []
     cache = lambda config: Cache(
         directory_path=config.koi_net.cache_directory_path)
     identity = NodeIdentity
@@ -61,8 +62,8 @@ class BaseNode(NodeAssembler):
     sync_manager = SyncManager
     response_handler = ResponseHandler
     resolver = NetworkResolver
-    effector = Effector
     handler_context = HandlerContext
+    effector = Effector
     pipeline = KnowledgePipeline
     kobj_worker = KnowledgeProcessingWorker
     event_worker = EventProcessingWorker

src/koi_net/effector.py

@@ -1,27 +1,33 @@
-import structlog
+from dataclasses import dataclass
 from typing import Callable
 from enum import StrEnum
+
+import structlog
 from rid_lib.ext import Cache, Bundle
 from rid_lib.core import RID, RIDType
 from rid_lib.types import KoiNetNode
+
+from .processor.context import HandlerContext
 from .network.resolver import NetworkResolver
 from .processor.kobj_queue import KobjQueue
-from .identity import NodeIdentity
 
 log = structlog.stdlib.get_logger()
 
-
-class ActionContext:
-    """Provides action handlers access to other subsystems."""
+@dataclass
+class DerefHandler:
+    func: Callable[[HandlerContext, RID], Bundle | None]
+    rid_types: tuple[RIDType]
 
-    identity: NodeIdentity
-
-    def __init__(
-        self,
-        identity: NodeIdentity,
-    ):
-        self.identity = identity
+    def __call__(self, ctx: HandlerContext, rid: RID) -> Bundle | None:
+        return self.func(ctx, rid)
 
+    @classmethod
+    def create(cls, rid_types: tuple[RIDType]):
+        def decorator(func: Callable) -> DerefHandler:
+            handler = cls(func, rid_types)
+            return handler
+        return decorator
+
 
 class BundleSource(StrEnum):
     CACHE = "CACHE"
@@ -33,50 +39,23 @@ class Effector:
     cache: Cache
     resolver: NetworkResolver
     kobj_queue: KobjQueue
-    action_context: ActionContext
-    _action_table: dict[
-        type[RID], 
-        Callable[
-            [ActionContext, RID], 
-            Bundle | None
-        ]
-    ] = dict()
+    handler_context: HandlerContext
 
     def __init__(
         self, 
         cache: Cache,
         resolver: NetworkResolver,
         kobj_queue: KobjQueue,
-        identity: NodeIdentity
+        handler_context: HandlerContext,
+        deref_handlers: list[DerefHandler]
     ):
         self.cache = cache
         self.resolver = resolver
         self.kobj_queue = kobj_queue
-        self.action_context = ActionContext(identity)
-        self._action_table = self.__class__._action_table.copy()
-    
-    @classmethod
-    def register_default_action(cls, rid_type: RIDType):
-        def decorator(func: Callable) -> Callable:
-            cls._action_table[rid_type] = func
-            return func
-        return decorator
-        
-    def register_action(self, rid_type: RIDType):
-        """Registers a new dereference action for an RID type.
+        self.handler_context = handler_context
+        self.deref_handlers = deref_handlers
 
-        Example:
-            This function should be used as a decorator on an action function::
-            
-                @node.register_action(KoiNetNode)
-                def deref_koi_net_node(ctx: ActionContext, rid: KoiNetNode):
-                    # return a Bundle or None
-                    return
-        """
-        def decorator(func: Callable) -> Callable:
-            self._action_table[rid_type] = func
-            return func
-        return decorator
+        self.handler_context.set_effector(self)
 
     def _try_cache(self, rid: RID) -> tuple[Bundle, BundleSource] | None:
         bundle = self.cache.read(rid)
@@ -87,26 +66,27 @@ def _try_cache(self, rid: RID) -> tuple[Bundle, BundleSource] | None:
         else:
             log.debug("Cache miss")
             return None
-                    
+            
     def _try_action(self, rid: RID) -> tuple[Bundle, BundleSource] | None:
-        if type(rid) not in self._action_table:
-            log.debug("No action available")
+        action = None
+        for handler in self.deref_handlers:
+            if type(rid) not in handler.rid_types:
+                continue
+            action = handler
+            break
+        
+        if not action:
+            log.debug("No action found")
             return None
 
-        log.debug("Action available")
-        func = self._action_table[type(rid)]
-        bundle = func(
-            ctx=self.action_context, 
-            rid=rid
-        )
+        bundle = action(ctx=self.handler_context, rid=rid)
 
         if bundle:
             log.debug("Action hit")
             return bundle, BundleSource.ACTION
         else:
             log.debug("Action miss")
             return None
-
 
     def _try_network(self, rid: RID) -> tuple[Bundle, KoiNetNode] | None:
         bundle, source = self.resolver.fetch_remote_bundle(rid)
@@ -118,13 +98,13 @@ def _try_network(self, rid: RID) -> tuple[Bundle, KoiNetNode] | None:
             log.debug("Network miss")
             return None
 
-    
     def deref(
         self, 
         rid: RID,
         refresh_cache: bool = False,
         use_network: bool = False,
-        handle_result: bool = True
+        handle_result: bool = True,
+        write_through: bool = False
     ) -> Bundle | None:
         """Dereferences an RID.
         
@@ -136,7 +116,7 @@ def deref(
             rid: RID to dereference
             refresh_cache: skips cache read when `True` 
             use_network: enables fetching from other nodes when `True`
-            handle_result: handles resulting bundle with knowledge pipeline when `True`
+            handle_result: sends resulting bundle to kobj queue when `True`
         """
 
         log.debug(f"Dereferencing {rid!r}")
@@ -159,6 +139,9 @@ def deref(
                 bundle=bundle, 
                 source=source if type(source) is KoiNetNode else None
             )
+            
+            if write_through:
+                self.kobj_queue.q.join()
 
             # TODO: refactor for general solution, param to write through to cache before continuing
             # like `self.processor.kobj_queue.join()``

src/koi_net/processor/context.py

@@ -1,14 +1,17 @@
+from typing import TYPE_CHECKING
 from rid_lib.ext import Cache
 
-from koi_net.effector import Effector
-from koi_net.network.resolver import NetworkResolver
+from ..network.resolver import NetworkResolver
 from ..config.core import NodeConfig
 from ..network.graph import NetworkGraph
 from ..network.event_queue import EventQueue
 from ..network.request_handler import RequestHandler
 from ..identity import NodeIdentity
 from .kobj_queue import KobjQueue
 
+if TYPE_CHECKING:
+    from ..effector import Effector
+
 
 class HandlerContext:
     """Context object provides knowledge handlers access to other components."""
@@ -21,7 +24,7 @@ class HandlerContext:
     graph: NetworkGraph
     request_handler: RequestHandler
     resolver: NetworkResolver
-    effector: Effector
+    effector: "Effector"
 
     def __init__(
         self,
@@ -32,8 +35,7 @@ def __init__(
         kobj_queue: KobjQueue,
         graph: NetworkGraph,
         request_handler: RequestHandler,
-        resolver: NetworkResolver,
-        effector: Effector
+        resolver: NetworkResolver
     ):
         self.identity = identity
         self.config = config
@@ -43,4 +45,7 @@ def __init__(
         self.graph = graph
         self.request_handler = request_handler
         self.resolver = resolver
+    
+    def set_effector(self, effector: "Effector"):
+        """Post initialization injection of effector component."""
         self.effector = effector