D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
imunify360
/
venv
/
lib64
/
python3.11
/
site-packages
/
defence360agent
/
plugins
/
Filename :
config_merger.py
back
Copy
import logging from defence360agent.contracts.config import ConfigValidationError, Merger from defence360agent.contracts.messages import MessageType from defence360agent.contracts.plugins import MessageSink, expect logger = logging.getLogger(__name__) class ConfigMerger(MessageSink): PROCESSING_ORDER = MessageSink.ProcessingOrder.PRE_PROCESS_MESSAGE def __init__(self): self.loop = None async def create_sink(self, loop): self.loop = loop @expect(MessageType.ConfigUpdate) async def update_merged_config(self, message): try: Merger.update_merged_config() except ConfigValidationError as err: logger.error("Config is invalid. Will not update: %s", err) finally: if event := message.get("event"): event.set()