from typing import List, Dict, Any from config.settings import databases from providers.base import MetadataProvider from providers.hive import HiveMetadataProvider from providers.impala import ImpalaMetadataProvider from providers.mysql import MySQLMetadataProvider from services.schema.schema_engine import SchemaEngine class MetadataService: def __init__(self): self.providers: Dict[str, MetadataProvider] = {} for db_cfg in databases: name = db_cfg["name"] db_type = db_cfg["type"] try: if db_type == "hive": self.providers[db_type] = HiveMetadataProvider(db_cfg) elif db_type == "impala": self.providers[db_type] = ImpalaMetadataProvider(db_cfg) elif db_type == "mysql": self.providers[db_type] = MySQLMetadataProvider(db_cfg) else: print(f"⚠️ Unsupported DB type: {db_type}") database_list = self.providers[db_type].get_databases() print( str(database_list)) # tables = self.providers[db_type].get_tables(database_list[0]) # print( str(tables)) # columns = self.providers[db_type].get_columns(tables[0], database_list[0]) # print( str(columns)) except Exception as e: print(f"❌ Failed to init {name}: {e}") def get_databases(self, source: str) -> List[str]: provider = self.providers.get(source) return provider.get_databases() if provider else [] def get_tables(self, source: str, database: str) -> List[str]: provider = self.providers.get(source) return provider.get_tables(database) if provider else [] def get_columns(self, source: str, database: str, table: str) -> List[Dict[str, str]]: provider = self.providers.get(source) return provider.get_columns(table, database) if provider else [] def list_sources(self) -> List[str]: return list(self.providers.keys()) def build_mschema(self, source: str, database: str) -> str: provider = self.providers.get(source) return SchemaEngine(schema=database, metadata_provider=provider, db_name=database, include_tables=None).mschema.to_mschema() if __name__ == "__main__": svc = MetadataService() print(svc.build_mschema("hive", "default"))