mirror of
https://github.com/csunny/DB-GPT.git
synced 2025-08-03 01:12:15 +00:00
759f7d99cc
Co-authored-by: Florian <fanzhidongyzby@163.com> Co-authored-by: KingSkyLi <15566300566@163.com> Co-authored-by: aries_ckt <916701291@qq.com> Co-authored-by: Fangyin Cheng <staneyffer@gmail.com> Co-authored-by: yvonneyx <zhuyuxin0627@gmail.com>
74 lines
1.7 KiB
Python
74 lines
1.7 KiB
Python
"""Define Classes about Community."""
|
|
import logging
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional
|
|
|
|
from dbgpt.storage.graph_store.base import GraphStoreBase
|
|
from dbgpt.storage.graph_store.graph import Graph
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class Community:
|
|
"""Community class."""
|
|
|
|
id: str
|
|
data: Optional[Graph] = None
|
|
summary: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class CommunityTree:
|
|
"""Represents a community tree."""
|
|
|
|
|
|
class CommunityStoreAdapter(ABC):
|
|
"""Community Store Adapter."""
|
|
|
|
def __init__(self, graph_store: GraphStoreBase):
|
|
"""Initialize Community Store Adapter."""
|
|
self._graph_store = graph_store
|
|
|
|
@property
|
|
def graph_store(self) -> GraphStoreBase:
|
|
"""Get graph store."""
|
|
return self._graph_store
|
|
|
|
@abstractmethod
|
|
async def discover_communities(self, **kwargs) -> List[str]:
|
|
"""Run community discovery."""
|
|
|
|
@abstractmethod
|
|
async def get_community(self, community_id: str) -> Community:
|
|
"""Get community."""
|
|
|
|
|
|
class CommunityMetastore(ABC):
|
|
"""Community metastore class."""
|
|
|
|
@abstractmethod
|
|
def get(self, community_id: str) -> Community:
|
|
"""Get community."""
|
|
|
|
@abstractmethod
|
|
def list(self) -> List[Community]:
|
|
"""Get all communities."""
|
|
|
|
@abstractmethod
|
|
async def search(self, query: str) -> List[Community]:
|
|
"""Search communities relevant to query."""
|
|
|
|
@abstractmethod
|
|
async def save(self, communities: List[Community]):
|
|
"""Save communities."""
|
|
|
|
@abstractmethod
|
|
async def truncate(self):
|
|
"""Truncate all communities."""
|
|
|
|
@abstractmethod
|
|
def drop(self):
|
|
"""Drop community metastore."""
|