From f8b888855d94d0583dde919b8b3294cd383e2f3b Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Fri, 2 May 2025 18:40:51 +0100 Subject: [PATCH] Remove cruft, put code in right place --- .../document_embeddings_load.py | 0 .../gateway/{ => dispatch}/document_load.py | 0 .../{ => dispatch}/graph_embeddings_load.py | 0 .../gateway/{ => dispatch}/text_load.py | 0 .../gateway/{ => dispatch}/triples_load.py | 0 .../trustgraph/gateway/endpoint/endpoint.py | 65 ---------- .../gateway/endpoint/flow_endpoint.py | 74 ------------ .../trustgraph/gateway/endpoint/globals.py | 56 --------- .../trustgraph/gateway/endpoint/stream.py | 113 ------------------ 9 files changed, 308 deletions(-) rename trustgraph-flow/trustgraph/gateway/{ => dispatch}/document_embeddings_load.py (100%) rename trustgraph-flow/trustgraph/gateway/{ => dispatch}/document_load.py (100%) rename trustgraph-flow/trustgraph/gateway/{ => dispatch}/graph_embeddings_load.py (100%) rename trustgraph-flow/trustgraph/gateway/{ => dispatch}/text_load.py (100%) rename trustgraph-flow/trustgraph/gateway/{ => dispatch}/triples_load.py (100%) delete mode 100644 trustgraph-flow/trustgraph/gateway/endpoint/endpoint.py delete mode 100644 trustgraph-flow/trustgraph/gateway/endpoint/flow_endpoint.py delete mode 100644 trustgraph-flow/trustgraph/gateway/endpoint/globals.py delete mode 100644 trustgraph-flow/trustgraph/gateway/endpoint/stream.py diff --git a/trustgraph-flow/trustgraph/gateway/document_embeddings_load.py b/trustgraph-flow/trustgraph/gateway/dispatch/document_embeddings_load.py similarity index 100% rename from trustgraph-flow/trustgraph/gateway/document_embeddings_load.py rename to trustgraph-flow/trustgraph/gateway/dispatch/document_embeddings_load.py diff --git a/trustgraph-flow/trustgraph/gateway/document_load.py b/trustgraph-flow/trustgraph/gateway/dispatch/document_load.py similarity index 100% rename from trustgraph-flow/trustgraph/gateway/document_load.py rename to trustgraph-flow/trustgraph/gateway/dispatch/document_load.py diff --git a/trustgraph-flow/trustgraph/gateway/graph_embeddings_load.py b/trustgraph-flow/trustgraph/gateway/dispatch/graph_embeddings_load.py similarity index 100% rename from trustgraph-flow/trustgraph/gateway/graph_embeddings_load.py rename to trustgraph-flow/trustgraph/gateway/dispatch/graph_embeddings_load.py diff --git a/trustgraph-flow/trustgraph/gateway/text_load.py b/trustgraph-flow/trustgraph/gateway/dispatch/text_load.py similarity index 100% rename from trustgraph-flow/trustgraph/gateway/text_load.py rename to trustgraph-flow/trustgraph/gateway/dispatch/text_load.py diff --git a/trustgraph-flow/trustgraph/gateway/triples_load.py b/trustgraph-flow/trustgraph/gateway/dispatch/triples_load.py similarity index 100% rename from trustgraph-flow/trustgraph/gateway/triples_load.py rename to trustgraph-flow/trustgraph/gateway/dispatch/triples_load.py diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/endpoint.py deleted file mode 100644 index 94980e8b..00000000 --- a/trustgraph-flow/trustgraph/gateway/endpoint/endpoint.py +++ /dev/null @@ -1,65 +0,0 @@ - -import asyncio -from aiohttp import web -import uuid -import logging - -logger = logging.getLogger("endpoint") -logger.setLevel(logging.INFO) - -class ServiceEndpoint: - - def __init__(self, endpoint_path, auth, requestor): - - self.path = endpoint_path - - self.auth = auth - self.operation = "service" - - self.requestor = requestor - - async def start(self): - await self.requestor.start() - - 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.requestor.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/flow_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/flow_endpoint.py deleted file mode 100644 index e061ab2a..00000000 --- a/trustgraph-flow/trustgraph/gateway/endpoint/flow_endpoint.py +++ /dev/null @@ -1,74 +0,0 @@ - -import asyncio -from aiohttp import web -import uuid -import logging - -logger = logging.getLogger("flow-endpoint") -logger.setLevel(logging.INFO) - -class FlowEndpoint: - - def __init__(self, endpoint_path, auth, requestors): - - self.path = endpoint_path - - self.auth = auth - self.operation = "service" - - self.requestors = requestors - - 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, "...") - - flow_id = request.match_info['flow'] - kind = request.match_info['kind'] - k = (flow_id, kind) - - if k not in self.requestors: - raise web.HTTPBadRequest() - - requestor = self.requestors[k] - - 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 requestor.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/globals.py b/trustgraph-flow/trustgraph/gateway/endpoint/globals.py deleted file mode 100644 index 70943d2c..00000000 --- a/trustgraph-flow/trustgraph/gateway/endpoint/globals.py +++ /dev/null @@ -1,56 +0,0 @@ - -from . endpoint import ServiceEndpoint - -from .. dispatch.librarian import LibrarianRequestor -from .. dispatch.config import ConfigRequestor -from .. dispatch.flow import FlowRequestor -from . metrics import MetricsEndpoint - -class GlobalEndpointManager: - - def __init__(self, pulsar_client, auth, prometheus_url, timeout=600): - - self.pulsar_client = pulsar_client - - self.services = { - "librarian": LibrarianRequestor( - pulsar_client=self.pulsar_client, timeout=timeout, - ), - "config": ConfigRequestor( - pulsar_client=self.pulsar_client, timeout=timeout, - ), - "flow": FlowRequestor( - pulsar_client=self.pulsar_client, timeout=timeout, - ) - } - - self.endpoints = [ - ServiceEndpoint( - endpoint_path = "/api/v1/librarian", auth = auth, - requestor = self.services["librarian"], - ), - ServiceEndpoint( - endpoint_path = "/api/v1/config", auth = auth, - requestor = self.services["config"], - ), - ServiceEndpoint( - endpoint_path = "/api/v1/flow", auth = auth, - requestor = self.services["flow"], - ), - MetricsEndpoint( - endpoint_path = "/api/v1/metrics", - prometheus_url = prometheus_url, - auth = auth, - ), - ] - - def add_routes(self, app): - - for ep in self.endpoints: - ep.add_routes(app) - - async def start(self): - - for ep in self.endpoints: - await ep.start() - diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/stream.py b/trustgraph-flow/trustgraph/gateway/endpoint/stream.py deleted file mode 100644 index e3d1a97d..00000000 --- a/trustgraph-flow/trustgraph/gateway/endpoint/stream.py +++ /dev/null @@ -1,113 +0,0 @@ - -import asyncio -from aiohttp import web, WSMsgType -import logging - -from .. running import Running - -logger = logging.getLogger("socket") -logger.setLevel(logging.INFO) - -class StreamEndpoint: - - def __init__( - self, endpoint_path, auth, dispatcher, - ): - - self.path = endpoint_path - self.auth = auth - self.operation = "socket" - - self.dispatcher = dispatcher - - async def worker(self, ws, dispatcher, running): - - await dispatcher.run() - - async def listener(self, ws, dispatcher, running): - - async for msg in ws: - - # On error, finish - if msg.type == WSMsgType.TEXT: - await dispatcher.receive(msg) - continue - elif msg.type == WSMsgType.BINARY: - await dispatcher.receive(msg) - continue - else: - break - - running.stop() - await ws.close() - - async def handle(self, request): - - try: - token = request.query['token'] - except: - token = "" - - if not self.auth.permitted(token, self.operation): - return web.HTTPUnauthorized() - - # 50MB max message size - ws = web.WebSocketResponse(max_msg_size=52428800) - - await ws.prepare(request) - - try: - - async with asyncio.TaskGroup() as tg: - - running = Running() - - print("Create...") - print(self.dispatcher) - dispinst = await self.dispatcher.create(ws, running, request) - - print("Create worker...") - worker_task = tg.create_task( - self.worker(ws, dispinst, running) - ) - - print("Create listener") - lsnr_task = tg.create_task( - self.listener(ws, dispinst, running) - ) - - print("Created taskgroup, waiting...") - - # Wait for threads to complete - - print("Task group closed") - - # Finally? - await dispinst.destroy() - - except ExceptionGroup as e: - - print("Exception group:", flush=True) - - for se in e.exceptions: - print(" Type:", type(se), flush=True) - print(f" Exception: {se}", flush=True) - except Exception as e: - print("Socket exception:", e, flush=True) - - await ws.close() - - return ws - - async def start(self): - pass - - async def stop(self): - self.running.stop() - - def add_routes(self, app): - - app.add_routes([ - web.get(self.path, self.handle), - ]) -