gmakstutis/graphiti
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement build_indices_and_constraints for Kuzu and Neptune drivers

Browse source 
Both Kuzu and Neptune drivers were missing the build_indices_and_constraints
method required by the abstract GraphDriver base class. Kuzu uses a no-op
implementation since indices are created during schema setup. Neptune delegates
to its existing OpenSearch (AOSS) index creation logic.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Daniel Chalef 2025-11-05 08:34:53 -08:00
parent c144ff5995
commit 3fc86f3dec
2 changed files with 12 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
graphiti_core/driver/kuzu_driver.py
View file

@ -140,6 +140,12 @@ class KuzuDriver(GraphDriver):
def delete_all_indexes(self, database_: str): def delete_all_indexes(self, database_: str):
pass pass
async def build_indices_and_constraints(self, delete_existing: bool = False):
# Kuzu doesn't support dynamic index creation like Neo4j or FalkorDB
# Schema and indices are created during setup_schema()
# This method is required by the abstract base class but is a no-op for Kuzu
pass
def setup_schema(self): def setup_schema(self):
conn = kuzu.Connection(self.db) conn = kuzu.Connection(self.db)
conn.execute(SCHEMA_QUERIES) conn.execute(SCHEMA_QUERIES)

6
graphiti_core/driver/neptune_driver.py
View file

@ -243,6 +243,12 @@ class NeptuneDriver(GraphDriver):
if client.indices.exists(index=index_name): if client.indices.exists(index=index_name):
client.indices.delete(index=index_name) client.indices.delete(index=index_name)
async def build_indices_and_constraints(self, delete_existing: bool = False):
# Neptune uses OpenSearch (AOSS) for indexing
if delete_existing:
await self.delete_aoss_indices()
await self.create_aoss_indices()
def run_aoss_query(self, name: str, query_text: str, limit: int = 10) -> dict[str, Any]: def run_aoss_query(self, name: str, query_text: str, limit: int = 10) -> dict[str, Any]:
for index in aoss_indices: for index in aoss_indices:
if name.lower() == index['index_name']: if name.lower() == index['index_name']: