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

feat: Add singleton way of working to graph databases

Browse source 
Added signleton type of functioning to graph databases

Feature
This commit is contained in:
Igor Ilic 2024-11-28 22:15:53 +01:00
parent be6eebfbb1
commit be5cb86cb2
1 changed files with 27 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
cognee/infrastructure/databases/graph/get_graph_engine.py
View file

@ -1,4 +1,5 @@
"""Factory function to get the appropriate graph client based on the graph type.""" """Factory function to get the appropriate graph client based on the graph type."""
from functools import lru_cache
from .config import get_graph_config from .config import get_graph_config
from .graph_db_interface import GraphDBInterface from .graph_db_interface import GraphDBInterface
@ -6,6 +7,22 @@ from .graph_db_interface import GraphDBInterface
async def get_graph_engine() -> GraphDBInterface: async def get_graph_engine() -> GraphDBInterface:
"""Factory function to get the appropriate graph client based on the graph type.""" """Factory function to get the appropriate graph client based on the graph type."""
graph_client = create_graph_engine()
# Async functions can't be cached. After creating and caching the graph engine
# handle all necessary async operations for different graph types bellow.
config = get_graph_config()
# Handle loading of graph for NetworkX
if config.graph_database_provider.lower() == "networkx" and graph_client.graph is None:
await graph_client.load_graph_from_file()
return graph_client
@lru_cache
def create_graph_engine() -> GraphDBInterface:
"""Factory function to create the appropriate graph client based on the graph type."""
config = get_graph_config() config = get_graph_config()
if config.graph_database_provider == "neo4j": if config.graph_database_provider == "neo4j":
@ -38,7 +55,4 @@ async def get_graph_engine() -> GraphDBInterface :
from .networkx.adapter import NetworkXAdapter from .networkx.adapter import NetworkXAdapter
graph_client = NetworkXAdapter(filename=config.graph_file_path) graph_client = NetworkXAdapter(filename=config.graph_file_path)
if graph_client.graph is None:
await graph_client.load_graph_from_file()
return graph_client return graph_client