WebSocketHandler.get_handler() - Code Metrics - Inspection of "Socketio" - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#436)

by Carlos Eduardo

created 2017-05-23 22:05 UTC

WebSocketHandler.get_handler() A

↳ Parent: WebSocketHandler

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.512

Importance

Changes	1
Bugs	1	Features	0

Metric	Value
cc	1
dl	0
loc	12
ccs	1
cts	5
cp	0.2
crap	1.512
rs	9.4285
c	1
b	1
f	0

"""WebSocket abstraction."""
import logging

__all__ = ('WebSocketHandler')


class WebSocketHandler:
    """Log handler that logs to web socket."""

    @classmethod
    def get_handler(cls, socket):
        """Output logs to a web socket, filtering unwanted messages.

        Args:
            socket: socketio socket.
            stream: Object that supports ``write()`` and ``flush()``.
        """
        stream = WebSocketStream(socket)
        handler = logging.StreamHandler(stream)
        handler.addFilter(cls._filter_web_requests)
        return handler

    @staticmethod
    def _filter_web_requests(record):
        """Only allow web server messages with level higher than info.

        Do not print web requests (INFO level) to avoid infinit loop when
        printing the logs in the web interface with long-polling mode.
        """
        return record.name != 'werkzeug' or record.levelno > logging.INFO


class WebSocketStream:
    """Make loggers write to web socket."""

    def __init__(self, socketio):
        """Receive the socket to write to."""
        super().__init__()
        self._io = socketio
        self._content = ''

    def write(self, content):
        """Store a new line."""
        self._content += content

    def flush(self):
        """Send lines and reset the content."""
        lines = self._content.split('\n')[:-1]
        self._content = ''
        self._io.emit('show logs', lines, room='log')


1		"""WebSocket abstraction."""
2	1	import logging
3	1
4		__all__ = ('WebSocketHandler')
5	1
6
7		class WebSocketHandler:
8	1	"""Log handler that logs to web socket."""
9
10		@classmethod
11	1	def get_handler(cls, socket):
12		"""Output logs to a web socket, filtering unwanted messages.
13
14		Args:
15		socket: socketio socket.
16		stream: Object that supports ``write()`` and ``flush()``.
17		"""
18		stream = WebSocketStream(socket)
19		handler = logging.StreamHandler(stream)
20		handler.addFilter(cls._filter_web_requests)
21		return handler
22
23	1	@staticmethod
24		def _filter_web_requests(record):
25		"""Only allow web server messages with level higher than info.
26
27		Do not print web requests (INFO level) to avoid infinit loop when
28		printing the logs in the web interface with long-polling mode.
29	1	"""
30		return record.name != 'werkzeug' or record.levelno > logging.INFO
31
32
33		class WebSocketStream:
34	1	"""Make loggers write to web socket."""
35
36		def __init__(self, socketio):
37		"""Receive the socket to write to."""
38		super().__init__()
39		self._io = socketio
40		self._content = ''
41
42		def write(self, content):
43		"""Store a new line."""
44		self._content += content
45
46		def flush(self):
47		"""Send lines and reset the content."""
48		lines = self._content.split('\n')[:-1]
49	1	self._content = ''
50		self._io.emit('show logs', lines, room='log')
51

kytos / kytos

Pull Request — master (#436)

WebSocketHandler.get_handler() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like