| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556 |
- from abc import ABC, abstractmethod
- from typing import List, Dict, Any
- class MetadataProvider(ABC):
- def __init__(self, config: Dict[str, Any]):
- self.config = config
- self.name = config["name"]
- def get_type(self) -> str:
- return self.config["type"]
- @abstractmethod
- def has_table(self, table_name:str, schema:str) -> bool:
- """检查表是否存在"""
- pass
- @abstractmethod
- def get_table_comment(self, table_name:str, schema:str) -> str:
- """检查表是否存在"""
- pass
- def get_pk_constraint(self, table_name:str, schema:str) -> Dict:
- return {}
- def get_unique_constraints(self, table_name:str, schema:str) -> Dict:
- return {}
- def get_foreign_keys(self, table_name:str, schema:str) -> list[Dict]:
- return []
- @abstractmethod
- def fectch_distinct_values(self, column_name: str, table_name:str, schema:str, max_num: int = 5) -> list[str]:
- """检查表是否存在"""
- pass
- @abstractmethod
- def get_databases(self) -> List[str]:
- """返回所有数据库/Schema 名称"""
- pass
- @abstractmethod
- def get_tables(self, database: str) -> List[str]:
- """返回某数据库下的所有表名"""
- pass
- @abstractmethod
- def get_columns(self, table: str, database: str) -> List[Dict[str, Any]]:
- """
- 返回字段列表,每个字段为 dict:
- {
- "name": "id",
- "type": "INT",
- "comment": "用户ID"
- }
- """
- pass
|