✏️ Editing: files_recurring_update.py Read Only

import logging

from defence360agent import files
from defence360agent.contracts import config, messages
from defence360agent.contracts.plugins import MessageSource
from defence360agent.utils import recurring_check

logger = logging.getLogger(__name__)

class FilesRecurringUpdateTask(MessageSource):
    async def _on_files_update(
        self, index: files.Index, is_updated: bool
    ) -> None:
        if is_updated:
            message = messages.MessageType.FilesUpdated(index.type, index)
            await self._sink.process_message(message)

async def create_source(self, loop, sink):
        self._loop = loop
        self._sink = sink
        self._task = loop.create_task(self._update_task())
        # subscribe to file updates
        for type_ in files.Index.types():
            files.Index.add_hook(type_, self._on_files_update)

async def shutdown(self):
        self._task.cancel()
        # CancelledError is handled by @recurring_check():
        await self._task

@recurring_check(config.FilesUpdate.PERIOD)
    async def _update_task(self):
        await files.update_and_log_error()

🔐 Sid Gifari File Manager Pro