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

Fixing collection management translation

Browse source
This commit is contained in:
Cyber MacGeddon 2025-09-19 12:17:09 +01:00
parent 6ba56a2b42
commit 78f84bce47
1 changed files with 37 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

74
trustgraph-base/trustgraph/messaging/translators/collection.py
View file

@ -8,43 +8,43 @@ class CollectionManagementRequestTranslator(MessageTranslator):
def to_pulsar(self, data: Dict[str, Any]) -> CollectionManagementRequest: def to_pulsar(self, data: Dict[str, Any]) -> CollectionManagementRequest:
return CollectionManagementRequest( return CollectionManagementRequest(
operation=data.get("operation", ""), operation=data.get("operation"),
user=data.get("user", ""), user=data.get("user"),
collection=data.get("collection", ""), collection=data.get("collection"),
timestamp=data.get("timestamp", ""), timestamp=data.get("timestamp"),
name=data.get("name", ""), name=data.get("name"),
description=data.get("description", ""), description=data.get("description"),
tags=data.get("tags", []), tags=data.get("tags"),
created_at=data.get("created_at", ""), created_at=data.get("created_at"),
updated_at=data.get("updated_at", ""), updated_at=data.get("updated_at"),
tag_filter=data.get("tag_filter", []), tag_filter=data.get("tag_filter"),
limit=data.get("limit", 50) limit=data.get("limit")
) )
def from_pulsar(self, obj: CollectionManagementRequest) -> Dict[str, Any]: def from_pulsar(self, obj: CollectionManagementRequest) -> Dict[str, Any]:
result = {} result = {}
if obj.operation: if obj.operation is not None:
result["operation"] = obj.operation result["operation"] = obj.operation
if obj.user: if obj.user is not None:
result["user"] = obj.user result["user"] = obj.user
if obj.collection: if obj.collection is not None:
result["collection"] = obj.collection result["collection"] = obj.collection
if obj.timestamp: if obj.timestamp is not None:
result["timestamp"] = obj.timestamp result["timestamp"] = obj.timestamp
if obj.name: if obj.name is not None:
result["name"] = obj.name result["name"] = obj.name
if obj.description: if obj.description is not None:
result["description"] = obj.description result["description"] = obj.description
if obj.tags: if obj.tags is not None:
result["tags"] = list(obj.tags) result["tags"] = list(obj.tags)
if obj.created_at: if obj.created_at is not None:
result["created_at"] = obj.created_at result["created_at"] = obj.created_at
if obj.updated_at: if obj.updated_at is not None:
result["updated_at"] = obj.updated_at result["updated_at"] = obj.updated_at
if obj.tag_filter: if obj.tag_filter is not None:
result["tag_filter"] = list(obj.tag_filter) result["tag_filter"] = list(obj.tag_filter)
if obj.limit: if obj.limit is not None:
result["limit"] = obj.limit result["limit"] = obj.limit
return result return result
@ -59,8 +59,8 @@ class CollectionManagementResponseTranslator(MessageTranslator):
if "error" in data and data["error"]: if "error" in data and data["error"]:
error_data = data["error"] error_data = data["error"]
error = Error( error = Error(
type=error_data.get("type", ""), type=error_data.get("type"),
message=error_data.get("message", "") message=error_data.get("message")
) )
# Handle collections array # Handle collections array
@ -68,35 +68,35 @@ class CollectionManagementResponseTranslator(MessageTranslator):
if "collections" in data: if "collections" in data:
for coll_data in data["collections"]: for coll_data in data["collections"]:
collections.append(CollectionMetadata( collections.append(CollectionMetadata(
user=coll_data.get("user", ""), user=coll_data.get("user"),
collection=coll_data.get("collection", ""), collection=coll_data.get("collection"),
name=coll_data.get("name", ""), name=coll_data.get("name"),
description=coll_data.get("description", ""), description=coll_data.get("description"),
tags=coll_data.get("tags", []), tags=coll_data.get("tags"),
created_at=coll_data.get("created_at", ""), created_at=coll_data.get("created_at"),
updated_at=coll_data.get("updated_at", "") updated_at=coll_data.get("updated_at")
)) ))
return CollectionManagementResponse( return CollectionManagementResponse(
success=data.get("success", ""), success=data.get("success"),
error=error, error=error,
timestamp=data.get("timestamp", ""), timestamp=data.get("timestamp"),
collections=collections collections=collections
) )
def from_pulsar(self, obj: CollectionManagementResponse) -> Dict[str, Any]: def from_pulsar(self, obj: CollectionManagementResponse) -> Dict[str, Any]:
result = {} result = {}
if obj.success: if obj.success is not None:
result["success"] = obj.success result["success"] = obj.success
if obj.error: if obj.error is not None:
result["error"] = { result["error"] = {
"type": obj.error.type, "type": obj.error.type,
"message": obj.error.message "message": obj.error.message
} }
if obj.timestamp: if obj.timestamp is not None:
result["timestamp"] = obj.timestamp result["timestamp"] = obj.timestamp
if obj.collections: if obj.collections is not None:
result["collections"] = [] result["collections"] = []
for coll in obj.collections: for coll in obj.collections:
result["collections"].append({ result["collections"].append({