Spaces:
Sleeping
Sleeping
| """ | |
| SkyOS Sync Manager. | |
| Central orchestration layer for the synchronization subsystem. | |
| The SyncManager coordinates: | |
| - adapter registration and lookup | |
| - scheduled synchronization | |
| - structured logging | |
| - conflict resolution | |
| Actual data transport remains the responsibility of adapters. | |
| """ | |
| from __future__ import annotations | |
| from typing import Any, Dict, Iterable, List, Optional | |
| from .conflict_resolver import ConflictResolver | |
| from .logger import SyncLogger | |
| from .registry import AdapterRegistry | |
| from .scheduler import SyncScheduler | |
| class SyncManager: | |
| """ | |
| Central coordinator for the SkyOS synchronization layer. | |
| The manager does not directly implement provider-specific synchronization. | |
| Instead, it coordinates registered adapters and the surrounding sync | |
| infrastructure. | |
| """ | |
| def __init__( | |
| self, | |
| registry: Optional[AdapterRegistry] = None, | |
| scheduler: Optional[SyncScheduler] = None, | |
| logger: Optional[SyncLogger] = None, | |
| conflict_resolver: Optional[ConflictResolver] = None, | |
| ) -> None: | |
| self.registry = registry or AdapterRegistry() | |
| self.scheduler = scheduler or SyncScheduler() | |
| self.logger = logger or SyncLogger() | |
| self.conflict_resolver = ( | |
| conflict_resolver or ConflictResolver() | |
| ) | |
| self._initialized = False | |
| def initialize(self) -> None: | |
| """ | |
| Initialize the synchronization manager. | |
| Initialization is intentionally idempotent so the manager can safely | |
| be initialized by application startup code more than once. | |
| """ | |
| if self._initialized: | |
| return | |
| self._initialized = True | |
| self.logger.info( | |
| "SyncManager initialized", | |
| component="sync_manager", | |
| ) | |
| def register_adapter( | |
| self, | |
| name: str, | |
| adapter: Any, | |
| ) -> Any: | |
| """ | |
| Register an adapter with the central registry. | |
| Args: | |
| name: Logical adapter name. | |
| adapter: Adapter instance. | |
| Returns: | |
| The registered adapter. | |
| """ | |
| registered_adapter = self.registry.register(name, adapter) | |
| self.logger.info( | |
| "Sync adapter registered", | |
| adapter=name, | |
| component="sync_manager", | |
| ) | |
| return registered_adapter | |
| def get_adapter(self, name: str) -> Any: | |
| """ | |
| Retrieve a registered adapter. | |
| Args: | |
| name: Logical adapter name. | |
| Returns: | |
| Registered adapter instance. | |
| """ | |
| return self.registry.get(name) | |
| def list_adapters(self) -> List[str]: | |
| """ | |
| Return the names of all registered adapters. | |
| """ | |
| return list(self.registry.list()) | |
| def resolve_conflict( | |
| self, | |
| local_data: Any, | |
| remote_data: Any, | |
| context: Optional[Dict[str, Any]] = None, | |
| ) -> Any: | |
| """ | |
| Delegate conflict resolution to the ConflictResolver. | |
| """ | |
| return self.conflict_resolver.resolve( | |
| local_data=local_data, | |
| remote_data=remote_data, | |
| context=context or {}, | |
| ) | |
| def schedule( | |
| self, | |
| name: str, | |
| callback: Any, | |
| *args: Any, | |
| **kwargs: Any, | |
| ) -> Any: | |
| """ | |
| Register a synchronization task with the scheduler. | |
| """ | |
| self.logger.info( | |
| "Synchronization task scheduled", | |
| task=name, | |
| component="sync_manager", | |
| ) | |
| return self.scheduler.schedule( | |
| name, | |
| callback, | |
| *args, | |
| **kwargs, | |
| ) | |
| def run_adapter( | |
| self, | |
| adapter_name: str, | |
| *args: Any, | |
| **kwargs: Any, | |
| ) -> Any: | |
| """ | |
| Execute a registered adapter. | |
| The SyncManager intentionally performs only orchestration here. | |
| Provider-specific behavior remains inside the adapter. | |
| """ | |
| adapter = self.get_adapter(adapter_name) | |
| self.logger.info( | |
| "Executing sync adapter", | |
| adapter=adapter_name, | |
| component="sync_manager", | |
| ) | |
| if hasattr(adapter, "sync"): | |
| return adapter.sync(*args, **kwargs) | |
| if hasattr(adapter, "run"): | |
| return adapter.run(*args, **kwargs) | |
| raise AttributeError( | |
| f"Adapter '{adapter_name}' does not provide " | |
| "'sync' or 'run'." | |
| ) | |
| def sync( | |
| self, | |
| source: str, | |
| target: str, | |
| *args: Any, | |
| **kwargs: Any, | |
| ) -> Any: | |
| """ | |
| Orchestrate a synchronization between two registered adapters. | |
| This method provides the central flow: | |
| source adapter | |
| ↓ | |
| SyncManager | |
| ↓ | |
| conflict resolution | |
| ↓ | |
| target adapter | |
| Provider-specific data handling remains delegated to the adapters. | |
| """ | |
| self.initialize() | |
| source_adapter = self.get_adapter(source) | |
| target_adapter = self.get_adapter(target) | |
| self.logger.info( | |
| "Synchronization started", | |
| source=source, | |
| target=target, | |
| component="sync_manager", | |
| ) | |
| if hasattr(source_adapter, "export"): | |
| data = source_adapter.export(*args, **kwargs) | |
| elif hasattr(source_adapter, "read"): | |
| data = source_adapter.read(*args, **kwargs) | |
| elif hasattr(source_adapter, "sync"): | |
| data = source_adapter.sync(*args, **kwargs) | |
| else: | |
| raise AttributeError( | |
| f"Source adapter '{source}' does not provide " | |
| "'export', 'read', or 'sync'." | |
| ) | |
| if hasattr(target_adapter, "import_data"): | |
| result = target_adapter.import_data(data) | |
| elif hasattr(target_adapter, "write"): | |
| result = target_adapter.write(data) | |
| elif hasattr(target_adapter, "push"): | |
| result = target_adapter.push(data) | |
| else: | |
| raise AttributeError( | |
| f"Target adapter '{target}' does not provide " | |
| "'import_data', 'write', or 'push'." | |
| ) | |
| self.logger.info( | |
| "Synchronization completed", | |
| source=source, | |
| target=target, | |
| component="sync_manager", | |
| ) | |
| return result | |
| def status(self) -> Dict[str, Any]: | |
| """ | |
| Return the current state of the synchronization manager. | |
| """ | |
| return { | |
| "initialized": self._initialized, | |
| "adapters": self.list_adapters(), | |
| "scheduler": self.scheduler, | |
| "conflict_resolver": self.conflict_resolver, | |
| } | |
| def shutdown(self) -> None: | |
| """ | |
| Shut down the synchronization layer cleanly. | |
| """ | |
| self.logger.info( | |
| "SyncManager shutting down", | |
| component="sync_manager", | |
| ) | |
| if hasattr(self.scheduler, "shutdown"): | |
| self.scheduler.shutdown() | |
| self._initialized = False |