mirror of https://github.com/kortix-ai/suna.git
44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
from typing import Dict, List, Optional
|
|
from .core import TriggerProvider, TriggerType
|
|
from .providers import TelegramTriggerProvider
|
|
|
|
class TriggerRegistry:
|
|
"""Registry for trigger providers."""
|
|
|
|
def __init__(self):
|
|
self._providers: Dict[TriggerType, TriggerProvider] = {}
|
|
self._initialize_default_providers()
|
|
|
|
def _initialize_default_providers(self):
|
|
"""Initialize default trigger providers."""
|
|
self.register_provider(TelegramTriggerProvider())
|
|
|
|
def register_provider(self, provider: TriggerProvider):
|
|
"""Register a trigger provider."""
|
|
self._providers[provider.trigger_type] = provider
|
|
|
|
def get_provider(self, trigger_type: TriggerType) -> Optional[TriggerProvider]:
|
|
"""Get a trigger provider by type."""
|
|
return self._providers.get(trigger_type)
|
|
|
|
def get_all_providers(self) -> Dict[TriggerType, TriggerProvider]:
|
|
"""Get all registered providers."""
|
|
return self._providers.copy()
|
|
|
|
def get_supported_types(self) -> List[TriggerType]:
|
|
"""Get list of supported trigger types."""
|
|
return list(self._providers.keys())
|
|
|
|
def get_provider_schemas(self) -> Dict[str, Dict]:
|
|
"""Get configuration schemas for all providers."""
|
|
schemas = {}
|
|
for trigger_type, provider in self._providers.items():
|
|
trigger_type_str = trigger_type.value if hasattr(trigger_type, 'value') else str(trigger_type)
|
|
schemas[trigger_type_str] = provider.get_config_schema()
|
|
return schemas
|
|
|
|
def is_supported(self, trigger_type: TriggerType) -> bool:
|
|
"""Check if a trigger type is supported."""
|
|
return trigger_type in self._providers
|
|
|
|
trigger_registry = TriggerRegistry() |