149 lines
4.2 KiB
Python
149 lines
4.2 KiB
Python
"""
|
|
Provider Registry - Singleton pattern for managing translation providers.
|
|
|
|
Provides a central registry for all translation providers with:
|
|
- Registration and retrieval by name
|
|
- Listing available providers
|
|
- Fallback chain support
|
|
"""
|
|
|
|
from typing import Dict, List, Optional
|
|
import threading
|
|
|
|
from .base import TranslationProvider
|
|
|
|
|
|
class ProviderRegistry:
|
|
"""
|
|
Singleton registry for translation providers.
|
|
|
|
Thread-safe implementation for managing multiple translation providers
|
|
with support for fallback chains.
|
|
"""
|
|
|
|
_instance: Optional["ProviderRegistry"] = None
|
|
_lock: threading.Lock = threading.Lock()
|
|
|
|
def __new__(cls) -> "ProviderRegistry":
|
|
"""Create or return the singleton instance."""
|
|
if cls._instance is None:
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
cls._instance._providers: Dict[str, TranslationProvider] = {}
|
|
cls._instance._providers_lock = threading.RLock()
|
|
return cls._instance
|
|
|
|
def register(self, name: str, provider: TranslationProvider) -> None:
|
|
"""
|
|
Register a translation provider.
|
|
|
|
Args:
|
|
name: Unique name for the provider (e.g., "google", "deepl")
|
|
provider: TranslationProvider instance
|
|
"""
|
|
with self._providers_lock:
|
|
self._providers[name] = provider
|
|
|
|
def unregister(self, name: str) -> bool:
|
|
"""
|
|
Unregister a translation provider.
|
|
|
|
Args:
|
|
name: Name of the provider to remove
|
|
|
|
Returns:
|
|
True if provider was removed, False if not found
|
|
"""
|
|
with self._providers_lock:
|
|
if name in self._providers:
|
|
del self._providers[name]
|
|
return True
|
|
return False
|
|
|
|
def get(self, name: str) -> Optional[TranslationProvider]:
|
|
"""
|
|
Get a registered provider by name.
|
|
|
|
Args:
|
|
name: Provider name
|
|
|
|
Returns:
|
|
TranslationProvider instance or None if not found
|
|
"""
|
|
with self._providers_lock:
|
|
return self._providers.get(name)
|
|
|
|
def list_all(self) -> List[str]:
|
|
"""
|
|
List all registered provider names.
|
|
|
|
Returns:
|
|
List of provider names
|
|
"""
|
|
with self._providers_lock:
|
|
return list(self._providers.keys())
|
|
|
|
def list_available(self) -> List[str]:
|
|
"""
|
|
List names of all available (reachable) providers.
|
|
|
|
Returns:
|
|
List of provider names that are currently available
|
|
"""
|
|
with self._providers_lock:
|
|
return [
|
|
name
|
|
for name, provider in self._providers.items()
|
|
if provider.is_available()
|
|
]
|
|
|
|
def get_first_available(self, names: List[str]) -> Optional[TranslationProvider]:
|
|
"""
|
|
Get the first available provider from a list of names (fallback chain).
|
|
|
|
Iterates through the list in order and returns the first provider
|
|
that is available. This enables graceful degradation when providers
|
|
are unavailable.
|
|
|
|
Args:
|
|
names: List of provider names in priority order
|
|
|
|
Returns:
|
|
First available TranslationProvider or None if all are unavailable
|
|
"""
|
|
for name in names:
|
|
provider = self.get(name)
|
|
if provider is not None and provider.is_available():
|
|
return provider
|
|
return None
|
|
|
|
def clear(self) -> None:
|
|
"""Remove all registered providers."""
|
|
with self._providers_lock:
|
|
self._providers.clear()
|
|
|
|
def __len__(self) -> int:
|
|
"""Return the number of registered providers."""
|
|
with self._providers_lock:
|
|
return len(self._providers)
|
|
|
|
def __contains__(self, name: str) -> bool:
|
|
"""Check if a provider is registered."""
|
|
with self._providers_lock:
|
|
return name in self._providers
|
|
|
|
|
|
def get_registry() -> ProviderRegistry:
|
|
"""
|
|
Get the global provider registry instance.
|
|
|
|
Returns:
|
|
The singleton ProviderRegistry instance
|
|
"""
|
|
return ProviderRegistry()
|
|
|
|
|
|
# Global registry instance
|
|
registry = ProviderRegistry()
|