72 lines
3.0 KiB
Python
72 lines
3.0 KiB
Python
# [DEF:backend.src.services.mapping_service:Module]
|
|
#
|
|
# @SEMANTICS: service, mapping, fuzzy-matching, superset
|
|
# @PURPOSE: Orchestrates database fetching and fuzzy matching suggestions.
|
|
# @LAYER: Service
|
|
# @RELATION: DEPENDS_ON -> backend.src.core.superset_client
|
|
# @RELATION: DEPENDS_ON -> backend.src.core.utils.matching
|
|
#
|
|
# @INVARIANT: Suggestions are based on database names.
|
|
|
|
# [SECTION: IMPORTS]
|
|
from typing import List, Dict
|
|
from backend.src.core.logger import belief_scope
|
|
from backend.src.core.superset_client import SupersetClient
|
|
from backend.src.core.utils.matching import suggest_mappings
|
|
# [/SECTION]
|
|
|
|
# [DEF:MappingService:Class]
|
|
# @PURPOSE: Service for handling database mapping logic.
|
|
class MappingService:
|
|
|
|
# [DEF:__init__: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.
|
|
def __init__(self, config_manager):
|
|
with belief_scope("MappingService.__init__"):
|
|
self.config_manager = config_manager
|
|
# [/DEF:__init__:Function]
|
|
|
|
# [DEF:_get_client: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.
|
|
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)
|
|
# [/DEF:_get_client:Function]
|
|
|
|
# [DEF:get_suggestions: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.
|
|
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)
|
|
# [/DEF:get_suggestions:Function]
|
|
|
|
# [/DEF:MappingService:Class]
|
|
|
|
# [/DEF:backend.src.services.mapping_service:Module]
|