"""Service discovery.""" import json import os from typing import Dict, Tuple from .constants import PROD_ENVIRONMENT, QA_ENVIRONMENT, UAT_ENVIRONMENT SERVICE_URL_SPEC = "{protocol}://{environment}-{service_name}.{domain}" SERVICE_DEFAULT_DOMAIN = "theorchard.io" def get_owsclient_service_map() -> Dict[str, str]: """Get the service mapping from environment.""" service_map = os.environ.get("OWSREQUEST_SERVICE_MAP") if service_map: try: mapping = json.loads(service_map) except (json.JSONDecodeError, TypeError): pass else: if isinstance(mapping, dict): return mapping return {} owsclient_service_map = get_owsclient_service_map() def discover_service_url(environment: str, service_name: str) -> Tuple[str, str]: """Discover the service url for a given service name and environment.""" service_name = service_name.lower() environment = environment.lower() if environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT]: url = SERVICE_URL_SPEC.format( environment=environment, service_name=service_name, domain=SERVICE_DEFAULT_DOMAIN, protocol="https", ) return environment, url if environment == UAT_ENVIRONMENT: if service_name in owsclient_service_map: return environment, owsclient_service_map[service_name] return environment, SERVICE_URL_SPEC.format( environment=environment, service_name=service_name, domain=SERVICE_DEFAULT_DOMAIN, protocol="https", ) if environment in ["test", "dev"] and service_name in owsclient_service_map: try: return environment, owsclient_service_map[service_name] except KeyError: pass # Default to qa url = SERVICE_URL_SPEC.format( environment=QA_ENVIRONMENT, service_name=service_name, domain=SERVICE_DEFAULT_DOMAIN, protocol="https", ) return QA_ENVIRONMENT, url