Tidy endpoints, clean dispatch interface

trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py

@@ -0,0 +1,65 @@
+
+import asyncio
+from aiohttp import web
+import uuid
+import logging
+
+logger = logging.getLogger("endpoint")
+logger.setLevel(logging.INFO)
+
+class ConstantEndpoint:
+
+    def __init__(self, endpoint_path, auth, dispatcher):
+
+        self.path = endpoint_path
+
+        self.auth = auth
+        self.operation = "service"
+
+        self.dispatcher = dispatcher
+
+    async def start(self):
+        pass
+
+    def add_routes(self, app):
+
+        app.add_routes([
+            web.post(self.path, self.handle),
+        ])
+
+    async def handle(self, request):
+
+        print(request.path, "...")
+
+        try:
+            ht = request.headers["Authorization"]
+            tokens = ht.split(" ", 2)
+            if tokens[0] != "Bearer":
+                return web.HTTPUnauthorized()
+            token = tokens[1]
+        except:
+            token = ""
+
+        if not self.auth.permitted(token, self.operation):
+            return web.HTTPUnauthorized()
+
+        try:
+
+            data = await request.json()
+
+            print(data)
+
+            async def responder(x, fin):
+                print(x)
+
+            resp = await self.dispatcher.process(data, responder)
+
+            return web.json_response(resp)
+
+        except Exception as e:
+            logging.error(f"Exception: {e}")
+
+            return web.json_response(
+                { "error": str(e) }
+            )
+

trustgraph-flow/trustgraph/gateway/endpoint/flows.py

@@ -3,35 +3,24 @@ import asyncio
 
 from aiohttp import web
 
-from . endpoint import ServiceEndpoint
+#from . endpoint import ServiceEndpoint
-
-from . flow_endpoint import FlowEndpoint
-
-from .. dispatch.agent import AgentRequestor
-from .. dispatch.text_completion import TextCompletionRequestor
-from .. dispatch.prompt import PromptRequestor
-from .. dispatch.graph_rag import GraphRagRequestor
-from .. dispatch.document_rag import DocumentRagRequestor
-from .. dispatch.triples_query import TriplesQueryRequestor
-from .. dispatch.embeddings import EmbeddingsRequestor
-from .. dispatch.graph_embeddings_query import GraphEmbeddingsQueryRequestor
-from .. dispatch.prompt import PromptRequestor
-
-from .. dispatch.triples_stream import TriplesStream
 
+from . constant_endpoint import ConstantEndpoint
+from . variable_endpoint import VariableEndpoint
 from . socket import SocketEndpoint
 
-class Dispatcher:
+#from . flow_endpoint import FlowEndpoint
-    def __init__(
+
-            self, pulsar_client, timeout, queue, schema,
+# from .. dispatch.agent import AgentRequestor
-            consumer, subscriber
+# from .. dispatch.text_completion import TextCompletionRequestor
-    ):
+# from .. dispatch.prompt import PromptRequestor
-        self.pulsar_client = pulsar_client
+# from .. dispatch.graph_rag import GraphRagRequestor
-        self.timeout = timeout
+# from .. dispatch.document_rag import DocumentRagRequestor
-        self.queue = queue
+# from .. dispatch.triples_query import TriplesQueryRequestor
-        self.schema = schema
+# from .. dispatch.embeddings import EmbeddingsRequestor
-        self.consumer = consumer
+# from .. dispatch.graph_embeddings_query import GraphEmbeddingsQueryRequestor
-        self.subscriber = subscriber
+# from .. dispatch.prompt import PromptRequestor
+# from .. dispatch.triples_stream import TriplesStream
 
 class FlowEndpointManager:
 
@@ -44,22 +33,97 @@ class FlowEndpointManager:
         self.services = {
         }
 
