DB-GPT/dbgpt/storage/knowledge_graph/community/factory.py

"""CommunityStoreAdapter factory."""
import logging

from dbgpt.storage.graph_store.base import GraphStoreBase
from dbgpt.storage.graph_store.tugraph_store import TuGraphStore
from dbgpt.storage.knowledge_graph.community.base import CommunityStoreAdapter
from dbgpt.storage.knowledge_graph.community.tugraph_adapter import (
    TuGraphCommunityStoreAdapter,
)

logger = logging.getLogger(__name__)


class CommunityStoreAdapterFactory:
    """Factory for community store adapter."""

    @staticmethod
    def create(graph_store: GraphStoreBase) -> CommunityStoreAdapter:
        """Create a CommunityStoreAdapter instance.

        Args:
            - graph_store_type: graph store type Memory, TuGraph, Neo4j
        """
        if isinstance(graph_store, TuGraphStore):
            return TuGraphCommunityStoreAdapter(graph_store)
        else:
            raise Exception(
                "create community store adapter for %s failed",
                graph_store.__class__.__name__,
            )