feat(graph): add GraphBackendFactory singleton
This commit is contained in:
parent
247ecc86ae
commit
e073ef8716
|
|
@ -1 +1,3 @@
|
|||
# Populated in Task 4 once factory.py exists
|
||||
from .factory import get_graph_backend
|
||||
|
||||
__all__ = ["get_graph_backend"]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,39 @@
|
|||
"""Graph backend factory — returns singleton based on GRAPH_BACKEND env var."""
|
||||
from typing import Optional
|
||||
|
||||
from .base import GraphBackend
|
||||
from ..utils.logger import get_logger
|
||||
|
||||
logger = get_logger('mirofish.graph.factory')
|
||||
|
||||
_backend_instance: Optional[GraphBackend] = None
|
||||
|
||||
|
||||
def get_graph_backend() -> GraphBackend:
|
||||
"""Return the configured graph backend singleton."""
|
||||
global _backend_instance
|
||||
if _backend_instance is not None:
|
||||
return _backend_instance
|
||||
|
||||
from ..config import Config
|
||||
backend_type = Config.GRAPH_BACKEND
|
||||
logger.info(f"Initializing graph backend: {backend_type}")
|
||||
|
||||
if backend_type == "zep":
|
||||
from .zep_backend import ZepBackend
|
||||
_backend_instance = ZepBackend()
|
||||
elif backend_type == "graphiti":
|
||||
from .graphiti_backend import GraphitiBackend
|
||||
_backend_instance = GraphitiBackend()
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Unknown GRAPH_BACKEND='{backend_type}'. Valid values: 'zep', 'graphiti'."
|
||||
)
|
||||
|
||||
return _backend_instance
|
||||
|
||||
|
||||
def reset_graph_backend() -> None:
|
||||
"""Reset singleton (useful for testing)."""
|
||||
global _backend_instance
|
||||
_backend_instance = None
|
||||
|
|
@ -0,0 +1,12 @@
|
|||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_graph_factory_singleton():
|
||||
"""Reset the graph backend singleton before each test to avoid cross-test contamination."""
|
||||
yield
|
||||
try:
|
||||
import backend.app.graph.factory as fmod
|
||||
fmod._backend_instance = None
|
||||
except ImportError:
|
||||
pass
|
||||
|
|
@ -54,6 +54,38 @@ def test_zep_backend_raises_without_key():
|
|||
cfg_mod.Config.ZEP_API_KEY = orig
|
||||
|
||||
|
||||
def test_factory_returns_zep_by_default():
|
||||
import backend.app.graph.factory as fmod
|
||||
import backend.app.config as cfg
|
||||
orig_backend = cfg.Config.GRAPH_BACKEND
|
||||
orig_key = cfg.Config.ZEP_API_KEY
|
||||
try:
|
||||
cfg.Config.GRAPH_BACKEND = "zep"
|
||||
cfg.Config.ZEP_API_KEY = "test-key"
|
||||
fmod._backend_instance = None
|
||||
backend_instance = fmod.get_graph_backend()
|
||||
from backend.app.graph.zep_backend import ZepBackend
|
||||
assert isinstance(backend_instance, ZepBackend)
|
||||
finally:
|
||||
cfg.Config.GRAPH_BACKEND = orig_backend
|
||||
cfg.Config.ZEP_API_KEY = orig_key
|
||||
fmod._backend_instance = None
|
||||
|
||||
|
||||
def test_factory_raises_on_unknown_backend():
|
||||
import backend.app.graph.factory as fmod
|
||||
import backend.app.config as cfg
|
||||
orig = cfg.Config.GRAPH_BACKEND
|
||||
try:
|
||||
cfg.Config.GRAPH_BACKEND = "unknown"
|
||||
fmod._backend_instance = None
|
||||
with pytest.raises(ValueError, match="Unknown GRAPH_BACKEND"):
|
||||
fmod.get_graph_backend()
|
||||
finally:
|
||||
cfg.Config.GRAPH_BACKEND = orig
|
||||
fmod._backend_instance = None
|
||||
|
||||
|
||||
def test_config_graphiti_errors_when_missing():
|
||||
import backend.app.config as cfg_mod
|
||||
orig_backend = cfg_mod.Config.GRAPH_BACKEND
|
||||
|
|
|
|||
Loading…
Reference in New Issue