From cfa4cfa26ba42b8c1bf0a0597ef05c3aa58bf43e Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Fri, 2 May 2025 14:20:07 +0100 Subject: [PATCH] Tidy endpoints, clean dispatch interface --- .../gateway/endpoint/constant_endpoint.py | 65 +++++++++ .../trustgraph/gateway/endpoint/flows.py | 134 +++++++++++++----- .../trustgraph/gateway/endpoint/socket.py | 9 +- .../gateway/endpoint/variable_endpoint.py | 67 +++++++++ 4 files changed, 237 insertions(+), 38 deletions(-) create mode 100644 trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py create mode 100644 trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/constant_endpoint.py new file mode 100644 index 00000000..eda9da4d --- /dev/null +++ b/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) } + ) + diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/flows.py b/trustgraph-flow/trustgraph/gateway/endpoint/flows.py index a7b90765..8deef287 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/flows.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/flows.py @@ -3,35 +3,24 @@ import asyncio from aiohttp import web -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 . endpoint import ServiceEndpoint +from . constant_endpoint import ConstantEndpoint +from . variable_endpoint import VariableEndpoint from . socket import SocketEndpoint -class Dispatcher: - def __init__( - self, pulsar_client, timeout, queue, schema, - consumer, subscriber - ): - self.pulsar_client = pulsar_client - self.timeout = timeout - self.queue = queue - self.schema = schema - self.consumer = consumer - self.subscriber = subscriber +#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 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( - endpoint_path = "/api/v1/flow/{flow}/{kind}", + ConstantEndpoint( + 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 = { @@ -146,7 +212,9 @@ class FlowEndpointManager: async def stop_flow(self, id, flow): - print("STOP FLOW", id) + print("STOP FLOW", id) + + return svc_list = list(self.services.keys()) diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/socket.py b/trustgraph-flow/trustgraph/gateway/endpoint/socket.py index 543d54c2..efe55caa 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/socket.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/socket.py @@ -63,17 +63,16 @@ class SocketEndpoint: running = Running() print("Create...") - print(self.dispatcher) - dispinst = await self.dispatcher(ws, running, request) + dispatcher = 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: diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/variable_endpoint.py new file mode 100644 index 00000000..8813165c --- /dev/null +++ b/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) } + ) +