diff --git a/trustgraph-base/trustgraph/base/cassandra_config.py b/trustgraph-base/trustgraph/base/cassandra_config.py index 34875571..46a1745d 100644 --- a/trustgraph-base/trustgraph/base/cassandra_config.py +++ b/trustgraph-base/trustgraph/base/cassandra_config.py @@ -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) \ No newline at end of file diff --git a/trustgraph-flow/trustgraph/query/objects/cassandra/service.py b/trustgraph-flow/trustgraph/query/objects/cassandra/service.py index b71b0f06..2d357ff7 100644 --- a/trustgraph-flow/trustgraph/query/objects/cassandra/service.py +++ b/trustgraph-flow/trustgraph/query/objects/cassandra/service.py @@ -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', diff --git a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py index b38532d0..a0dde295 100755 --- a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py +++ b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py @@ -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 diff --git a/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py b/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py index 0d7fafc6..8cc75318 100644 --- a/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/objects/cassandra/write.py @@ -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) diff --git a/trustgraph-flow/trustgraph/storage/rows/cassandra/write.py b/trustgraph-flow/trustgraph/storage/rows/cassandra/write.py index 343a8dfd..ef79e605 100755 --- a/trustgraph-flow/trustgraph/storage/rows/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/rows/cassandra/write.py @@ -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( diff --git a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py index b83d7cd6..06e8f4e0 100755 --- a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py @@ -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, )