Coverage for src/qdrant_loader_core/graph/base.py: 100%

1from __future__ import annotations 

2 

3from abc import ABC, abstractmethod 

4from typing import Any 

5 

6from .models import GraphEdge, GraphNode, SubGraph 

7 

8 

9class GraphStore(ABC): 

10 """ 

11 Backend-agnostic Graph Store interface. 

12 Supports FalkorDB, Neptune, etc. 

13 """ 

14 

15 @abstractmethod 

16 async def upsert_node(self, node: GraphNode) -> None: 

17 """Insert or update a node""" 

18 raise NotImplementedError 

19 

20 @abstractmethod 

21 async def upsert_edge(self, edge: GraphEdge) -> None: 

22 """Insert or update an edge""" 

23 raise NotImplementedError 

24 

25 @abstractmethod 

26 async def upsert_nodes_batch(self, nodes: list[GraphNode]) -> None: 

27 """Batch insert/update nodes""" 

28 raise NotImplementedError 

29 

30 @abstractmethod 

31 async def upsert_edges_batch(self, edges: list[GraphEdge]) -> None: 

32 """Batch insert/update edges""" 

33 raise NotImplementedError 

34 

35 @abstractmethod 

36 async def neighbors( 

37 self, 

38 node_id: str, 

39 depth: int, 

40 edge_types: list[str] | None = None, 

41 project: str | None = None, 

42 ) -> SubGraph: 

43 """ 

44 Get neighbors up to a certain depth. 

45 Optionally filter by edge types. 

46 """ 

47 raise NotImplementedError 

48 

49 @abstractmethod 

50 async def query_cypher( 

51 self, 

52 cypher: str, 

53 params: dict[str, Any], 

54 ) -> list[list[Any]]: 

55 """ 

56 Execute Cypher query (for graph DBs that support it). 

57 Returns rows as returned by the underlying driver (row-oriented, 

58 one list per row in RETURN-clause order), not column-keyed dicts. 

59 """ 

60 raise NotImplementedError