diff --git a/services/providers/adapters.py b/services/providers/adapters.py new file mode 100644 index 0000000..8fb13ec --- /dev/null +++ b/services/providers/adapters.py @@ -0,0 +1,314 @@ +""" +Track C1.1 — Provider interface adapters. + +Two adapter classes that bridge the legacy `TranslationProvider` +interface (translate(text, target, source) -> str) and the new +TranslationProvider interface (translate_text(request) -> Response). + +Why? + The codebase has TWO provider interfaces: + 1. New: services/providers/base.py + - translate_text(TranslationRequest) -> TranslationResponse + - translate_batch(List[TranslationRequest]) -> List[TranslationResponse] + 2. Legacy: services/translation_service.py + - translate(text, target, source) -> str + - translate_batch(texts, target, source) -> List[str] + + Both are used in production. Translators sometimes receive the new + interface, sometimes the legacy one. Previously the translators + duck-typed the provider to figure out which interface it supported + (see e.g. pptx_translator._translate_with_provider). That was hacky. + + These adapters give a single, explicit way to bridge the two + interfaces, so translators can always treat the provider as + whichever interface they prefer. + +Usage: + # Wrap a new-style provider as legacy + legacy = NewProviderAsLegacyAdapter(new_provider) + legacy.translate("Hello", "fr", "auto") # returns str + + # Wrap a legacy-style provider as new + new = LegacyProviderAsNewAdapter(legacy_provider) + new.translate_text(TranslationRequest(...)) # returns TranslationResponse +""" + +from __future__ import annotations + +from typing import List, Optional + +from core.logging import get_logger + +from .base import TranslationProvider as NewTranslationProvider +from .schemas import TranslationRequest, TranslationResponse, ProviderHealthStatus + +logger = get_logger(__name__) + + +# ============================================================================= +# New-style → Legacy +# ============================================================================= + +class NewProviderAsLegacyAdapter: + """Adapt a new-style provider (translate_text) to the legacy interface. + + The legacy interface is what most translators and the legacy + `translation_service` expect: + translate(text, target_language, source_language="auto") -> str + translate_batch(texts, target, source) -> List[str] + get_name() -> str + is_available() -> bool + + This adapter is for INTERNAL use — it adapts at the boundary so + that a single provider instance can be passed to any code path. + """ + + def __init__(self, new_provider: NewTranslationProvider): + self._new = new_provider + + # ---- Core legacy methods ---- + + def translate( + self, text: str, target_language: str, source_language: str = "auto" + ) -> str: + """Translate a single text — legacy signature.""" + # Lazy import to avoid circular import + try: + from .schemas import TranslationRequest + request = TranslationRequest( + text=text, + target_language=target_language, + source_language=source_language, + ) + response = self._new.translate_text(request) + if response.error: + raise Exception( + f"[{response.error_code or 'UNKNOWN'}] {response.error}" + ) + return response.translated_text + except Exception as e: + # If the new provider raised, re-raise so the caller + # can decide how to handle it (legacy code expects exceptions). + logger.debug( + "new_provider_as_legacy_translate_error", + error=str(e)[:200], + ) + raise + + def translate_batch( + self, + texts: List[str], + target_language: str, + source_language: str = "auto", + ) -> List[str]: + """Translate multiple texts — legacy signature.""" + if not texts: + return [] + + try: + from .schemas import TranslationRequest + requests = [ + TranslationRequest( + text=t, + target_language=target_language, + source_language=source_language, + ) + for t in texts + ] + responses = self._new.translate_batch(requests) + return [r.translated_text for r in responses] + except Exception as e: + logger.debug( + "new_provider_as_legacy_batch_error", + error=str(e)[:200], + count=len(texts), + ) + raise + + # ---- Identity methods (pass-through) ---- + + def get_name(self) -> str: + return self._new.get_name() + + def is_available(self) -> bool: + try: + return bool(self._new.is_available()) + except Exception: + return False + + def health_check(self) -> dict: + """Best-effort health check, legacy-shaped (plain dict).""" + try: + status = self._new.health_check() + return { + "name": status.name, + "available": status.available, + "latency_ms": status.latency_ms, + "error": status.error, + } + except Exception as e: + return { + "name": self.get_name(), + "available": False, + "error": str(e)[:200], + } + + +# ============================================================================= +# Legacy-style → New +# ============================================================================= + +class LegacyProviderAsNewAdapter(NewTranslationProvider): + """Adapt a legacy-style provider (translate(text, ...)) to the new interface. + + Most new code expects the new interface: + translate_text(TranslationRequest) -> TranslationResponse + translate_batch(List[TranslationRequest]) -> List[TranslationResponse] + + A "legacy" provider is anything that has: + translate(text, target, source) -> str + translate_batch(texts, target, source) -> List[str] + get_name() -> str + is_available() -> bool + """ + + def __init__(self, legacy_provider, provider_name: str = None): + self._legacy = legacy_provider + # If a custom name is provided, use it; otherwise inherit from the + # legacy provider. + if provider_name is not None: + self._name = provider_name + else: + try: + self._name = legacy_provider.get_name() + except Exception: + self._name = "legacy" + + # ---- Implement the abstract methods of NewTranslationProvider ---- + + def translate_text(self, request: TranslationRequest) -> TranslationResponse: + """Translate a single text — new interface.""" + start = time.time() + try: + translated = self._legacy.translate( + text=request.text, + target_language=request.target_language, + source_language=request.source_language or "auto", + ) + elapsed_ms = round((time.time() - start) * 1000, 2) + return TranslationResponse( + translated_text=translated or "", + provider_name=self._name, + source_language=request.source_language or "auto", + target_language=request.target_language, + elapsed_ms=elapsed_ms, + ) + except Exception as e: + elapsed_ms = round((time.time() - start) * 1000, 2) + return TranslationResponse( + translated_text="", + provider_name=self._name, + source_language=request.source_language or "auto", + target_language=request.target_language, + elapsed_ms=elapsed_ms, + error=str(e)[:200], + error_code="LEGACY_PROVIDER_ERROR", + ) + + def translate_batch( + self, requests: List[TranslationRequest] + ) -> List[TranslationResponse]: + """Translate a batch of texts — new interface.""" + if not requests: + return [] + + # Group by language pair so we can call the legacy + # translate_batch efficiently (legacy signature is per-language-pair). + from collections import defaultdict + groups: dict = defaultdict(list) + for i, r in enumerate(requests): + key = (r.target_language, r.source_language or "auto") + groups[key].append((i, r)) + + results: List[TranslationResponse] = [None] * len(requests) # type: ignore + + for (target_lang, source_lang), items in groups.items(): + texts = [r.text for _, r in items] + try: + translated = self._legacy.translate_batch( + texts=texts, + target_language=target_lang, + source_language=source_lang, + ) + except Exception as e: + # If legacy batch fails, fall back to per-text + logger.debug( + "legacy_batch_fallback_to_individual", + error=str(e)[:200], + ) + translated = [] + for t in texts: + try: + translated.append( + self._legacy.translate(t, target_lang, source_lang) + ) + except Exception as individual_err: + translated.append("") # legacy code expects str + + for (orig_idx, r), t in zip(items, translated): + results[orig_idx] = TranslationResponse( + translated_text=t or "", + provider_name=self._name, + source_language=source_lang, + target_language=target_lang, + ) + + return results + + def get_name(self) -> str: + # Always prefer the explicit name set at construction. Only fall + # back to the legacy provider's get_name if no name was set. + if hasattr(self, "_name") and self._name: + return self._name + try: + return self._legacy.get_name() + except Exception: + return "legacy" + + def is_available(self) -> bool: + try: + return bool(self._legacy.is_available()) + except Exception: + return False + + +# ============================================================================= +# Auto-detect helper +# ============================================================================= + +import time # noqa: E402 + + +def coerce_to_legacy(provider) -> NewProviderAsLegacyAdapter: + """Coerce any provider to the legacy interface. + + If the provider is already a legacy provider, return it as-is. + If it's a new-style provider, wrap it in NewProviderAsLegacyAdapter. + """ + if isinstance(provider, NewProviderAsLegacyAdapter): + return provider + if isinstance(provider, NewTranslationProvider): + return NewProviderAsLegacyAdapter(provider) + # Already legacy-shaped — pass through + return provider + + +def coerce_to_new(provider, name: str = None) -> NewTranslationProvider: + """Coerce any provider to the new interface. + + If the provider is already a new-style provider, return it as-is. + If it's a legacy provider, wrap it in LegacyProviderAsNewAdapter. + """ + if isinstance(provider, NewTranslationProvider): + return provider + return LegacyProviderAsNewAdapter(provider, provider_name=name) diff --git a/tests/test_providers_adapter.py b/tests/test_providers_adapter.py new file mode 100644 index 0000000..d079ad6 --- /dev/null +++ b/tests/test_providers_adapter.py @@ -0,0 +1,332 @@ +""" +Tests for Track C1.1 — NewProviderAsLegacyAdapter and LegacyProviderAsNewAdapter. + +Covers: + - NewProviderAsLegacyAdapter: + * translate() delegates to translate_text() + * translate_batch() delegates to translate_batch() + * get_name() and is_available() pass through + * errors propagate as exceptions (legacy code expects this) + - LegacyProviderAsNewAdapter: + * translate_text() wraps legacy.translate() + * translate_batch() groups by language pair + * get_name() and is_available() pass through + - coerce_to_legacy / coerce_to_new: + * no-op when already in target shape + * wraps when in the other shape +""" +import pytest +from typing import List +from unittest.mock import MagicMock + +from services.providers.base import TranslationProvider as NewTranslationProvider +from services.providers.adapters import ( + NewProviderAsLegacyAdapter, + LegacyProviderAsNewAdapter, + coerce_to_legacy, + coerce_to_new, +) +from services.providers.schemas import ( + TranslationRequest, + TranslationResponse, +) + + +# ============================================================================ +# Test helpers — new-style mock +# ============================================================================ + +class MockNewProvider(NewTranslationProvider): + """Minimal new-style provider for testing the adapter.""" + + def __init__(self, translations: dict = None, name: str = "mock-new"): + self._t = translations or {} + self._name = name + self.translate_text_calls = [] + self.translate_batch_calls = [] + + def translate_text(self, request: TranslationRequest) -> TranslationResponse: + self.translate_text_calls.append(request) + translated = self._t.get(request.text, f"NEW_{request.text}") + return TranslationResponse( + translated_text=translated, + provider_name=self._name, + source_language=request.source_language or "auto", + target_language=request.target_language, + ) + + def translate_batch( + self, requests: List[TranslationRequest] + ) -> List[TranslationResponse]: + """Override the default (which calls translate_text N times) so we + can verify the adapter calls translate_batch directly.""" + self.translate_batch_calls.append(requests) + return [self.translate_text(req) for req in requests] + + def get_name(self) -> str: + return self._name + + def is_available(self) -> bool: + return True + + +# ============================================================================ +# Test helpers — legacy mock +# ============================================================================ + +class MockLegacyProvider: + """Minimal legacy provider for testing the adapter.""" + + def __init__(self, translations: dict = None, name: str = "mock-legacy"): + self._t = translations or {} + self._name = name + self.calls = [] + + def translate(self, text, target_language, source_language="auto"): + self.calls.append((text, target_language, source_language)) + return self._t.get(text, f"LEG_{text}") + + def translate_batch(self, texts, target_language, source_language="auto"): + return [self.translate(t, target_language, source_language) for t in texts] + + def get_name(self): + return self._name + + def is_available(self): + return True + + +# ============================================================================ +# NewProviderAsLegacyAdapter +# ============================================================================ + +class TestNewProviderAsLegacyAdapter: + def test_translate_delegates(self): + new = MockNewProvider({"Hello": "Bonjour"}) + legacy = NewProviderAsLegacyAdapter(new) + + result = legacy.translate("Hello", "fr", "en") + assert result == "Bonjour" + assert len(new.translate_text_calls) == 1 + assert new.translate_text_calls[0].text == "Hello" + assert new.translate_text_calls[0].target_language == "fr" + + def test_translate_default_source(self): + new = MockNewProvider() + legacy = NewProviderAsLegacyAdapter(new) + legacy.translate("Hello", "fr") + assert new.translate_text_calls[0].source_language == "auto" + + def test_translate_batch_delegates(self): + new = MockNewProvider({"A": "TR_A", "B": "TR_B"}) + legacy = NewProviderAsLegacyAdapter(new) + result = legacy.translate_batch(["A", "B"], "fr") + assert result == ["TR_A", "TR_B"] + assert len(new.translate_batch_calls) == 1 + + def test_translate_batch_empty(self): + new = MockNewProvider() + legacy = NewProviderAsLegacyAdapter(new) + result = legacy.translate_batch([], "fr") + assert result == [] + + def test_get_name_passthrough(self): + new = MockNewProvider(name="my-new") + legacy = NewProviderAsLegacyAdapter(new) + assert legacy.get_name() == "my-new" + + def test_is_available_passthrough(self): + new = MockNewProvider() + legacy = NewProviderAsLegacyAdapter(new) + assert legacy.is_available() is True + + def test_translate_error_raises(self): + """Legacy code expects exceptions, not Response objects with errors.""" + + class FailingProvider(MockNewProvider): + def translate_text(self, request): + return TranslationResponse( + translated_text="", + provider_name="failing", + source_language="auto", + target_language="fr", + error="upstream timeout", + error_code="TIMEOUT", + ) + + legacy = NewProviderAsLegacyAdapter(FailingProvider()) + with pytest.raises(Exception, match="TIMEOUT"): + legacy.translate("Hello", "fr") + + def test_health_check(self): + new = MockNewProvider() + legacy = NewProviderAsLegacyAdapter(new) + status = legacy.health_check() + assert status["name"] == "mock-new" + assert status["available"] is True + + +# ============================================================================ +# LegacyProviderAsNewAdapter +# ============================================================================ + +class TestLegacyProviderAsNewAdapter: + def test_translate_text_wraps(self): + legacy = MockLegacyProvider({"Hello": "Bonjour"}) + adapter = LegacyProviderAsNewAdapter(legacy) + + result = adapter.translate_text(TranslationRequest( + text="Hello", target_language="fr", source_language="en", + )) + assert result.translated_text == "Bonjour" + assert result.provider_name == "mock-legacy" + assert not result.error + assert len(legacy.calls) == 1 + + def test_translate_text_error_returns_response(self): + """New interface wraps errors in the Response, doesn't raise.""" + class FailingLegacy: + def translate(self, *args, **kwargs): + raise RuntimeError("kaboom") + def get_name(self): return "failing" + def is_available(self): return True + + adapter = LegacyProviderAsNewAdapter(FailingLegacy()) + result = adapter.translate_text(TranslationRequest( + text="Hello", target_language="fr", source_language="en", + )) + assert result.translated_text == "" + assert "kaboom" in result.error + assert result.error_code == "LEGACY_PROVIDER_ERROR" + + def test_translate_batch_single_group(self): + legacy = MockLegacyProvider({"A": "TR_A", "B": "TR_B"}) + adapter = LegacyProviderAsNewAdapter(legacy) + + results = adapter.translate_batch([ + TranslationRequest(text="A", target_language="fr", source_language="en"), + TranslationRequest(text="B", target_language="fr", source_language="en"), + ]) + assert [r.translated_text for r in results] == ["TR_A", "TR_B"] + + def test_translate_batch_groups_by_language_pair(self): + """Requests with different (target, source) should NOT be batched together.""" + legacy = MockLegacyProvider() + adapter = LegacyProviderAsNewAdapter(legacy) + + adapter.translate_batch([ + TranslationRequest(text="A", target_language="fr", source_language="en"), + TranslationRequest(text="B", target_language="es", source_language="en"), + TranslationRequest(text="C", target_language="fr", source_language="en"), + ]) + + # Three calls: A→fr, B→es, C→fr (legacy translate_batch is one + # call per language pair) + # Two distinct (target, source) pairs → at most 2 batch calls + assert len(legacy.calls) >= 2 + + def test_translate_batch_empty(self): + legacy = MockLegacyProvider() + adapter = LegacyProviderAsNewAdapter(legacy) + results = adapter.translate_batch([]) + assert results == [] + + def test_translate_batch_order_preserved(self): + """The order of results must match the order of requests.""" + legacy = MockLegacyProvider({"X": "TR_X", "Y": "TR_Y", "Z": "TR_Z"}) + adapter = LegacyProviderAsNewAdapter(legacy) + + results = adapter.translate_batch([ + TranslationRequest(text="Z", target_language="fr", source_language="en"), + TranslationRequest(text="X", target_language="fr", source_language="en"), + TranslationRequest(text="Y", target_language="fr", source_language="en"), + ]) + assert [r.translated_text for r in results] == ["TR_Z", "TR_X", "TR_Y"] + + def test_get_name_passthrough(self): + legacy = MockLegacyProvider(name="my-legacy") + adapter = LegacyProviderAsNewAdapter(legacy) + assert adapter.get_name() == "my-legacy" + + def test_is_available_passthrough(self): + legacy = MockLegacyProvider() + legacy.is_available = lambda: False + adapter = LegacyProviderAsNewAdapter(legacy) + assert adapter.is_available() is False + + def test_custom_provider_name_overrides(self): + legacy = MockLegacyProvider(name="original-name") + adapter = LegacyProviderAsNewAdapter(legacy, provider_name="override-name") + assert adapter.get_name() == "override-name" + + +# ============================================================================ +# Coercion helpers +# ============================================================================ + +class TestCoercion: + def test_new_already_legacy_noop(self): + new = MockNewProvider() + adapter = NewProviderAsLegacyAdapter(new) + result = coerce_to_legacy(adapter) + # Should return the existing adapter, not wrap it again + assert result is adapter + + def test_new_provider_to_legacy_wraps(self): + new = MockNewProvider() + result = coerce_to_legacy(new) + assert isinstance(result, NewProviderAsLegacyAdapter) + assert result._new is new + + def test_legacy_provider_unchanged(self): + legacy = MockLegacyProvider() + result = coerce_to_legacy(legacy) + assert result is legacy + + def test_legacy_already_new_noop(self): + legacy = MockLegacyProvider() + adapter = LegacyProviderAsNewAdapter(legacy) + result = coerce_to_new(adapter) + assert result is adapter + + def test_legacy_provider_to_new_wraps(self): + legacy = MockLegacyProvider() + result = coerce_to_new(legacy) + assert isinstance(result, NewTranslationProvider) + assert isinstance(result, LegacyProviderAsNewAdapter) + + def test_new_provider_unchanged(self): + new = MockNewProvider() + result = coerce_to_new(new) + assert result is new + + +# ============================================================================ +# Round-trip +# ============================================================================ + +class TestRoundTrip: + """An adapter should be reversible: a string that comes out of one + side should match what the other side would have returned.""" + + def test_new_legacy_round_trip(self): + new = MockNewProvider({"Hello": "Bonjour", "World": "Monde"}) + legacy = NewProviderAsLegacyAdapter(new) + + # Round-trip: legacy.translate → new.translate_text + via_legacy = legacy.translate("Hello", "fr", "en") + via_new = new.translate_text(TranslationRequest( + text="Hello", target_language="fr", source_language="en", + )).translated_text + assert via_legacy == via_new + + def test_legacy_new_round_trip(self): + legacy = MockLegacyProvider({"Hello": "Bonjour"}) + adapter = LegacyProviderAsNewAdapter(legacy) + + # Round-trip: new.translate_text → legacy.translate + via_new = adapter.translate_text(TranslationRequest( + text="Hello", target_language="fr", source_language="en", + )).translated_text + via_legacy = legacy.translate("Hello", "fr", "en") + assert via_new == via_legacy