diff --git a/test-api/test-library-list-documents b/test-api/test-library-list-documents index 4166856d..9677ce4d 100755 --- a/test-api/test-library-list-documents +++ b/test-api/test-library-list-documents @@ -10,8 +10,6 @@ url = "http://localhost:8088/api/v1/" ############################################################################ -id = "http://trustgraph.ai/doc/9fdee98b-b259-40ac-bcb9-8e82ccedeb04" - input = { "operation": "list-documents", "user": "trustgraph", diff --git a/test-api/test-library-remove-document b/test-api/test-library-remove-document new file mode 100755 index 00000000..6354c292 --- /dev/null +++ b/test-api/test-library-remove-document @@ -0,0 +1,41 @@ +#!/usr/bin/env python3 + +import requests +import json +import sys +import base64 +import time + +url = "http://localhost:8088/api/v1/" + +############################################################################ + +id = "http://trustgraph.ai/doc/9fdee98b-b259-40ac-bcb9-8e82ccedeb04" + +input = { + "operation": "remove-document", + "user": "trustgraph", + "document-id": id +} + +resp = requests.post( + f"{url}librarian", + json=input, +) + +print(resp.text) +resp = resp.json() + +print(resp) + +if "error" in resp: + print(f"Error: {resp['error']}") + sys.exit(1) + +# print(resp["response"]) +print(resp) + +sys.exit(0) + +############################################################################ + diff --git a/test-api/test-library-remove-document2 b/test-api/test-library-remove-document2 new file mode 100755 index 00000000..fd57d025 --- /dev/null +++ b/test-api/test-library-remove-document2 @@ -0,0 +1,41 @@ +#!/usr/bin/env python3 + +import requests +import json +import sys +import base64 +import time + +url = "http://localhost:8088/api/v1/" + +############################################################################ + +id = "http://trustgraph.ai/doc/6d034da9-2759-45c2-af24-14db7f4c44c2" + +input = { + "operation": "remove-document", + "user": "trustgraph", + "document-id": id +} + +resp = requests.post( + f"{url}librarian", + json=input, +) + +print(resp.text) +resp = resp.json() + +print(resp) + +if "error" in resp: + print(f"Error: {resp['error']}") + sys.exit(1) + +# print(resp["response"]) +print(resp) + +sys.exit(0) + +############################################################################ + diff --git a/trustgraph-flow/trustgraph/librarian/blob_store.py b/trustgraph-flow/trustgraph/librarian/blob_store.py index 5cffef18..dcc8b112 100644 --- a/trustgraph-flow/trustgraph/librarian/blob_store.py +++ b/trustgraph-flow/trustgraph/librarian/blob_store.py @@ -37,7 +37,7 @@ class BlobStore: else: print("Bucket", self.bucket_name, "already exists", flush=True) - def add(self, object_id, blob, kind): + async def add(self, object_id, blob, kind): # FIXME: Loop retry self.minio.put_object( @@ -49,3 +49,14 @@ class BlobStore: ) print("Add blob complete", flush=True) + + async def remove(self, object_id): + + # FIXME: Loop retry + self.minio.remove_object( + bucket_name = self.bucket_name, + object_name = "doc/" + str(object_id), + ) + + print("Remove blob complete", flush=True) + diff --git a/trustgraph-flow/trustgraph/librarian/librarian.py b/trustgraph-flow/trustgraph/librarian/librarian.py index 88981f89..414e87be 100644 --- a/trustgraph-flow/trustgraph/librarian/librarian.py +++ b/trustgraph-flow/trustgraph/librarian/librarian.py @@ -53,10 +53,11 @@ class Librarian: print("Add blob...") - - self.blob_store.add(object_id, base64.b64decode(request.content), - request.document_metadata.kind) + await self.blob_store.add( + object_id, base64.b64decode(request.content), + request.document_metadata.kind + ) print("Add table...") @@ -75,7 +76,38 @@ class Librarian: ) async def remove_document(self, request): - raise RuntimeError("Not implemented") + + print("REMOVING...") + + if not await self.table_store.document_exists( + request.user, + request.document_id, + ): + raise RuntimeError("Document does not exist") + + object_id = await self.table_store.get_document_object_id( + request.user, + request.document_id + ) + + # Remove blob... + await self.blob_store.remove(object_id) + + # Remove doc table row + await self.table_store.remove_document( + request.user, + request.document_id + ) + + print("Remove complete", flush=True) + + return LibrarianResponse( + error = None, + document_metadata = None, + content = None, + document_metadatas = None, + processing_metadatas = None, + ) async def update_document(self, request): diff --git a/trustgraph-flow/trustgraph/librarian/table_store.py b/trustgraph-flow/trustgraph/librarian/table_store.py index 73cb3afe..b80d62b8 100644 --- a/trustgraph-flow/trustgraph/librarian/table_store.py +++ b/trustgraph-flow/trustgraph/librarian/table_store.py @@ -187,12 +187,16 @@ class TableStore: """) self.get_document_stmt = self.cassandra.prepare(""" - SELECT user, time, kind, title, comments, - metadata, tags, object_id + SELECT time, kind, title, comments, metadata, tags, object_id FROM document WHERE user = ? AND id = ? """) + self.delete_document_stmt = self.cassandra.prepare(""" + DELETE FROM document + WHERE user = ? AND id = ? + """) + self.test_document_exists_stmt = self.cassandra.prepare(""" SELECT id FROM document @@ -329,6 +333,31 @@ class TableStore: print("Update complete", flush=True) + async def remove_document(self, user, document_id): + + print("Removing document", document_id) + + while True: + + try: + + resp = self.cassandra.execute( + self.delete_document_stmt, + ( + user, document_id + ) + ) + + break + + except Exception as e: + + print("Exception:", type(e)) + print(f"{e}, retry...", flush=True) + await asyncio.sleep(1) + + print("Delete complete", flush=True) + async def add_triples(self, m): when = int(time.time() * 1000) @@ -423,6 +452,78 @@ class TableStore: return lst + async def get_document(self, user, id): + + print("GET") + + while True: + + try: + + resp = self.cassandra.execute( + self.get_document_stmt, + (user, id) + ) + + print("OK") + break + + except Exception as e: + print("Exception:", type(e)) + print(f"{e}, retry...", flush=True) + await asyncio.sleep(1) + + + for row in resp: + doc = DocumentMetadata( + id = id, + user = user, + time = int(time.mktime(row[0].timetuple())), + kind = row[1], + title = row[2], + comments = row[3], + 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[4] + ], + tags = row[5], + object_id = row[6], + ) + return doc + + raise RuntimeError("No such document row?") + + async def get_document_object_id(self, user, id): + + print("GET") + + while True: + + try: + + resp = self.cassandra.execute( + self.get_document_stmt, + (user, id) + ) + + print("OK") + break + + except Exception as e: + print("Exception:", type(e)) + print(f"{e}, retry...", flush=True) + await asyncio.sleep(1) + + + for row in resp: + return row[6] + + raise RuntimeError("No such document row?") + async def add_graph_embeddings(self, m): when = int(time.time() * 1000)