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

Fixing more Cassandra work

Browse source
This commit is contained in:
Cyber MacGeddon 2025-09-04 00:13:50 +01:00
parent ccaec88a72
commit 9bd8ee733d
6 changed files with 65 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

12
trustgraph-base/trustgraph/base/cassandra_config.py
View file

@ -119,26 +119,16 @@ def get_cassandra_config_from_params(params: dict) -> Tuple[List[str], Optional[
"""
Extract and resolve Cassandra configuration from a parameters dictionary.
Handles both old graph_* and new cassandra_* parameter names for backward compatibility.
Args:
params: Dictionary of parameters that may contain Cassandra configuration
Returns:
tuple: (hosts_list, username, password)
"""
# Check for new parameter names first
# Get Cassandra parameters
host = params.get('cassandra_host')
username = params.get('cassandra_username')
password = params.get('cassandra_password')
# Fall back to old graph_* names for backward compatibility
if not host:
host = params.get('graph_host')
if not username:
username = params.get('graph_username', params.get('cassandra_user'))
if not password:
password = params.get('graph_password')
# Use resolve function to handle defaults and list conversion
return resolve_cassandra_config(host=host, username=username, password=password)

53
trustgraph-flow/trustgraph/query/objects/cassandra/service.py
View file

@ -21,12 +21,12 @@ from strawberry.tools import create_type
from .... schema import ObjectsQueryRequest, ObjectsQueryResponse, GraphQLError
from .... schema import Error, RowSchema, Field as SchemaField
from .... base import FlowProcessor, ConsumerSpec, ProducerSpec
from .... base.cassandra_config import add_cassandra_args, resolve_cassandra_config
# Module logger
logger = logging.getLogger(__name__)
default_ident = "objects-query"
default_graph_host = 'localhost'
# GraphQL filter input types
@strawberry.input
@ -68,10 +68,22 @@ class Processor(FlowProcessor):
id = params.get("id", default_ident)
# Cassandra connection parameters
self.graph_host = params.get("graph_host", default_graph_host)
self.graph_username = params.get("graph_username", None)
self.graph_password = params.get("graph_password", None)
# Get Cassandra parameters
cassandra_host = params.get("cassandra_host")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
# Resolve configuration with environment variable fallback
hosts, username, password = resolve_cassandra_config(
host=cassandra_host,
username=cassandra_username,
password=cassandra_password
)
# Store resolved configuration with proper names
self.cassandra_host = hosts # Store as list
self.cassandra_username = username
self.cassandra_password = password
# Config key for schemas
self.config_key = params.get("config_type", "schema")
@ -124,20 +136,20 @@ class Processor(FlowProcessor):
return
try:
if self.graph_username and self.graph_password:
if self.cassandra_username and self.cassandra_password:
auth_provider = PlainTextAuthProvider(
username=self.graph_username,
password=self.graph_password
username=self.cassandra_username,
password=self.cassandra_password
)
self.cluster = Cluster(
contact_points=[self.graph_host],
contact_points=self.cassandra_host,
auth_provider=auth_provider
)
else:
self.cluster = Cluster(contact_points=[self.graph_host])
self.cluster = Cluster(contact_points=self.cassandra_host)
self.session = self.cluster.connect()
logger.info(f"Connected to Cassandra cluster at {self.graph_host}")
logger.info(f"Connected to Cassandra cluster at {self.cassandra_host}")
except Exception as e:
logger.error(f"Failed to connect to Cassandra: {e}", exc_info=True)
@ -712,24 +724,7 @@ class Processor(FlowProcessor):
"""Add command-line arguments"""
FlowProcessor.add_args(parser)
parser.add_argument(
'-g', '--graph-host',
default=default_graph_host,
help=f'Cassandra host (default: {default_graph_host})'
)
parser.add_argument(
'--graph-username',
default=None,
help='Cassandra username'
)
parser.add_argument(
'--graph-password',
default=None,
help='Cassandra password'
)
add_cassandra_args(parser)
parser.add_argument(
'--config-type',

22
trustgraph-flow/trustgraph/query/triples/cassandra/service.py
View file

@ -22,10 +22,10 @@ class Processor(TriplesQueryService):
def __init__(self, **params):
# Use new parameter names, fall back to old for compatibility
cassandra_host = params.get("cassandra_host", params.get("graph_host"))
cassandra_username = params.get("cassandra_username", params.get("graph_username"))
cassandra_password = params.get("cassandra_password", params.get("graph_password"))
# Get Cassandra parameters
cassandra_host = params.get("cassandra_host")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
# Resolve configuration with environment variable fallback
hosts, username, password = resolve_cassandra_config(
@ -41,9 +41,9 @@ class Processor(TriplesQueryService):
}
)
self.graph_host = hosts
self.username = username
self.password = password
self.cassandra_host = hosts
self.cassandra_username = username
self.cassandra_password = password
self.table = None
def create_value(self, ent):
@ -59,15 +59,15 @@ class Processor(TriplesQueryService):
table = (query.user, query.collection)
if table != self.table:
if self.username and self.password:
if self.cassandra_username and self.cassandra_password:
self.tg = TrustGraph(
hosts=self.graph_host,
hosts=self.cassandra_host,
keyspace=query.user, table=query.collection,
username=self.username, password=self.password
username=self.cassandra_username, password=self.cassandra_password
)
else:
self.tg = TrustGraph(
hosts=self.graph_host,
hosts=self.cassandra_host,
keyspace=query.user, table=query.collection,
)
self.table = table

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

@ -27,10 +27,10 @@ class Processor(FlowProcessor):
id = params.get("id", default_ident)
# Use new parameter names, fall back to old for compatibility
cassandra_host = params.get("cassandra_host", params.get("graph_host"))
cassandra_username = params.get("cassandra_username", params.get("graph_username"))
cassandra_password = params.get("cassandra_password", params.get("graph_password"))
# Get Cassandra parameters
cassandra_host = params.get("cassandra_host")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
# Resolve configuration with environment variable fallback
hosts, username, password = resolve_cassandra_config(
@ -39,10 +39,10 @@ class Processor(FlowProcessor):
password=cassandra_password
)
# Store resolved configuration
self.graph_host = hosts # Store as list
self.graph_username = username
self.graph_password = password
# Store resolved configuration with proper names
self.cassandra_host = hosts # Store as list
self.cassandra_username = username
self.cassandra_password = password
# Config key for schemas
self.config_key = params.get("config_type", "schema")
@ -82,20 +82,20 @@ class Processor(FlowProcessor):
return
try:
if self.graph_username and self.graph_password:
if self.cassandra_username and self.cassandra_password:
auth_provider = PlainTextAuthProvider(
username=self.graph_username,
password=self.graph_password
username=self.cassandra_username,
password=self.cassandra_password
)
self.cluster = Cluster(
contact_points=self.graph_host,
contact_points=self.cassandra_host,
auth_provider=auth_provider
)
else:
self.cluster = Cluster(contact_points=self.graph_host)
self.cluster = Cluster(contact_points=self.cassandra_host)
self.session = self.cluster.connect()
logger.info(f"Connected to Cassandra cluster at {self.graph_host}")
logger.info(f"Connected to Cassandra cluster at {self.cassandra_host}")
except Exception as e:
logger.error(f"Failed to connect to Cassandra: {e}", exc_info=True)

8
trustgraph-flow/trustgraph/storage/rows/cassandra/write.py
View file

@ -36,10 +36,10 @@ class Processor(Consumer):
input_queue = params.get("input_queue", default_input_queue)
subscriber = params.get("subscriber", default_subscriber)
# Use new parameter names, fall back to old for compatibility
cassandra_host = params.get("cassandra_host", params.get("graph_host"))
cassandra_username = params.get("cassandra_username", params.get("graph_username"))
cassandra_password = params.get("cassandra_password", params.get("graph_password"))
# Get Cassandra parameters
cassandra_host = params.get("cassandra_host")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
# Resolve configuration with environment variable fallback
hosts, username, password = resolve_cassandra_config(

22
trustgraph-flow/trustgraph/storage/triples/cassandra/write.py
View file

@ -26,10 +26,10 @@ class Processor(TriplesStoreService):
id = params.get("id", default_ident)
# Use new parameter names, fall back to old for compatibility
cassandra_host = params.get("cassandra_host", params.get("graph_host"))
cassandra_username = params.get("cassandra_username", params.get("graph_username"))
cassandra_password = params.get("cassandra_password", params.get("graph_password"))
# Get Cassandra parameters
cassandra_host = params.get("cassandra_host")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
# Resolve configuration with environment variable fallback
hosts, username, password = resolve_cassandra_config(
@ -45,9 +45,9 @@ class Processor(TriplesStoreService):
}
)
self.graph_host = hosts
self.username = username
self.password = password
self.cassandra_host = hosts
self.cassandra_username = username
self.cassandra_password = password
self.table = None
async def store_triples(self, message):
@ -59,16 +59,16 @@ class Processor(TriplesStoreService):
self.tg = None
try:
if self.username and self.password:
if self.cassandra_username and self.cassandra_password:
self.tg = TrustGraph(
hosts=self.graph_host,
hosts=self.cassandra_host,
keyspace=message.metadata.user,
table=message.metadata.collection,
username=self.username, password=self.password
username=self.cassandra_username, password=self.cassandra_password
)
else:
self.tg = TrustGraph(
hosts=self.graph_host,
hosts=self.cassandra_host,
keyspace=message.metadata.user,
table=message.metadata.collection,
)