mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-20 18:51:03 +02:00
Remove cruft, put code in right place
This commit is contained in:
Cyber MacGeddon
parent
67e5a968e6
commit
f8b888855d
9 changed files with 0 additions and 308 deletions
|
|
@ -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) }
|
||||
)
|
||||
|
||||
|
|
@ -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) }
|
||||
)
|
||||
|
||||
|
|
@ -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()
|
||||
|
||||
|
|
@ -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),
|
||||
])
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue