92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
# #region mapping_service [C:5] [TYPE Module] [SEMANTICS mapping, database, superset, fuzzy, suggestion]
|
|
#
|
|
# @BRIEF Orchestrates database fetching and fuzzy matching suggestions.
|
|
# @LAYER Service
|
|
# @PRE source/target environment identifiers are provided by caller.
|
|
# @POST Exposes stateless mapping suggestion orchestration over configured environments.
|
|
# @SIDE_EFFECT Performs remote metadata reads through Superset API clients.
|
|
# @DATA_CONTRACT Input[source_env_id: str, target_env_id: str] -> Output[List[Dict]]
|
|
# @RELATION DEPENDS_ON -> SupersetClient
|
|
# @RELATION DEPENDS_ON -> suggest_mappings
|
|
#
|
|
# @INVARIANT Suggestions are based on database names.
|
|
|
|
|
|
from ..core.logger import belief_scope
|
|
from ..core.superset_client import SupersetClient
|
|
from ..core.utils.matching import suggest_mappings
|
|
|
|
|
|
# #region MappingService [C:3] [TYPE Class]
|
|
# @BRIEF Service for handling database mapping logic.
|
|
# @PRE config_manager exposes get_environments() with environment objects containing id.
|
|
# @POST Provides client resolution and mapping suggestion methods.
|
|
# @SIDE_EFFECT Instantiates Superset clients and performs upstream metadata reads.
|
|
# @DATA_CONTRACT Input[config_manager] -> Output[List[Dict]]
|
|
# @RELATION DEPENDS_ON -> SupersetClient
|
|
# @RELATION DEPENDS_ON -> suggest_mappings
|
|
class MappingService:
|
|
# region init [TYPE Function]
|
|
# @PURPOSE: Initializes the mapping service with a config manager.
|
|
# @PRE config_manager is provided.
|
|
# @PARAM config_manager (ConfigManager) - The configuration manager.
|
|
# @POST Service is initialized.
|
|
# @RELATION DEPENDS_ON -> MappingService
|
|
def __init__(self, config_manager):
|
|
with belief_scope("MappingService.__init__"):
|
|
self.config_manager = config_manager
|
|
|
|
# endregion init
|
|
|
|
# region _get_client [TYPE Function]
|
|
# @PURPOSE: Helper to get an initialized SupersetClient for an environment.
|
|
# @PARAM env_id (str) - The ID of the environment.
|
|
# @PRE environment must exist in config.
|
|
# @POST Returns an initialized SupersetClient.
|
|
# @RETURN SupersetClient - Initialized client.
|
|
# @RELATION CALLS -> SupersetClient
|
|
def _get_client(self, env_id: str) -> SupersetClient:
|
|
with belief_scope("MappingService._get_client", f"env_id={env_id}"):
|
|
envs = self.config_manager.get_environments()
|
|
env = next((e for e in envs if e.id == env_id), None)
|
|
if not env:
|
|
raise ValueError(f"Environment {env_id} not found")
|
|
|
|
return SupersetClient(env)
|
|
|
|
# endregion _get_client
|
|
|
|
# region get_suggestions [TYPE Function]
|
|
# @PURPOSE: Fetches databases from both environments and returns fuzzy matching suggestions.
|
|
# @PARAM source_env_id (str) - Source environment ID.
|
|
# @PARAM target_env_id (str) - Target environment ID.
|
|
# @PRE Both environments must be accessible.
|
|
# @POST Returns fuzzy-matched database suggestions.
|
|
# @RETURN List[Dict] - Suggested mappings.
|
|
# @RELATION CALLS -> _get_client
|
|
# @RELATION CALLS -> suggest_mappings
|
|
async def get_suggestions(
|
|
self, source_env_id: str, target_env_id: str
|
|
) -> list[dict]:
|
|
with belief_scope(
|
|
"MappingService.get_suggestions",
|
|
f"source={source_env_id}, target={target_env_id}",
|
|
):
|
|
"""
|
|
Get suggested mappings between two environments.
|
|
"""
|
|
source_client = self._get_client(source_env_id)
|
|
target_client = self._get_client(target_env_id)
|
|
|
|
source_dbs = source_client.get_databases_summary()
|
|
target_dbs = target_client.get_databases_summary()
|
|
|
|
return suggest_mappings(source_dbs, target_dbs)
|
|
|
|
# endregion get_suggestions
|
|
|
|
|
|
# #endregion MappingService
|
|
|
|
# #endregion mapping_service
|