sepehr
5fd087979b
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m21s
123 lines
3.5 KiB
Python
123 lines
3.5 KiB
Python
"""
|
|
Abstract base class for translation providers.
|
|
Provides a common interface for all translation provider implementations.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Optional, List
|
|
import time
|
|
|
|
from .schemas import (
|
|
TranslationRequest,
|
|
TranslationResponse,
|
|
BatchTranslationRequest,
|
|
BatchTranslationResponse,
|
|
ProviderHealthStatus,
|
|
)
|
|
|
|
|
|
class TranslationProvider(ABC):
|
|
"""
|
|
Abstract base class for translation providers.
|
|
|
|
All translation providers must implement this interface to ensure
|
|
consistent behavior across different translation services.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def translate_text(self, request: TranslationRequest) -> TranslationResponse:
|
|
"""
|
|
Translate a single text string.
|
|
|
|
Args:
|
|
request: TranslationRequest containing text, target_language, and source_language
|
|
|
|
Returns:
|
|
TranslationResponse with translated text and metadata
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_name(self) -> str:
|
|
"""
|
|
Return the provider name for logging and registry.
|
|
|
|
Returns:
|
|
Provider name as a string (e.g., "google", "deepl", "openai")
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def is_available(self) -> bool:
|
|
"""
|
|
Check if the provider is configured and reachable.
|
|
|
|
Returns:
|
|
True if the provider can perform translations, False otherwise
|
|
"""
|
|
pass
|
|
|
|
def translate_batch(
|
|
self, requests: List[TranslationRequest]
|
|
) -> List[TranslationResponse]:
|
|
"""
|
|
Translate multiple texts. Default implementation uses individual calls.
|
|
|
|
Subclasses can override this for optimized batch processing.
|
|
|
|
Args:
|
|
requests: List of TranslationRequest objects
|
|
|
|
Returns:
|
|
List of TranslationResponse objects in the same order as requests
|
|
"""
|
|
return [self.translate_text(req) for req in requests]
|
|
|
|
def translate(
|
|
self, text: str, target_language: str, source_language: str = "auto"
|
|
) -> str:
|
|
"""
|
|
Compatibility method for the legacy interface.
|
|
Translates a single text string synchronously.
|
|
"""
|
|
req = TranslationRequest(
|
|
text=text,
|
|
target_language=target_language,
|
|
source_language=source_language,
|
|
)
|
|
resp = self.translate_text(req)
|
|
if resp.error:
|
|
raise Exception(f"[{resp.error_code or 'UNKNOWN'}] {resp.error}")
|
|
return resp.translated_text
|
|
|
|
|
|
def health_check(self) -> ProviderHealthStatus:
|
|
"""
|
|
Return health status details for the provider.
|
|
|
|
Performs a lightweight check to verify the provider is operational.
|
|
|
|
Returns:
|
|
ProviderHealthStatus with availability and latency information
|
|
"""
|
|
start_time = time.time()
|
|
|
|
try:
|
|
available = self.is_available()
|
|
latency_ms = (time.time() - start_time) * 1000
|
|
|
|
return ProviderHealthStatus(
|
|
name=self.get_name(),
|
|
available=available,
|
|
latency_ms=round(latency_ms, 2),
|
|
error=None if available else "Provider not available",
|
|
)
|
|
except Exception as e:
|
|
latency_ms = (time.time() - start_time) * 1000
|
|
return ProviderHealthStatus(
|
|
name=self.get_name(),
|
|
available=False,
|
|
latency_ms=round(latency_ms, 2),
|
|
error=str(e),
|
|
)
|