apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-20 18:51:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Added list function, incomplete

Browse source
This commit is contained in:
Cyber MacGeddon 2025-02-15 19:27:17 +00:00
parent a74119c93e
commit 8d4b67710c
6 changed files with 144 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

21
trustgraph-base/trustgraph/schema/library.py
View file

@ -1,11 +1,21 @@
from pulsar.schema import Record, Bytes, String, Array, Timestamp
from pulsar.schema import Record, Bytes, String, Array, Long
from . types import Triple
from . topic import topic
from . types import Error
from . metadata import Metadata
from . documents import Document, TextDocument
# add
# -> (id, document)
# <- ()
# <- (error)
# list
# -> (user, collection?)
# <- (info)
# <- (error)
# add(Metadata, Bytes) : error?
# copy(id, user, collection)
# move(id, user, collection)
@ -17,23 +27,24 @@ from . documents import Document, TextDocument
# search(<key,op,value>[]) : id[]
class DocumentPackage(Record):
metadata = Array(Triple())
document = Bytes()
kind = String()
user = String()
collection = String()
title = String()
comments = String()
time = Timestamp()
time = Long()
metadata = Array(Triple())
class DocumentInfo(Record):
metadata = Array(Triple())
id = String()
kind = String()
user = String()
collection = String()
title = String()
comments = String()
time = Timestamp()
time = Long()
metadata = Array(Triple())
class Criteria(Record):
key = String()

5
trustgraph-flow/trustgraph/gateway/librarian.py
View file

@ -53,7 +53,10 @@ class LibrarianRequestor(ServiceRequestor):
response["document"] = serialize_document_package(message.document)
if message.info:
response["info"] = serialize_document_info(message.info)
response["info"] = [
serialize_document_info(v)
for v in message.info
]
return response, True

16
trustgraph-flow/trustgraph/gateway/serialize.py
View file

@ -108,8 +108,8 @@ def serialize_document_info(message):
ret = {}
if message.metadata:
ret["metadata"] = serialize_subgraph(message.metdata)
if message.id:
ret["id"] = message.id
if message.kind:
ret["kind"] = message.kind
@ -120,6 +120,18 @@ def serialize_document_info(message):
if message.collection:
ret["collection"] = message.collection
if message.title:
ret["title"] = message.title
if message.comments:
ret["comments"] = message.comments
if message.time:
ret["time"] = message.time
if message.metadata:
ret["metadata"] = serialize_subgraph(message.metadata)
return ret
def to_document_package(x):

16
trustgraph-flow/trustgraph/librarian/librarian.py
View file

@ -1,4 +1,4 @@
from .. schema import LibrarianRequest, LibrarianResponse, Error
from .. schema import LibrarianRequest, LibrarianResponse, Error, Triple
from .. knowledge import hash
from .. exceptions import RequestError
from . table_store import TableStore
@ -53,6 +53,20 @@ class Librarian:
info = None,
)
async def list(self, user, collection):
print("list")
info = self.table_store.list(user, collection)
print(">>", info)
return LibrarianResponse(
error = None,
document = None,
info = info,
)
def handle_triples(self, m):
self.table_store.add_triples(m)

13
trustgraph-flow/trustgraph/librarian/service.py
View file

@ -272,6 +272,16 @@ class Processor(ConsumerProducer):
else:
raise RequestError("Invalid call")
if v.operation == "list":
if v.user:
return partial(
self.librarian.list,
user = v.user,
collection = v.collection,
)
else:
raise RequestError("Invalid call")
raise RequestError("Invalid operation: " + v.operation)
async def handle(self, msg):
@ -298,6 +308,7 @@ class Processor(ConsumerProducer):
try:
resp = await func()
print("->", resp)
except RequestError as e:
resp = LibrarianResponse(
error = Error(
@ -318,7 +329,7 @@ class Processor(ConsumerProducer):
await self.send(resp, properties={"id": id})
return
print("Send response...", flush=True)
print("Send response..!.", flush=True)
await self.send(resp, properties={"id": id})

84
trustgraph-flow/trustgraph/librarian/table_store.py
View file

@ -1,4 +1,5 @@
from .. schema import LibrarianRequest, LibrarianResponse, Error
from .. schema import LibrarianRequest, LibrarianResponse
from .. schema import DocumentInfo, Error, Triple, Value
from .. knowledge import hash
from .. exceptions import RequestError
@ -157,6 +158,20 @@ class TableStore:
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""")
self.list_document_stmt = self.cassandra.prepare("""
SELECT
id, kind, user, collection, title, comments, time, metadata
FROM document
WHERE user = ?
""")
self.list_document_by_collection_stmt = self.cassandra.prepare("""
SELECT
id, kind, user, collection, title, comments, time, metadata
FROM document
WHERE user = ? AND collection = ?
""")
self.insert_triples_stmt = self.cassandra.prepare("""
INSERT INTO triples
(
@ -275,6 +290,73 @@ class TableStore:
print(f"{e}, retry...", flush=True)
time.sleep(1)
def list(self, user, collection=None):
print("LIST")
while True:
print("TRY")
print(self.list_document_stmt)
try:
if collection:
resp = self.cassandra.execute(
self.list_document_by_collection_stmt,
(user, collection)
)
else:
resp = self.cassandra.execute(
self.list_document_stmt,
(user,)
)
break
print("OK")
except Exception as e:
print("Exception:", type(e))
print(f"{e}, retry...", flush=True)
time.sleep(1)
print("OK2")
info = [
DocumentInfo(
id = row[0],
kind = row[1],
user = row[2],
collection = row[3],
title = row[4],
comments = row[5],
time = int(1000 * row[6].timestamp()),
metadata = [
Triple(
s=Value(value=m[0], is_uri=m[1]),
p=Value(value=m[2], is_uri=m[3]),
o=Value(value=m[4], is_uri=m[5])
)
for m in row[7]
],
)
for row in resp
]
print("OK3")
print(info[0])
print(info[0].user)
print(info[0].time)
print(info[0].kind)
print(info[0].collection)
print(info[0].title)
print(info[0].comments)
print(info[0].metadata)
print(info[0].metadata)
return info
def add_graph_embeddings(self, m):
when = int(time.time() * 1000)