metadata_service.py 2.4 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. from typing import List, Dict, Any
  2. from config.settings import databases
  3. from providers.base import MetadataProvider
  4. from providers.hive import HiveMetadataProvider
  5. from providers.impala import ImpalaMetadataProvider
  6. from providers.mysql import MySQLMetadataProvider
  7. from services.schema.schema_engine import SchemaEngine
  8. class MetadataService:
  9. def __init__(self):
  10. self.providers: Dict[str, MetadataProvider] = {}
  11. for db_cfg in databases:
  12. name = db_cfg["name"]
  13. db_type = db_cfg["type"]
  14. try:
  15. if db_type == "hive":
  16. self.providers[db_type] = HiveMetadataProvider(db_cfg)
  17. elif db_type == "impala":
  18. self.providers[db_type] = ImpalaMetadataProvider(db_cfg)
  19. elif db_type == "mysql":
  20. self.providers[db_type] = MySQLMetadataProvider(db_cfg)
  21. else:
  22. print(f"⚠️ Unsupported DB type: {db_type}")
  23. database_list = self.providers[db_type].get_databases()
  24. print( str(database_list))
  25. # tables = self.providers[db_type].get_tables(database_list[0])
  26. # print( str(tables))
  27. # columns = self.providers[db_type].get_columns(tables[0], database_list[0])
  28. # print( str(columns))
  29. except Exception as e:
  30. print(f"❌ Failed to init {name}: {e}")
  31. def get_databases(self, source: str) -> List[str]:
  32. provider = self.providers.get(source)
  33. return provider.get_databases() if provider else []
  34. def get_tables(self, source: str, database: str) -> List[str]:
  35. provider = self.providers.get(source)
  36. return provider.get_tables(database) if provider else []
  37. def get_columns(self, source: str, database: str, table: str) -> List[Dict[str, str]]:
  38. provider = self.providers.get(source)
  39. return provider.get_columns(table, database) if provider else []
  40. def list_sources(self) -> List[str]:
  41. return list(self.providers.keys())
  42. def build_mschema(self, source: str, database: str) -> str:
  43. provider = self.providers.get(source)
  44. return SchemaEngine(schema=database, metadata_provider=provider, db_name=database, include_tables=None).mschema.to_mschema()
  45. if __name__ == "__main__":
  46. svc = MetadataService()
  47. print(svc.build_mschema("hive", "default"))