config.py
| 999 B | Satir:
0
| py
Geri
import asyncio import time from logging import getLogger from defence360agent.contracts import config, messages from defence360agent.feature_management import checkers CONFIG_UPDATE_TIMEOUT = config.SimpleRpc.CLIENT_TIMEOUT / 2 logger = getLogger(__name__) OBSOLETE_SECTION = "KERNELCARE" OBSOLETE_OPTION = "edf" def warn_obsolete_option(data): if OBSOLETE_OPTION in data.get(OBSOLETE_SECTION, dict()): logger.warning( "Configuration update with an obsolete kernelcare option 'edf'." " This option has no effect." ) async def update_config(sink, data, user=None): warn_obsolete_option(data) checkers.config_validation(data, user) conf = config.ConfigFile(user) conf.dict_to_config(data, without_defaults=True) updated = asyncio.Event() await sink.process_message( messages.ConfigUpdate(conf=conf, timestamp=time.time(), event=updated) ) await asyncio.wait_for(updated.wait(), timeout=CONFIG_UPDATE_TIMEOUT)
Kaydet
Ctrl+S ile kaydet