我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用logging.Handler()。
def __init_logging(self): try: # To avoid "No handler found" warnings for Python version < 2.7 from logging import NullHandler except ImportError: class NullHandler(logging.Handler): def emit(self, record): pass self.__logger = logging.getLogger(self.__logger_name) self.__logger.addHandler(NullHandler()) # defaults to the do-nothing NullHandler #self.__logger.setLevel(logging.INFO) # defaults to logging.INFO # Logger test messages #self.__logger.debug('debug message') #self.__logger.info('info message') # self.__logger.warn('warn message') # self.__logger.error('error message') # self.__logger.critical('critical message')
def __init__(self, host, port): """ Initializes the handler with a specific host address and port. The attribute 'closeOnError' is set to 1 - which means that if a socket error occurs, the socket is silently closed and then reopened on the next logging call. """ logging.Handler.__init__(self) self.host = host self.port = port self.sock = None self.closeOnError = 0 self.retryTime = None # # Exponential backoff parameters. # self.retryStart = 1.0 self.retryMax = 30.0 self.retryFactor = 2.0
def __init__(self, address=('localhost', SYSLOG_UDP_PORT), facility=LOG_USER, socktype=socket.SOCK_DGRAM): """ Initialize a handler. If address is specified as a string, a UNIX socket is used. To log to a local syslogd, "SysLogHandler(address="/dev/log")" can be used. If facility is not specified, LOG_USER is used. """ logging.Handler.__init__(self) self.address = address self.facility = facility self.socktype = socktype if isinstance(address, basestring): self.unixsocket = 1 self._connect_unixsocket(address) else: self.unixsocket = 0 self.socket = socket.socket(socket.AF_INET, socktype) if socktype == socket.SOCK_STREAM: self.socket.connect(address) self.formatter = None
def __init__(self, appname, dllname=None, logtype="Application"): logging.Handler.__init__(self) try: import win32evtlogutil, win32evtlog self.appname = appname self._welu = win32evtlogutil if not dllname: dllname = os.path.split(self._welu.__file__) dllname = os.path.split(dllname[0]) dllname = os.path.join(dllname[0], r'win32service.pyd') self.dllname = dllname self.logtype = logtype self._welu.AddSourceToRegistry(appname, dllname, logtype) self.deftype = win32evtlog.EVENTLOG_ERROR_TYPE self.typemap = { logging.DEBUG : win32evtlog.EVENTLOG_INFORMATION_TYPE, logging.INFO : win32evtlog.EVENTLOG_INFORMATION_TYPE, logging.WARNING : win32evtlog.EVENTLOG_WARNING_TYPE, logging.ERROR : win32evtlog.EVENTLOG_ERROR_TYPE, logging.CRITICAL: win32evtlog.EVENTLOG_ERROR_TYPE, } except ImportError: print("The Python Win32 extensions for NT (service, event "\ "logging) appear not to be available.") self._welu = None
def patch_async_emit(handler: Handler): base_emit = handler.emit queue = Queue() def loop(): while True: record = queue.get() try: base_emit(record) except: print(sys.exc_info()) def async_emit(record): queue.put(record) thread = Thread(target=loop) thread.daemon = True thread.start() handler.emit = async_emit return handler
def __init__(self, access_token=None, environment=None, level=logging.INFO, history_size=10, history_level=logging.DEBUG): logging.Handler.__init__(self) if access_token is not None: rollbar.init(access_token, environment) self.notify_level = level self.history_size = history_size if history_size > 0: self._history.records = [] self.setHistoryLevel(history_level)
def __init__(self, *args, **kwargs): client = kwargs.pop('client_cls', Client) if len(args) == 1: arg = args[0] args = args[1:] if isinstance(arg, Client): self.client = arg else: raise ValueError( 'The first argument to %s must be a Client instance, ' 'got %r instead.' % ( self.__class__.__name__, arg, )) elif 'client' in kwargs: self.client = kwargs.pop('client') else: self.client = client(*args, **kwargs) logging.Handler.__init__(self, level=kwargs.get('level', logging.NOTSET))
def __init__(self, address=('localhost', SYSLOG_UDP_PORT), facility=LOG_USER, socktype=socket.SOCK_DGRAM): """ Initialize a handler. If address is specified as a string, a UNIX socket is used. To log to a local syslogd, "SysLogHandler(address="/dev/log")" can be used. If facility is not specified, LOG_USER is used. """ logging.Handler.__init__(self) self.address = address self.facility = facility self.socktype = socktype if isinstance(address, str): self.unixsocket = 1 self._connect_unixsocket(address) else: self.unixsocket = 0 self.socket = socket.socket(socket.AF_INET, socktype) if socktype == socket.SOCK_STREAM: self.socket.connect(address) self.formatter = None
def test_load_handlers_onlyused(self): self.m_obj.load_config({ "loggers" : { "root" : { "handlers" : ["h2"] } }, "formatters" : { "default" : { "class" : "logging.Formatter" } }, "handlers" : { "h1" : { "class" : "logging.Handler" }, "h2" : { "class" : "logging.Handler" } } }) self.m_obj._load_formatters() self.m_obj._load_handlers() self.assertEqual(len(self.m_obj.m_handlers), 1) self.assertEqual(list(self.m_obj.m_handlers.keys())[0], "h2")
def test_load_handlers_noclass(self): self.m_obj.load_config({ "loggers" : { "root" : { "handlers" : ["h2"] } }, "formatters" : { "default" : { "class" : "logging.Formatter" } }, "handlers" : { "h2" : { "attrib" : "logging.Handler" } } }) self.m_obj._load_formatters() with self.assertRaises(error.XtdError): self.m_obj._load_handlers()
def __init__(self, level=logging.NOTSET, host=mongo_server, port=27017, database_name='logs', collection='logs', username=None, password=None, fail_silently=False, formatter=None): """Setting up mongo handler, initializing mongo database connection via pymongo.""" logging.Handler.__init__(self, level) self.host = host self.port = port self.database_name = database_name self.collection_name = collection self.username = username self.password = password self.fail_silently = fail_silently self.connection = None self.db = None self.collection = None self.authenticated = False self.formatter = formatter or MongoFormatter() self._connect()
def __init__(self, host, port): """ Initializes the handler with a specific host address and port. When the attribute *closeOnError* is set to True - if a socket error occurs, the socket is silently closed and then reopened on the next logging call. """ logging.Handler.__init__(self) self.host = host self.port = port self.sock = None self.closeOnError = False self.retryTime = None # # Exponential backoff parameters. # self.retryStart = 1.0 self.retryMax = 30.0 self.retryFactor = 2.0
def test_log_stdlib(framework_aio): # for cases when we never call start_logging(), ensure we didn't # no-op out the info messages. import logging lg = logging.getLogger() lg.setLevel(logging.INFO) records = [] class TestHandler(logging.Handler): def emit(self, record): records.append(record.msg) handler = TestHandler() lg.addHandler(handler) try: log = txaio.make_logger() log.info("foo={foo}", foo='bar') finally: lg.removeHandler(handler) assert 'foo=bar' in records
def __init__(self): logging.Handler.__init__(self)
def __init__(self, *args, **kwargs): logging.Handler.__init__(self, *args, **kwargs)
def __init__(self, *args, **kwds): logging.Handler.__init__(self, *args, **kwds ) self.log4pyProps = {} self.channelName = None self.nameContext = None self.prodId = "RESOURCE.ID" self.prodName = "RESOURCE.NAME" self.prodFQN = "RESOURCE.FQN" self._channelName = None self._nameContext = None self._event_channel = None self._ecm = None self._pub = None self._enable = False self.threshold = logging.NOTSET
def handleError(self, record): """ Handle an error during logging. An error has occurred during logging. Most likely cause - connection lost. Close the socket so that we can retry on the next event. """ if self.closeOnError and self.sock: self.sock.close() self.sock = None #try to reconnect next time else: logging.Handler.handleError(self, record)
def close(self): """ Closes the socket. """ if self.sock: self.sock.close() self.sock = None logging.Handler.close(self)
def close (self): """ Closes the socket. """ if self.unixsocket: self.socket.close() logging.Handler.close(self)
def close(self): """ Clean up this handler. You can remove the application name from the registry as a source of event log entries. However, if you do this, you will not be able to see the events as you intended in the Event Log Viewer - it needs to be able to access the registry to get the DLL name. """ #self._welu.RemoveSourceFromRegistry(self.appname, self.logtype) logging.Handler.close(self)
def __init__(self, host, url, method="GET"): """ Initialize the instance with the host, the request URL, and the method ("GET" or "POST") """ logging.Handler.__init__(self) method = method.upper() if method not in ["GET", "POST"]: raise ValueError("method must be GET or POST") self.host = host self.url = url self.method = method
def __init__(self, capacity): """ Initialize the handler with the buffer size. """ logging.Handler.__init__(self) self.capacity = capacity self.buffer = []
def close(self): """ Close the handler. This version just flushes and chains to the parent class' close(). """ self.flush() logging.Handler.close(self)
def __init__(self, *args, **kwargs): self.reset() logging.Handler.__init__(self, *args, **kwargs)
def __init__(self, max_records=200): logging.Handler.__init__(self) self.logrecordstotal = 0 self.max_records = max_records try: self.db = deque([], max_records) except TypeError: # pre 2.6 self.db = deque([])
def pytest_sessionstart(self, session): handler = Handler() logging.root.addHandler(handler) yield
def _setup_log(): from usb import _debug logger = logging.getLogger('usb') debug_level = os.getenv('PYUSB_DEBUG') if debug_level is not None: _debug.enable_tracing(True) filename = os.getenv('PYUSB_LOG_FILENAME') LEVELS = {'debug': logging.DEBUG, 'info': logging.INFO, 'warning': logging.WARNING, 'error': logging.ERROR, 'critical': logging.CRITICAL} level = LEVELS.get(debug_level, logging.CRITICAL + 10) logger.setLevel(level = level) try: handler = logging.FileHandler(filename) except: handler = logging.StreamHandler() fmt = logging.Formatter('%(asctime)s %(levelname)s:%(name)s:%(message)s') handler.setFormatter(fmt) logger.addHandler(handler) else: class NullHandler(logging.Handler): def emit(self, record): pass # We set the log level to avoid delegation to the # parent log handler (if there is one). # Thanks to Chris Clark to pointing this out. logger.setLevel(logging.CRITICAL + 10) logger.addHandler(NullHandler())
def __init__(self, room): logging.Handler.__init__(self) self.room = room formatter = logging.Formatter( "%(asctime)s %(levelname)s %(message)s", "%Y-%m-%d %H:%M:%SZ") formatter.converter = time.gmtime self.setFormatter(formatter)
def handle(handler_spec, input_events): r""" Return a list of event dictionaries collected by handling input_events using handler_spec. >>> from abusehelper.core.events import Event >>> from abusehelper.core.transformation import Handler >>> >>> class MyHandler(Handler): ... @idiokit.stream ... def transform(self): ... while True: ... event = yield idiokit.next() ... event.add("a", "b") ... yield idiokit.send(event) ... >>> handle(MyHandler, [{}]) [{u'a': [u'b']}] Note that to simplify testing the output is a list of dictionaries instead of abusehelper.core.events.Event objects. """ handler_type = handlers.load_handler(handler_spec) log = logging.getLogger("Null") log_handler = _NullHandler() log.addHandler(log_handler) try: handler = handler_type(log=log) return idiokit.main_loop(idiokit.pipe( _feed(itertools.imap(events.Event, input_events)), handler.transform(), _collect_events() )) finally: log.removeHandler(log_handler)
def __init__(self, include_html=False, email_backend=None): logging.Handler.__init__(self) self.include_html = include_html self.email_backend = email_backend
def __init__(self, interface_or_socket, context=None): logging.Handler.__init__(self) if isinstance(interface_or_socket, zmq.Socket): self.socket = interface_or_socket self.ctx = self.socket.context else: self.ctx = context or zmq.Context() self.socket = self.ctx.socket(zmq.PUB) self.socket.bind(interface_or_socket)
def __init__(self, music_player, embed, line): """ Args: embed (ui_embed.UI): line (int): """ logging.Handler.__init__(self) self.music_player = music_player self.embed = embed self.line = line
def close(self): """ Closes the socket. """ self.acquire() try: if self.sock: self.sock.close() self.sock = None finally: self.release() logging.Handler.close(self)
def __init__(self, address=('localhost', SYSLOG_UDP_PORT), facility=LOG_USER, socktype=None): """ Initialize a handler. If address is specified as a string, a UNIX socket is used. To log to a local syslogd, "SysLogHandler(address="/dev/log")" can be used. If facility is not specified, LOG_USER is used. If socktype is specified as socket.SOCK_DGRAM or socket.SOCK_STREAM, that specific socket type will be used. For Unix sockets, you can also specify a socktype of None, in which case socket.SOCK_DGRAM will be used, falling back to socket.SOCK_STREAM. """ logging.Handler.__init__(self) self.address = address self.facility = facility self.socktype = socktype if isinstance(address, basestring): self.unixsocket = 1 self._connect_unixsocket(address) else: self.unixsocket = 0 if socktype is None: socktype = socket.SOCK_DGRAM self.socket = socket.socket(socket.AF_INET, socktype) if socktype == socket.SOCK_STREAM: self.socket.connect(address) self.socktype = socktype self.formatter = None
def close (self): """ Closes the socket. """ self.acquire() try: if self.unixsocket: self.socket.close() finally: self.release() logging.Handler.close(self)
