Spaces:
Sleeping
Sleeping
File size: 2,137 Bytes
9547843 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """
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()
) |