Spaces:
Sleeping
Sleeping
| """ | |
| sync_manager.py | |
| SkyOS Sync Manager | |
| Zentrale Steuerung aller Synchronisationsprozesse. | |
| Autor: Sky Meilin | |
| Projekt: SkyOS | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from .registry import SyncRegistry | |
| logger = logging.getLogger("SkyOS.Sync") | |
| class SyncManager: | |
| """ | |
| Zentrale Synchronisationsinstanz. | |
| """ | |
| def __init__(self) -> None: | |
| self.registry = SyncRegistry() | |
| # --------------------------------------------------------- | |
| # Quellen | |
| # --------------------------------------------------------- | |
| def register_source( | |
| self, | |
| name: str, | |
| source, | |
| ) -> None: | |
| """ | |
| Registriert eine Datenquelle. | |
| """ | |
| self.registry.register( | |
| name, | |
| source, | |
| ) | |
| logger.info( | |
| "Quelle registriert: %s", | |
| name, | |
| ) | |
| # --------------------------------------------------------- | |
| # Synchronisation | |
| # --------------------------------------------------------- | |
| def sync_all(self) -> None: | |
| """ | |
| Synchronisiert alle registrierten Quellen. | |
| """ | |
| logger.info( | |
| "Starte Gesamtsynchronisation..." | |
| ) | |
| for name, source in self.registry.sources.items(): | |
| try: | |
| logger.info( | |
| "Synchronisiere %s", | |
| name, | |
| ) | |
| if hasattr( | |
| source, | |
| "sync", | |
| ): | |
| source.sync() | |
| except Exception as error: | |
| logger.exception( | |
| "Fehler bei %s: %s", | |
| name, | |
| error, | |
| ) | |
| logger.info( | |
| "Synchronisation abgeschlossen." | |
| ) | |
| # --------------------------------------------------------- | |
| # Informationen | |
| # --------------------------------------------------------- | |
| def list_sources(self) -> list[str]: | |
| """ | |
| Gibt alle registrierten Quellen zurück. | |
| """ | |
| return list( | |
| self.registry.sources.keys() | |
| ) |