apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-06-09 14:55:13 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Fixing more Cassandra consistency issues (#488)

Browse source 
* Fixing more Cassandra work

* Fix tests
This commit is contained in:
cybermaggedon 2025-09-04 00:58:11 +01:00 • committed by GitHub
parent ccaec88a72
commit 85e669c763
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 196 additions and 208 deletions
Download patch file Download diff file Expand all files Collapse all files

36
tests/integration/test_cassandra_config_end_to_end.py
View file

@ -107,7 +107,7 @@ class TestEndToEndConfigurationFlow:
# Verify KnowledgeTableStore was created with env config
mock_table_store.assert_called_once_with(
cassandra_host=['kg-host1', 'kg-host2', 'kg-host3', 'kg-host4'],
cassandra_user='kg-user',
cassandra_username='kg-user',
cassandra_password='kg-pass',
keyspace='knowledge'
)
@ -177,9 +177,9 @@ class TestConfigurationPriorityEndToEnd:
# Verify mixed configuration
mock_table_store.assert_called_once_with(
cassandra_host=['partial-host'], # From parameter
cassandra_user='fallback-user', # From environment
cassandra_password='fallback-pass', # From environment
cassandra_host=['partial-host'], # From parameter
cassandra_username='fallback-user', # From environment
cassandra_password='fallback-pass', # From environment
keyspace='knowledge'
)
@ -218,19 +218,19 @@ class TestConfigurationPriorityEndToEnd:
assert 'auth_provider' not in call_args.kwargs # No auth with default config
class TestBackwardCompatibilityEndToEnd:
"""Test backward compatibility with old parameter names end-to-end."""
class TestNoBackwardCompatibilityEndToEnd:
"""Test that backward compatibility with old parameter names is removed."""
@pytest.mark.asyncio
@patch('trustgraph.direct.cassandra.Cluster')
async def test_old_graph_params_still_work_end_to_end(self, mock_cluster):
"""Test that old graph_* parameters still work end-to-end."""
async def test_old_graph_params_no_longer_work_end_to_end(self, mock_cluster):
"""Test that old graph_* parameters no longer work end-to-end."""
mock_cluster_instance = MagicMock()
mock_session = MagicMock()
mock_cluster_instance.connect.return_value = mock_session
mock_cluster.return_value = mock_cluster_instance
# Use old parameter names
# Use old parameter names (should be ignored)
processor = TriplesWriter(
taskgroup=MagicMock(),
graph_host='legacy-host',
@ -246,30 +246,30 @@ class TestBackwardCompatibilityEndToEnd:
await processor.store_triples(mock_message)
# Should work with legacy parameters
# Should use defaults since old parameters are not recognized
mock_cluster.assert_called_once()
call_args = mock_cluster.call_args
assert call_args.args[0] == ['legacy-host']
assert 'auth_provider' in call_args.kwargs # Should have auth since credentials provided
assert call_args.args[0] == ['cassandra'] # Default, not legacy-host
assert 'auth_provider' not in call_args.kwargs # No auth since no valid credentials
@patch('trustgraph.storage.knowledge.store.KnowledgeTableStore')
def test_old_cassandra_user_param_still_works_end_to_end(self, mock_table_store):
"""Test that old cassandra_user parameter still works end-to-end."""
def test_old_cassandra_user_param_no_longer_works_end_to_end(self, mock_table_store):
"""Test that old cassandra_user parameter no longer works."""
mock_store_instance = MagicMock()
mock_table_store.return_value = mock_store_instance
# Use old cassandra_user parameter
# Use old cassandra_user parameter (should be ignored)
processor = KgStore(
taskgroup=MagicMock(),
cassandra_host='legacy-kg-host',
cassandra_user='legacy-kg-user', # Old parameter name
cassandra_user='legacy-kg-user', # Old parameter name - not supported
cassandra_password='legacy-kg-pass'
)
# Should work with old parameter name
# cassandra_user should be ignored, only cassandra_username works
mock_table_store.assert_called_once_with(
cassandra_host=['legacy-kg-host'],
cassandra_user='legacy-kg-user',
cassandra_username=None, # Should be None since cassandra_user is not recognized
cassandra_password='legacy-kg-pass',
keyspace='knowledge'
)

4
tests/integration/test_objects_cassandra_integration.py
View file

@ -294,8 +294,8 @@ class TestObjectsCassandraIntegration:
async def test_authentication_handling(self, processor_with_mocks):
"""Test Cassandra authentication"""
processor, mock_cluster, mock_session = processor_with_mocks
processor.graph_username = "cassandra_user"
processor.graph_password = "cassandra_pass"
processor.cassandra_username = "cassandra_user"
processor.cassandra_password = "cassandra_pass"
with patch('trustgraph.storage.objects.cassandra.write.Cluster') as mock_cluster_class:
with patch('trustgraph.storage.objects.cassandra.write.PlainTextAuthProvider') as mock_auth:

29
tests/unit/test_base/test_cassandra_config.py
View file

@ -257,8 +257,8 @@ class TestGetCassandraConfigFromParams:
assert username == 'new-user'
assert password == 'new-pass'
def test_backward_compatibility_graph_params(self):
"""Test backward compatibility with old graph_* parameter names."""
def test_no_backward_compatibility_graph_params(self):
"""Test that old graph_* parameter names are no longer supported."""
params = {
'graph_host': 'old-host',
'graph_username': 'old-user',
@ -267,26 +267,27 @@ class TestGetCassandraConfigFromParams:
hosts, username, password = get_cassandra_config_from_params(params)
assert hosts == ['old-host']
assert username == 'old-user'
assert password == 'old-pass'
# Should use defaults since graph_* params are not recognized
assert hosts == ['cassandra'] # Default
assert username is None
assert password is None
def test_old_cassandra_user_compatibility(self):
"""Test backward compatibility with cassandra_user (vs cassandra_username)."""
def test_no_old_cassandra_user_compatibility(self):
"""Test that cassandra_user is no longer supported (must be cassandra_username)."""
params = {
'cassandra_host': 'compat-host',
'cassandra_user': 'compat-user', # Old name
'cassandra_user': 'compat-user', # Old name - not supported
'cassandra_password': 'compat-pass'
}
hosts, username, password = get_cassandra_config_from_params(params)
assert hosts == ['compat-host']
assert username == 'compat-user'
assert username is None # cassandra_user is not recognized
assert password == 'compat-pass'
def test_parameter_precedence(self):
"""Test that new parameter names take precedence over old ones."""
def test_only_new_parameters_work(self):
"""Test that only new parameter names are recognized."""
params = {
'cassandra_host': 'new-host',
'graph_host': 'old-host',
@ -299,9 +300,9 @@ class TestGetCassandraConfigFromParams:
hosts, username, password = get_cassandra_config_from_params(params)
assert hosts == ['new-host'] # New takes precedence
assert username == 'new-user' # New takes precedence
assert password == 'new-pass' # New takes precedence
assert hosts == ['new-host'] # Only cassandra_* params work
assert username == 'new-user' # Only cassandra_* params work
assert password == 'new-pass' # Only cassandra_* params work
def test_empty_params_with_env_fallback(self):
"""Test that empty params falls back to environment variables."""

2
tests/unit/test_cores/test_knowledge_manager.py
View file

@ -56,7 +56,7 @@ def knowledge_manager(mock_flow_config):
with patch('trustgraph.cores.knowledge.KnowledgeTableStore') as mock_store_class:
manager = KnowledgeManager(
cassandra_host=["localhost"],
cassandra_user="test_user",
cassandra_username="test_user",
cassandra_password="test_pass",
keyspace="test_keyspace",
flow_config=mock_flow_config

24
tests/unit/test_query/test_triples_cassandra_query.py
View file

@ -83,7 +83,7 @@ class TestCassandraQueryProcessor:
processor = Processor(
taskgroup=MagicMock(),
id='test-cassandra-query',
graph_host='localhost'
cassandra_host='localhost'
)
# Create query request with all SPO values
@ -122,9 +122,9 @@ class TestCassandraQueryProcessor:
processor = Processor(taskgroup=taskgroup_mock)
assert processor.graph_host == ['cassandra'] # Updated default
assert processor.username is None
assert processor.password is None
assert processor.cassandra_host == ['cassandra'] # Updated default
assert processor.cassandra_username is None
assert processor.cassandra_password is None
assert processor.table is None
def test_processor_initialization_with_custom_params(self):
@ -133,14 +133,14 @@ class TestCassandraQueryProcessor:
processor = Processor(
taskgroup=taskgroup_mock,
graph_host='cassandra.example.com',
graph_username='queryuser',
graph_password='querypass'
cassandra_host='cassandra.example.com',
cassandra_username='queryuser',
cassandra_password='querypass'
)
assert processor.graph_host == ['cassandra.example.com']
assert processor.username == 'queryuser'
assert processor.password == 'querypass'
assert processor.cassandra_host == ['cassandra.example.com']
assert processor.cassandra_username == 'queryuser'
assert processor.cassandra_password == 'querypass'
assert processor.table is None
@pytest.mark.asyncio
@ -387,8 +387,8 @@ class TestCassandraQueryProcessor:
processor = Processor(
taskgroup=MagicMock(),
graph_username='authuser',
graph_password='authpass'
cassandra_username='authuser',
cassandra_password='authpass'
)
query = TriplesQueryRequest(

79
tests/unit/test_storage/test_cassandra_config_integration.py
View file

@ -30,9 +30,9 @@ class TestTriplesWriterConfiguration:
with patch.dict(os.environ, env_vars, clear=True):
processor = TriplesWriter(taskgroup=MagicMock())
assert processor.graph_host == ['env-host1', 'env-host2']
assert processor.username == 'env-user'
assert processor.password == 'env-pass'
assert processor.cassandra_host == ['env-host1', 'env-host2']
assert processor.cassandra_username == 'env-user'
assert processor.cassandra_password == 'env-pass'
@patch('trustgraph.direct.cassandra.TrustGraph')
def test_parameter_override_environment(self, mock_trust_graph):
@ -51,13 +51,13 @@ class TestTriplesWriterConfiguration:
cassandra_password='param-pass'
)
assert processor.graph_host == ['param-host1', 'param-host2']
assert processor.username == 'param-user'
assert processor.password == 'param-pass'
assert processor.cassandra_host == ['param-host1', 'param-host2']
assert processor.cassandra_username == 'param-user'
assert processor.cassandra_password == 'param-pass'
@patch('trustgraph.direct.cassandra.TrustGraph')
def test_backward_compatibility_graph_params(self, mock_trust_graph):
"""Test backward compatibility with old graph_* parameter names."""
def test_no_backward_compatibility_graph_params(self, mock_trust_graph):
"""Test that old graph_* parameter names are no longer supported."""
processor = TriplesWriter(
taskgroup=MagicMock(),
graph_host='compat-host',
@ -65,9 +65,10 @@ class TestTriplesWriterConfiguration:
graph_password='compat-pass'
)
assert processor.graph_host == ['compat-host']
assert processor.username == 'compat-user'
assert processor.password == 'compat-pass'
# Should use defaults since graph_* params are not recognized
assert processor.cassandra_host == ['cassandra'] # Default
assert processor.cassandra_username is None
assert processor.cassandra_password is None
@patch('trustgraph.direct.cassandra.TrustGraph')
def test_default_configuration(self, mock_trust_graph):
@ -75,9 +76,9 @@ class TestTriplesWriterConfiguration:
with patch.dict(os.environ, {}, clear=True):
processor = TriplesWriter(taskgroup=MagicMock())
assert processor.graph_host == ['cassandra']
assert processor.username is None
assert processor.password is None
assert processor.cassandra_host == ['cassandra']
assert processor.cassandra_username is None
assert processor.cassandra_password is None
class TestObjectsWriterConfiguration:
@ -98,9 +99,9 @@ class TestObjectsWriterConfiguration:
with patch.dict(os.environ, env_vars, clear=True):
processor = ObjectsWriter(taskgroup=MagicMock())
assert processor.graph_host == ['obj-env-host1', 'obj-env-host2']
assert processor.graph_username == 'obj-env-user'
assert processor.graph_password == 'obj-env-pass'
assert processor.cassandra_host == ['obj-env-host1', 'obj-env-host2']
assert processor.cassandra_username == 'obj-env-user'
assert processor.cassandra_password == 'obj-env-pass'
@patch('trustgraph.storage.objects.cassandra.write.Cluster')
def test_cassandra_connection_with_hosts_list(self, mock_cluster):
@ -174,24 +175,24 @@ class TestTriplesQueryConfiguration:
with patch.dict(os.environ, env_vars, clear=True):
processor = TriplesQuery(taskgroup=MagicMock())
assert processor.graph_host == ['query-env-host1', 'query-env-host2']
assert processor.username == 'query-env-user'
assert processor.password == 'query-env-pass'
assert processor.cassandra_host == ['query-env-host1', 'query-env-host2']
assert processor.cassandra_username == 'query-env-user'
assert processor.cassandra_password == 'query-env-pass'
@patch('trustgraph.direct.cassandra.TrustGraph')
def test_mixed_old_new_parameters(self, mock_trust_graph):
"""Test mixing old and new parameter names (new should win)."""
def test_only_new_parameters_work(self, mock_trust_graph):
"""Test that only new parameters work."""
processor = TriplesQuery(
taskgroup=MagicMock(),
cassandra_host='new-host',
graph_host='old-host',
graph_host='old-host', # Should be ignored
cassandra_username='new-user',
graph_username='old-user'
graph_username='old-user' # Should be ignored
)
# New parameters should take precedence
assert processor.graph_host == ['new-host']
assert processor.username == 'new-user'
# Only new parameters should work
assert processor.cassandra_host == ['new-host']
assert processor.cassandra_username == 'new-user'
class TestKgStoreConfiguration:
@ -215,7 +216,7 @@ class TestKgStoreConfiguration:
# Verify KnowledgeTableStore was called with resolved config
mock_table_store.assert_called_once_with(
cassandra_host=['kg-env-host1', 'kg-env-host2', 'kg-env-host3'],
cassandra_user='kg-env-user',
cassandra_username='kg-env-user',
cassandra_password='kg-env-pass',
keyspace='knowledge'
)
@ -236,28 +237,28 @@ class TestKgStoreConfiguration:
# Verify KnowledgeTableStore was called with explicit config
mock_table_store.assert_called_once_with(
cassandra_host=['explicit-host'],
cassandra_user='explicit-user',
cassandra_username='explicit-user',
cassandra_password='explicit-pass',
keyspace='knowledge'
)
@patch('trustgraph.storage.knowledge.store.KnowledgeTableStore')
def test_backward_compatibility_cassandra_user(self, mock_table_store):
"""Test backward compatibility with cassandra_user parameter."""
def test_no_backward_compatibility_cassandra_user(self, mock_table_store):
"""Test that cassandra_user parameter is no longer supported."""
mock_store_instance = MagicMock()
mock_table_store.return_value = mock_store_instance
processor = KgStore(
taskgroup=MagicMock(),
cassandra_host='compat-host',
cassandra_user='compat-user', # Old parameter name
cassandra_user='compat-user', # Old parameter name - should be ignored
cassandra_password='compat-pass'
)
# Should still work with old parameter name
# cassandra_user should be ignored
mock_table_store.assert_called_once_with(
cassandra_host=['compat-host'],
cassandra_user='compat-user',
cassandra_username=None, # Should be None since cassandra_user is ignored
cassandra_password='compat-pass',
keyspace='knowledge'
)
@ -274,7 +275,7 @@ class TestKgStoreConfiguration:
# Should use defaults
mock_table_store.assert_called_once_with(
cassandra_host=['cassandra'],
cassandra_user=None,
cassandra_username=None,
cassandra_password=None,
keyspace='knowledge'
)
@ -396,9 +397,9 @@ class TestConfigurationPriorityIntegration:
# Password not provided - should fall back to env
)
assert processor.graph_host == ['cli-host1', 'cli-host2'] # From CLI
assert processor.username == 'cli-user' # From CLI
assert processor.password == 'env-pass' # From env
assert processor.cassandra_host == ['cli-host1', 'cli-host2'] # From CLI
assert processor.cassandra_username == 'cli-user' # From CLI
assert processor.cassandra_password == 'env-pass' # From env
@patch('trustgraph.storage.knowledge.store.KnowledgeTableStore')
def test_kg_store_priority_chain(self, mock_table_store):
@ -422,7 +423,7 @@ class TestConfigurationPriorityIntegration:
# Verify correct priority resolution
mock_table_store.assert_called_once_with(
cassandra_host=['param-host'], # From parameter
cassandra_user='env-user', # From environment
cassandra_username='env-user', # From environment
cassandra_password='env-pass', # From environment
keyspace='knowledge'
)

35
tests/unit/test_storage/test_triples_cassandra_storage.py
View file

@ -20,9 +20,9 @@ class TestCassandraStorageProcessor:
with patch.dict('os.environ', {}, clear=True):
processor = Processor(taskgroup=taskgroup_mock)
assert processor.graph_host == ['cassandra'] # Updated default
assert processor.username is None
assert processor.password is None
assert processor.cassandra_host == ['cassandra'] # Updated default
assert processor.cassandra_username is None
assert processor.cassandra_password is None
assert processor.table is None
def test_processor_initialization_with_custom_params(self):
@ -37,9 +37,9 @@ class TestCassandraStorageProcessor:
cassandra_password='testpass'
)
assert processor.graph_host == ['cassandra.example.com']
assert processor.username == 'testuser'
assert processor.password == 'testpass'
assert processor.cassandra_host == ['cassandra.example.com']
assert processor.cassandra_username == 'testuser'
assert processor.cassandra_password == 'testpass'
assert processor.table is None
def test_processor_initialization_with_partial_auth(self):
@ -51,11 +51,11 @@ class TestCassandraStorageProcessor:
cassandra_username='testuser'
)
assert processor.username == 'testuser'
assert processor.password is None
assert processor.cassandra_username == 'testuser'
assert processor.cassandra_password is None
def test_processor_initialization_backward_compatibility(self):
"""Test processor initialization with old graph_* parameters (backward compatibility)"""
def test_processor_no_backward_compatibility(self):
"""Test that old graph_* parameters are no longer supported"""
taskgroup_mock = MagicMock()
processor = Processor(
@ -65,12 +65,13 @@ class TestCassandraStorageProcessor:
graph_password='old-pass'
)
assert processor.graph_host == ['old-host']
assert processor.username == 'old-user'
assert processor.password == 'old-pass'
# Should use defaults since graph_* params are not recognized
assert processor.cassandra_host == ['cassandra'] # Default
assert processor.cassandra_username is None
assert processor.cassandra_password is None
def test_processor_parameter_precedence(self):
"""Test that new cassandra_* parameters take precedence over old graph_* parameters"""
def test_processor_only_new_parameters_work(self):
"""Test that only new cassandra_* parameters work"""
taskgroup_mock = MagicMock()
processor = Processor(
@ -81,8 +82,8 @@ class TestCassandraStorageProcessor:
graph_username='old-user' # Should be ignored
)
assert processor.graph_host == ['new-host'] # New parameter wins
assert processor.username == 'new-user' # New parameter wins
assert processor.cassandra_host == ['new-host'] # Only cassandra_* params work
assert processor.cassandra_username == 'new-user' # Only cassandra_* params work
@pytest.mark.asyncio
@patch('trustgraph.storage.triples.cassandra.write.TrustGraph')

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)

4
trustgraph-flow/trustgraph/config/service/config.py
View file

@ -45,13 +45,13 @@ class Configuration:
# FIXME: Some version vs config race conditions
def __init__(self, push, host, user, password, keyspace):
def __init__(self, push, host, username, password, keyspace):
# External function to respond to update
self.push = push
self.table_store = ConfigTableStore(
host, user, password, keyspace
host, username, password, keyspace
)
async def inc_version(self):

6
trustgraph-flow/trustgraph/config/service/service.py
View file

@ -61,7 +61,7 @@ class Processor(AsyncProcessor):
)
cassandra_host = params.get("cassandra_host", default_cassandra_host)
cassandra_user = params.get("cassandra_user")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
id = params.get("id")
@ -77,7 +77,7 @@ class Processor(AsyncProcessor):
"flow_request_schema": FlowRequest.__name__,
"flow_response_schema": FlowResponse.__name__,
"cassandra_host": cassandra_host,
"cassandra_user": cassandra_user,
"cassandra_username": cassandra_username,
}
)
@ -143,7 +143,7 @@ class Processor(AsyncProcessor):
self.config = Configuration(
host = cassandra_host.split(","),
user = cassandra_user,
username = cassandra_username,
password = cassandra_password,
keyspace = keyspace,
push = self.push

4
trustgraph-flow/trustgraph/cores/knowledge.py
View file

@ -16,12 +16,12 @@ logger = logging.getLogger(__name__)
class KnowledgeManager:
def __init__(
self, cassandra_host, cassandra_user, cassandra_password,
self, cassandra_host, cassandra_username, cassandra_password,
keyspace, flow_config,
):
self.table_store = KnowledgeTableStore(
cassandra_host, cassandra_user, cassandra_password, keyspace
cassandra_host, cassandra_username, cassandra_password, keyspace
)
self.loader_queue = asyncio.Queue(maxsize=20)

4
trustgraph-flow/trustgraph/cores/service.py
View file

@ -50,7 +50,7 @@ class Processor(AsyncProcessor):
)
cassandra_host = params.get("cassandra_host", default_cassandra_host)
cassandra_user = params.get("cassandra_user")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
super(Processor, self).__init__(
@ -90,7 +90,7 @@ class Processor(AsyncProcessor):
self.knowledge = KnowledgeManager(
cassandra_host = cassandra_host.split(","),
cassandra_user = cassandra_user,
cassandra_username = cassandra_username,
cassandra_password = cassandra_password,
keyspace = keyspace,
flow_config = self,

4
trustgraph-flow/trustgraph/librarian/librarian.py
View file

@ -16,7 +16,7 @@ class Librarian:
def __init__(
self,
cassandra_host, cassandra_user, cassandra_password,
cassandra_host, cassandra_username, cassandra_password,
minio_host, minio_access_key, minio_secret_key,
bucket_name, keyspace, load_document,
):
@ -26,7 +26,7 @@ class Librarian:
)
self.table_store = LibraryTableStore(
cassandra_host, cassandra_user, cassandra_password, keyspace
cassandra_host, cassandra_username, cassandra_password, keyspace
)
self.load_document = load_document

6
trustgraph-flow/trustgraph/librarian/service.py
View file

@ -67,7 +67,7 @@ class Processor(AsyncProcessor):
)
cassandra_host = params.get("cassandra_host", default_cassandra_host)
cassandra_user = params.get("cassandra_user")
cassandra_username = params.get("cassandra_username")
cassandra_password = params.get("cassandra_password")
super(Processor, self).__init__(
@ -77,7 +77,7 @@ class Processor(AsyncProcessor):
"minio_host": minio_host,
"minio_access_key": minio_access_key,
"cassandra_host": cassandra_host,
"cassandra_user": cassandra_user,
"cassandra_username": cassandra_username,
}
)
@ -109,7 +109,7 @@ class Processor(AsyncProcessor):
self.librarian = Librarian(
cassandra_host = cassandra_host.split(","),
cassandra_user = cassandra_user,
cassandra_username = cassandra_username,
cassandra_password = cassandra_password,
minio_host = minio_host,
minio_access_key = minio_access_key,

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

4
trustgraph-flow/trustgraph/storage/knowledge/store.py
View file

@ -25,7 +25,7 @@ class Processor(FlowProcessor):
# Use helper to resolve configuration
hosts, username, password = resolve_cassandra_config(
host=params.get("cassandra_host"),
username=params.get("cassandra_user", params.get("cassandra_username")),
username=params.get("cassandra_username"),
password=params.get("cassandra_password")
)
@ -55,7 +55,7 @@ class Processor(FlowProcessor):
self.table_store = KnowledgeTableStore(
cassandra_host = hosts,
cassandra_user = username,
cassandra_username = username,
cassandra_password = password,
keyspace = keyspace,
)

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,
)

