""" Unified service configuration accessors. This module is now a thin adapter over ``config.loader.get_app_config`` and contains no independent parsing or precedence logic. """ from __future__ import annotations import os from typing import Any, Dict, Tuple from config.loader import get_app_config from config.schema import ( EmbeddingServiceConfig, RerankServiceConfig, RerankServiceInstanceConfig, TranslationServiceConfig, ) def get_translation_config() -> Dict[str, Any]: return get_app_config().services.translation.as_dict() def get_embedding_config() -> EmbeddingServiceConfig: return get_app_config().services.embedding def get_rerank_config() -> RerankServiceConfig: return get_app_config().services.rerank def get_rerank_instance_config(profile: str | None = None) -> RerankServiceInstanceConfig: cfg = get_app_config().services.rerank instance_name = str( profile or os.getenv("RERANK_INSTANCE") or cfg.default_instance ).strip() or cfg.default_instance return cfg.get_instance(instance_name) def get_translation_base_url() -> str: return get_app_config().services.translation.endpoint def get_translation_cache_config() -> Dict[str, Any]: return dict(get_app_config().services.translation.cache) def get_embedding_text_base_url() -> str: provider_cfg = get_app_config().services.embedding.get_provider_config() base = provider_cfg.get("text_base_url") if not base: raise ValueError("Embedding text base_url is not configured") return str(base).rstrip("/") def get_embedding_image_base_url() -> str: provider_cfg = get_app_config().services.embedding.get_provider_config() base = provider_cfg.get("image_base_url") if not base: raise ValueError("Embedding image base_url is not configured") return str(base).rstrip("/") def get_embedding_backend_config() -> Tuple[str, Dict[str, Any]]: cfg = get_app_config().services.embedding return cfg.backend, cfg.get_backend_config() def get_embedding_image_backend_config() -> Tuple[str, Dict[str, Any]]: cfg = get_app_config().services.embedding return cfg.image_backend, cfg.get_image_backend_config() def get_rerank_backend_config(profile: str | None = None) -> Tuple[str, Dict[str, Any]]: cfg = get_app_config().services.rerank instance = get_rerank_instance_config(profile) backend = str(os.getenv("RERANK_BACKEND") or instance.backend).strip() if backend != instance.backend: backend_cfg = cfg.backends.get(backend) if backend_cfg is None: raise ValueError(f"Unknown rerank backend override from RERANK_BACKEND: {backend!r}") return backend, dict(backend_cfg) return instance.backend, cfg.get_backend_config(profile) def get_rerank_base_url(profile: str | None = None) -> str: cfg = get_app_config().services.rerank instance = get_rerank_instance_config(profile) provider_cfg = cfg.get_provider_config() profile_name = str(profile or os.getenv("RERANK_INSTANCE") or cfg.default_instance).strip() or cfg.default_instance base = None provider_instances = provider_cfg.get("instances") if isinstance(provider_instances, dict): instance_provider_cfg = provider_instances.get(profile_name) if isinstance(instance_provider_cfg, dict): base = instance_provider_cfg.get("base_url") or instance_provider_cfg.get("service_url") if not base: base = instance.base_url or instance.service_url if not base: base = provider_cfg.get("service_url") or provider_cfg.get("base_url") if not base: raise ValueError("Rerank service URL is not configured") base = str(base).rstrip("/") if base.endswith("/rerank"): base = base[: -len("/rerank")] return base def get_rerank_service_url(profile: str | None = None) -> str: cfg = get_app_config().services.rerank instance = get_rerank_instance_config(profile) provider_cfg = cfg.get_provider_config() profile_name = str(profile or os.getenv("RERANK_INSTANCE") or cfg.default_instance).strip() or cfg.default_instance service_url = None provider_instances = provider_cfg.get("instances") if isinstance(provider_instances, dict): instance_provider_cfg = provider_instances.get(profile_name) if isinstance(instance_provider_cfg, dict): service_url = instance_provider_cfg.get("service_url") if not service_url: service_url = instance.service_url if not service_url: service_url = f"{get_rerank_base_url(profile=profile)}/rerank" return str(service_url).rstrip("/")