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

Fix object writer management

Browse source
This commit is contained in:
Cyber MacGeddon 2025-09-30 22:04:57 +01:00
parent 292d5cb2a1
commit 1d634df25a
1 changed files with 10 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

19
trustgraph-flow/trustgraph/storage/objects/cassandra/write.py
View file

@ -456,35 +456,36 @@ class Processor(FlowProcessor):
async def on_storage_management(self, msg, consumer, flow): async def on_storage_management(self, msg, consumer, flow):
"""Handle storage management requests for collection operations""" """Handle storage management requests for collection operations"""
logger.info(f"Received storage management request: {msg.operation} for {msg.user}/{msg.collection}") request = msg.value()
logger.info(f"Received storage management request: {request.operation} for {request.user}/{request.collection}")
try: try:
if msg.operation == "create-collection": if request.operation == "create-collection":
await self.create_collection(msg.user, msg.collection) await self.create_collection(request.user, request.collection)
# Send success response # Send success response
response = StorageManagementResponse( response = StorageManagementResponse(
error=None # No error means success error=None # No error means success
) )
await self.storage_response_producer.send(response) await self.storage_response_producer.send(response)
logger.info(f"Successfully created collection {msg.user}/{msg.collection}") logger.info(f"Successfully created collection {request.user}/{request.collection}")
elif msg.operation == "delete-collection": elif request.operation == "delete-collection":
await self.delete_collection(msg.user, msg.collection) await self.delete_collection(request.user, request.collection)
# Send success response # Send success response
response = StorageManagementResponse( response = StorageManagementResponse(
error=None # No error means success error=None # No error means success
) )
await self.storage_response_producer.send(response) await self.storage_response_producer.send(response)
logger.info(f"Successfully deleted collection {msg.user}/{msg.collection}") logger.info(f"Successfully deleted collection {request.user}/{request.collection}")
else: else:
logger.warning(f"Unknown storage management operation: {msg.operation}") logger.warning(f"Unknown storage management operation: {request.operation}")
# Send error response # Send error response
from .... schema import Error from .... schema import Error
response = StorageManagementResponse( response = StorageManagementResponse(
error=Error( error=Error(
type="unknown_operation", type="unknown_operation",
message=f"Unknown operation: {msg.operation}" message=f"Unknown operation: {request.operation}"
) )
) )
await self.storage_response_producer.send(response) await self.storage_response_producer.send(response)