apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-04-25 00:16:23 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Fix async/sync load issues with knowledge streaming APIs (#315)

Browse source
This commit is contained in:
cybermaggedon 2025-03-15 12:38:10 +00:00 • committed by GitHub
parent c08779ff3c
commit a922852017
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 6 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

2
trustgraph-flow/trustgraph/gateway/document_embeddings_load.py
View file

@ -59,6 +59,6 @@ class DocumentEmbeddingsLoadEndpoint(SocketEndpoint):
],
)
await self.publisher.send(None, elt)
self.publisher.send(None, elt)
running.stop()

4
trustgraph-flow/trustgraph/gateway/graph_embeddings_load.py
View file

@ -36,6 +36,7 @@ class GraphEmbeddingsLoadEndpoint(SocketEndpoint):
async def listener(self, ws, running):
async for msg in ws:
# On error, finish
if msg.type == WSMsgType.ERROR:
break
@ -59,7 +60,6 @@ class GraphEmbeddingsLoadEndpoint(SocketEndpoint):
]
)
await self.publisher.send(None, elt)
self.publisher.send(None, elt)
running.stop()

2
trustgraph-flow/trustgraph/gateway/requestor.py
View file

@ -53,11 +53,9 @@ class ServiceRequestor:
q = self.sub.subscribe(id)
print("BOUT TO SEDN")
await asyncio.to_thread(
self.pub.send, id, self.to_request(request)
)
print("SENT")
while True:

4
trustgraph-flow/trustgraph/gateway/socket.py
View file

@ -19,7 +19,7 @@ class SocketEndpoint:
self.operation = "socket"
async def listener(self, ws, running):
async for msg in ws:
# On error, finish
if msg.type == WSMsgType.TEXT:
@ -53,7 +53,7 @@ class SocketEndpoint:
try:
await self.listener(ws, running)
except Exception as e:
print(e, flush=True)
print("Socket exception:", e, flush=True)
running.stop()

2
trustgraph-flow/trustgraph/gateway/triples_load.py
View file

@ -51,7 +51,7 @@ class TriplesLoadEndpoint(SocketEndpoint):
triples=to_subgraph(data["triples"]),
)
await self.publisher.send(None, elt)
self.publisher.send(None, elt)
running.stop()