+        class Dispatcher:
+            def __init__(self, mode, name, timeout=timeout):
+                self.mode = mode
+                self.name = name
+                self.pulsar_client = pulsar_client
+                timeout = timeout
+            async def process(self, data, responder):
+                result = { "result": "Hello world" }
+
+                if responder:
+                    await responder(result, True)
+
+                return result
+                
+        class Dispatcher2:
+            def __init__(self, mode, name, timeout=timeout):
+                self.mode = mode
+                self.name = name
+                self.pulsar_client = pulsar_client
+                timeout = timeout
+            async def process(self, data, responder, params):
+
+                thing = params['thing']
+
+                result = { "result": "Hello world", "thing": thing }
+
+                if responder:
+                    await responder(result, True)
+
+                return result
+                
+        class Dispatcher3:
+            def __init__(self, mode, name, timeout=timeout):
+                self.mode = mode
+                self.name = name
+                self.pulsar_client = pulsar_client
+                timeout = timeout
+
+            async def dispatch(self, ws, running, request):
+
+                class Runner:
+                    def __init__(self, ws, running):
+                        self.ws = ws
+                        self.running = running
+
+                    async def destroy(self):
+
+                        if self.ws:
+                            await self.ws.close()
+                            self.ws = None
+
+                        self.running.stop()
+
+                    async def run(self):
+
+                        i = 0
+
+                        while self.running.get():
+                            await self.ws.send_json({"i": i})
+                            i += 1
+                            await asyncio.sleep(1)
+
+                        await self.ws.close()
+                        self.ws = None
+
+                    async def receive(self, msg):
+                        print("Receive:", msg.data)
+
+                return Runner(ws, running)
+
         self.endpoints = [
-            FlowEndpoint(
+            ConstantEndpoint(
-                endpoint_path = "/api/v1/flow/{flow}/{kind}",
+                endpoint_path = "/api/v1/test",
                 auth = auth,
-                requestors = self.services,
+                dispatcher = Dispatcher(None, "test")
+            ),
+            VariableEndpoint(
+                endpoint_path = "/api/v1/test/{thing}",
+                auth = auth,
+                dispatcher = Dispatcher2(None, "test")
             ),
             SocketEndpoint(
-                endpoint_path = "/api/v1/flow/{flow}/stream/{kind}",
+                endpoint_path = "/api/v1/test2",
                 auth = auth,
-                dispatcher = self.create_stream_dispatch
+                dispatcher = Dispatcher3(None, "test2").dispatch,
             ),
         ]
 
         self.config_receiver.add_handler(self)
 
-    async def create_stream_dispatch(self, ws, running, request):
+    async def ASDcreate_stream_dispatch(self, ws, running, request):
 
         flow_id = request.match_info['flow']
         kind = request.match_info['kind']
@@ -86,6 +150,8 @@ class FlowEndpointManager:
 
         print("START FLOW", id)
 
+        return
+
         intf = flow["interfaces"]
 
         kinds = {
@@ -148,6 +214,8 @@ class FlowEndpointManager:
 
         print("STOP FLOW", id)
 
+        return
+
         svc_list = list(self.services.keys())
 
         for k in svc_list:

trustgraph-flow/trustgraph/gateway/endpoint/socket.py

@@ -63,17 +63,16 @@ class SocketEndpoint:
                 running = Running()
 
                 print("Create...")
-                print(self.dispatcher)
+                dispatcher = await self.dispatcher(ws, running, request)
-                dispinst = await self.dispatcher(ws, running, request)
 
                 print("Create worker...")
                 worker_task = tg.create_task(
-                    self.worker(ws, dispinst, running)
+                    self.worker(ws, dispatcher, running)
                 )
 
                 print("Create listener")
                 lsnr_task = tg.create_task(
-                    self.listener(ws, dispinst, running)
+                    self.listener(ws, dispatcher, running)
                 )
 
                 print("Created taskgroup, waiting...")
@@ -83,7 +82,7 @@ class SocketEndpoint:
             print("Task group closed")
 
             # Finally?
-            await dispinst.destroy()
+            await dispatcher.destroy()
 
         except ExceptionGroup as e:
 

trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py

@@ -0,0 +1,67 @@
+
+import asyncio
+from aiohttp import web
+import uuid
+import logging
+
+logger = logging.getLogger("endpoint")
+logger.setLevel(logging.INFO)
+
+class VariableEndpoint:
+
+    def __init__(self, endpoint_path, auth, dispatcher):
+
+        self.path = endpoint_path
+
+        self.auth = auth
+        self.operation = "service"
+
+        self.dispatcher = dispatcher
+
+    async def start(self):
+        pass
+
+    def add_routes(self, app):
+
+        app.add_routes([
+            web.post(self.path, self.handle),
+        ])
+
+    async def handle(self, request):
+
+        print(request.path, "...")
+
+        try:
+            ht = request.headers["Authorization"]
+            tokens = ht.split(" ", 2)
+            if tokens[0] != "Bearer":
+                return web.HTTPUnauthorized()
+            token = tokens[1]
+        except:
+            token = ""
+
+        if not self.auth.permitted(token, self.operation):
+            return web.HTTPUnauthorized()
+
+        try:
+
+            data = await request.json()
+
+            print(data)
+
+            async def responder(x, fin):
+                print(x)
+
+            resp = await self.dispatcher.process(
+                data, responder, request.match_info
+            )
+
+            return web.json_response(resp)
+
+        except Exception as e:
+            logging.error(f"Exception: {e}")
+
+            return web.json_response(
+                { "error": str(e) }
+            )
+