| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556 |
- 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"))
|