# 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 signal import warnings import logging from lvestats.lib.commons import logsetup def no_sigpipe(func): def wrapper(*args, **kwargs): old_pipe_signal = signal.getsignal(signal.SIGPIPE) signal.signal(signal.SIGPIPE, signal.SIG_DFL) with warnings.catch_warnings(record=True) as recorded_warnings: retval = func(*args, **kwargs) if recorded_warnings and logsetup.FILE_LOG is not None: log = logging.getLogger("Recorded warnings") for warning in recorded_warnings: log.warning(str(warning)) signal.signal(signal.SIGPIPE, old_pipe_signal) return retval return wrapper