1#@!#!123s

: / opt / cloudlinux / venv / lib / python3.11 / site-packages / clevents /
Filename : event.py
back
#!/opt/cloudlinux/venv/bin/python3 -bb
# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import syslog
import types  # NOQA
from traceback import format_exc


class Event:
    """
    Event is an object that can have several listeners.
    Register new handler using '.register()' method.
    When you call '.throw_event(*a, **kw)' , class notifies all registered listeners.

    This class is logging his actions to syslog.
    Exceptions happened in hooks are logged as warnings, all
    other messages have debug level. For debug purposes you can
    manually change syslog level in /etc/rsyslog.conf file.
    """
    # messages with level 'DEBUG' needed only for developers
    # so hide them by default in order not to confuse clients
    WRITE_DEBUG_TO_SYSLOG = False

    def __init__(self):
        self._listeners = set()

    @classmethod
    def _log_message(cls, level, message):
        # type: (int, str) -> None
        if level == syslog.LOG_DEBUG and not cls.WRITE_DEBUG_TO_SYSLOG:
            return
        syslog.syslog(level, message)

    # noinspection PyBroadException
    def _run_or_log_exception(self, func, *args, **kwargs):
        # type: (types.FunctionType, list, dict) -> None
        """Run callable object func and forward exceptions to syslog"""
        try:
            func(*args, **kwargs)
        except BaseException:
            footprint = self._get_function_footprint(func)
            message = (
                "WARNING: An error occurred while notifying handler "
                f"{footprint} with ({args}, {kwargs}). Following error raised: {format_exc()}."
                "Please, contact CloudLinux support if it happens again."
            )
            self._log_message(syslog.LOG_WARNING, message)
        else:
            message = f"DEBUG: Handler {func} notified with params: ({args}, {kwargs})"
            self._log_message(syslog.LOG_DEBUG, message)

    @staticmethod
    def _get_function_footprint(func):
        # type: (types.FunctionType) -> str
        return func.__name__ + ':' + func.__module__

    # TODO: add arguments match validation here
    def register(self, func):
        # type: (types.FunctionType) -> types.FunctionType
        self._listeners.add(func)
        self._log_message(
            syslog.LOG_DEBUG,
            f"DEBUG: Registered new handler {self._get_function_footprint(func)}."
        )
        # return func here in order to be able to use multiple decorators at once
        # otherwise, decorated function becomes 'None'
        return func

    def unregister(self, func):
        # type: (types.FunctionType) -> None
        if func in self._listeners:
            self._listeners.remove(func)
        self._log_message(
            syslog.LOG_DEBUG,
            f"DEBUG: Unregister handler {self._get_function_footprint(func)}."
        )

    def throw_event(self, *args, **kwargs):
        # type: (list, dict) -> None
        for func in self._listeners:
            self._run_or_log_exception(func, *args, **kwargs)