|
2 | 2 |
|
3 | 3 | from typing import Any |
4 | 4 |
|
5 | | -from orb.application.ports.scheduler_registry_port import SchedulerRegistryPort |
6 | 5 | from orb.domain.base.ports.logging_port import LoggingPort |
7 | 6 |
|
8 | 7 |
|
9 | 8 | class SchedulerRegistryService: |
10 | 9 | """Application service interface for scheduler registry operations.""" |
11 | 10 |
|
12 | | - def __init__(self, registry: SchedulerRegistryPort, logger: LoggingPort): |
| 11 | + def __init__(self, registry: Any, logger: LoggingPort): |
13 | 12 | self._registry = registry |
14 | 13 | self._logger = logger |
15 | 14 |
|
16 | 15 | def get_available_schedulers(self) -> list[str]: |
17 | 16 | """Get list of available scheduler types.""" |
18 | | - return self._registry.get_registered_types() # type: ignore[attr-defined] |
| 17 | + return self._registry.get_registered_types() |
19 | 18 |
|
20 | 19 | def create_scheduler_strategy(self, scheduler_type: str, config: Any) -> Any: |
21 | 20 | """Create scheduler strategy instance.""" |
22 | | - return self._registry.create_strategy(scheduler_type, config) # type: ignore[attr-defined] |
| 21 | + return self._registry.create_strategy(scheduler_type, config) |
23 | 22 |
|
24 | 23 | def is_scheduler_registered(self, scheduler_type: str) -> bool: |
25 | 24 | """Check if scheduler type is registered.""" |
26 | | - return self._registry.is_registered(scheduler_type) # type: ignore[attr-defined] |
| 25 | + return self._registry.is_registered(scheduler_type) |
27 | 26 |
|
28 | 27 | def get_scheduler_capabilities(self, scheduler_type: str) -> dict[str, Any]: |
29 | 28 | """Get scheduler capabilities (if supported).""" |
30 | 29 | try: |
31 | | - strategy = self._registry.create_strategy(scheduler_type, {}) # type: ignore[attr-defined] |
| 30 | + strategy = self._registry.create_strategy(scheduler_type, {}) |
32 | 31 | return getattr(strategy, "get_capabilities", lambda: {})() |
33 | 32 | except Exception: |
34 | 33 | return {} |
0 commit comments