base.py 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. from abc import ABC, abstractmethod
  2. from typing import List, Dict, Any
  3. class MetadataProvider(ABC):
  4. def __init__(self, config: Dict[str, Any]):
  5. self.config = config
  6. self.name = config["name"]
  7. def get_type(self) -> str:
  8. return self.config["type"]
  9. @abstractmethod
  10. def has_table(self, table_name:str, schema:str) -> bool:
  11. """检查表是否存在"""
  12. pass
  13. @abstractmethod
  14. def get_table_comment(self, table_name:str, schema:str) -> str:
  15. """检查表是否存在"""
  16. pass
  17. def get_pk_constraint(self, table_name:str, schema:str) -> Dict:
  18. return {}
  19. def get_unique_constraints(self, table_name:str, schema:str) -> Dict:
  20. return {}
  21. def get_foreign_keys(self, table_name:str, schema:str) -> list[Dict]:
  22. return []
  23. @abstractmethod
  24. def fectch_distinct_values(self, column_name: str, table_name:str, schema:str, max_num: int = 5) -> list[str]:
  25. """检查表是否存在"""
  26. pass
  27. @abstractmethod
  28. def get_databases(self) -> List[str]:
  29. """返回所有数据库/Schema 名称"""
  30. pass
  31. @abstractmethod
  32. def get_tables(self, database: str) -> List[str]:
  33. """返回某数据库下的所有表名"""
  34. pass
  35. @abstractmethod
  36. def get_columns(self, table: str, database: str) -> List[Dict[str, Any]]:
  37. """
  38. 返回字段列表,每个字段为 dict:
  39. {
  40. "name": "id",
  41. "type": "INT",
  42. "comment": "用户ID"
  43. }
  44. """
  45. pass