6
trustgraph-flow/trustgraph/tables/config.py
View file

@ -17,7 +17,7 @@ class ConfigTableStore:
def __init__(
self,
cassandra_host, cassandra_user, cassandra_password, keyspace,
cassandra_host, cassandra_username, cassandra_password, keyspace,
):
self.keyspace = keyspace
@ -28,10 +28,10 @@ class ConfigTableStore:
if isinstance(cassandra_host, str):
cassandra_host = [h.strip() for h in cassandra_host.split(',')]
if cassandra_user and cassandra_password:
if cassandra_username and cassandra_password:
ssl_context = SSLContext(PROTOCOL_TLSv1_2)
auth_provider = PlainTextAuthProvider(
username=cassandra_user, password=cassandra_password
username=cassandra_username, password=cassandra_password
)
self.cluster = Cluster(
cassandra_host,

6
trustgraph-flow/trustgraph/tables/knowledge.py
View file

@ -17,7 +17,7 @@ class KnowledgeTableStore:
def __init__(
self,
cassandra_host, cassandra_user, cassandra_password, keyspace,
cassandra_host, cassandra_username, cassandra_password, keyspace,
):
self.keyspace = keyspace
@ -28,10 +28,10 @@ class KnowledgeTableStore:
if isinstance(cassandra_host, str):
cassandra_host = [h.strip() for h in cassandra_host.split(',')]
if cassandra_user and cassandra_password:
if cassandra_username and cassandra_password:
ssl_context = SSLContext(PROTOCOL_TLSv1_2)
auth_provider = PlainTextAuthProvider(
username=cassandra_user, password=cassandra_password
username=cassandra_username, password=cassandra_password
)
self.cluster = Cluster(
cassandra_host,

6
trustgraph-flow/trustgraph/tables/library.py
View file

@ -21,7 +21,7 @@ class LibraryTableStore:
def __init__(
self,
cassandra_host, cassandra_user, cassandra_password, keyspace,
cassandra_host, cassandra_username, cassandra_password, keyspace,
):
self.keyspace = keyspace
@ -32,10 +32,10 @@ class LibraryTableStore:
if isinstance(cassandra_host, str):
cassandra_host = [h.strip() for h in cassandra_host.split(',')]
if cassandra_user and cassandra_password:
if cassandra_username and cassandra_password:
ssl_context = SSLContext(PROTOCOL_TLSv1_2)
auth_provider = PlainTextAuthProvider(
username=cassandra_user, password=cassandra_password
username=cassandra_username, password=cassandra_password
)
self.cluster = Cluster(
cassandra_host,