MAINHACK

Viewing / Editing File: backup_info_sender.py

File is not writable. Editing disabled.

import asyncio
from datetime import timedelta
from logging import getLogger

from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import MessageSource
from defence360agent.subsys.backup_systems import (
    get_current_backend,
    get_last_backup_timestamp,
)
from defence360agent.utils import Scope

logger = getLogger(__name__)

SEND_INTERVAL = int(timedelta(hours=24).total_seconds())


class BackupInfoSender(MessageSource):
    SCOPE = Scope.IM360

    async def create_source(self, loop, sink):
        self._loop = loop
        self._sink = sink
        self._task = self._loop.create_task(self._recurring_check())

    async def shutdown(self):
        self._task.cancel()
        await self._task

    async def _recurring_check(self):
        while True:
            try:
                await asyncio.sleep(SEND_INTERVAL)
                await self._send_server_config()
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.exception("Failed to collect backup info: %s", e)

    async def _send_server_config(self):
        confg_msg = MessageType.BackupInfo(
            backup_provider_type=get_current_backend(),
            last_backup_timestamp=await get_last_backup_timestamp(),
        )
        await self._sink.process_message(confg_msg)

Cancel / Back

[ MAINHACK ]

Viewing / Editing File: backup_info_sender.py

Server Info

System Features

User Info