X7ROOT File Manager
Current Path:
/opt/cloudlinux/venv/lib/python3.11/site-packages/clevents
opt
/
cloudlinux
/
venv
/
lib
/
python3.11
/
site-packages
/
clevents
/
ðŸ“
..
📄
__init__.py
(369 B)
ðŸ“
__pycache__
📄
event.py
(3.1 KB)
📄
events.py
(555 B)
📄
handlers.py
(1.22 KB)
Editing: event.py
#!/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)
Upload File
Create Folder