22

✏️ Editing: incidents.py

from defence360agent.model.simplification import remove_old_and_truncate
from defence360agent.rpc_tools import ValidationError, lookup
from defence360agent.rpc_tools.utils import run_in_executor_decorator
import defence360agent.subsys.panels.hosting_panel as hp
from defence360agent.utils import Scope
from im360.contracts import config
from im360.model.firewall import Purpose
from im360.model.incident import Incident

class IncidentsEndpoints(lookup.RootEndpoints):
    SCOPE = Scope.IM360

@lookup.bind("get")
    async def get_incident_list(self, *args, **kwargs):
        if ("since" not in kwargs) and ("period" not in kwargs):
            raise ValidationError("Either since or period should be specified")
        period = kwargs.pop("period", None)
        if period is not None:
            kwargs["since"] = period.since
            kwargs["to"] = period.to
        severity = kwargs.pop("severity", None)
        if kwargs.get("by_purpose"):
            # backward compatibility
            kwargs["by_list"] = [
                Purpose.listname(item) for item in kwargs.pop("by_purpose")
            ]
        by_panel_user = kwargs.pop("by_panel_user", None)
        user_domains = None
        if by_panel_user is not None:
            user_domains = (
                await hp.HostingPanel().get_domains_per_user()
            ).get(by_panel_user, [])

return Incident.get_sorted_incident_list(
            *args,
            severity=severity or config.IncidentLogging.MIN_LOG_LEVEL,
            by_domains=user_domains,
            **kwargs
        )

@lookup.bind("clean")
    @run_in_executor_decorator
    def clean_incidents(self, days=None, limit=None):
        return remove_old_and_truncate(
            Incident,
            days or config.IncidentLogging.NUM_DAYS,
            limit or config.IncidentLogging.LIMIT,
        )

MR SKB