UvEventLoop.callback() - Code Metrics - Inspection of "[WIP/RFC] allow polling on a provided fd" - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#72)

by Björn

created 2016-08-28 19:26 UTC

UvEventLoop.callback() A

↳ Parent: UvEventLoop._poll_fd()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

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

"""Event loop implementation that uses pyuv(libuv-python bindings)."""
import sys
from collections import deque

import pyuv
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from .base import BaseEventLoop


class UvEventLoop(BaseEventLoop):

    """`BaseEventLoop` subclass that uses `pvuv` as a backend."""

    def _init(self):
        self._loop = pyuv.Loop()
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._async = pyuv.Async(self._loop, self._on_async)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._connection_error = None
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._error_stream = None
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._callbacks = deque()
class Foo:
    def __init__(self, x=None):
        self.x = x

    def _on_connect(self, stream, error):
        self.stop()
        if error:
            msg = 'Cannot connect to {0}: {1}'.format(
                self._connect_address, pyuv.errno.strerror(error))
            self._connection_error = IOError(msg)
class Foo:
    def __init__(self, x=None):
        self.x = x
            return
        self._read_stream = self._write_stream = stream
class Foo:
    def __init__(self, x=None):
        self.x = x

    def _on_read(self, handle, data, error):
        if error or not data:
            msg = pyuv.errno.strerror(error) if error else 'EOF'
            self._on_error(msg)
            return
        if handle == self._error_stream:
            return
        self._on_data(data)

    def _on_write(self, handle, error):

        if error:
            msg = pyuv.errno.strerror(error)
            self._on_error(msg)

    def _on_exit(self, handle, exit_status, term_signal):

        self._on_error('EOF')

    def _disconnected(self, *args):
class Foo:
    def some_method(self, x, y):
        return x + y;
        raise IOError('Not connected to Nvim')

    def _connect_tcp(self, address, port):
        stream = pyuv.TCP(self._loop)
        self._connect_address = '{0}:{1}'.format(address, port)
class Foo:
    def __init__(self, x=None):
        self.x = x
        stream.connect((address, port), self._on_connect)

    def _connect_socket(self, path):
        stream = pyuv.Pipe(self._loop)
        self._connect_address = path
class Foo:
    def __init__(self, x=None):
        self.x = x
        stream.connect(path, self._on_connect)

    def _connect_stdio(self):
        self._read_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._read_stream.open(sys.stdin.fileno())
        self._write_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._write_stream.open(sys.stdout.fileno())


    def _connect_child(self, argv):
        self._write_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._read_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._error_stream = pyuv.Pipe(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        stdin = pyuv.StdIO(self._write_stream,
                           flags=pyuv.UV_CREATE_PIPE + pyuv.UV_READABLE_PIPE)
        stdout = pyuv.StdIO(self._read_stream,
                            flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
        stderr = pyuv.StdIO(self._error_stream,
                            flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
        self._process = pyuv.Process(self._loop)
class Foo:
    def __init__(self, x=None):
        self.x = x
        self._process.spawn(file=argv[0],
                            exit_callback=self._on_exit,
                            args=argv[1:],
                            flags=pyuv.UV_PROCESS_WINDOWS_HIDE,
                            stdio=(stdin, stdout, stderr,))
        self._error_stream.start_read(self._on_read)

    def _start_reading(self):
        if self._transport_type in ['tcp', 'socket']:
            self._loop.run()
            if self._connection_error:
                self.run = self.send = self._disconnected
                raise self._connection_error

        self._read_stream.start_read(self._on_read)

    def _send(self, data):
        self._write_stream.write(data, self._on_write)

    def _run(self):
        self._loop.run(pyuv.UV_RUN_DEFAULT)

    def _stop(self):
        self._loop.stop()

    def _threadsafe_call(self, fn):
        self._callbacks.append(fn)
        self._async.send()

    def _on_async(self, handle):

        while self._callbacks:
            self._callbacks.popleft()()

    def _poll_fd(self, fd, on_readable, on_writable):
        poll = pyuv.Poll(self._loop, fd)
        events = 0
        if on_readable is not None:
            events |= pyuv.UV_READABLE
        if on_writable is not None:
            events |= pyuv.UV_WRITABLE
        def callback(poll_handle, evts, errorno):

            if evts & pyuv.UV_READABLE:
                on_readable()
            if evts & pyuv.UV_WRITABLE:
                on_writable()

        poll.start(events, callback)
        return poll.stop

    def _setup_signals(self, signals):
        self._signal_handles = []
class Foo:
    def __init__(self, x=None):
        self.x = x

        def handler(h, signum):

            self._on_signal(signum)

        for signum in signals:
            handle = pyuv.Signal(self._loop)
            handle.start(handler, signum)
            self._signal_handles.append(handle)

    def _teardown_signals(self):
        for handle in self._signal_handles:
            handle.stop()


1		"""Event loop implementation that uses pyuv(libuv-python bindings)."""
2	6	import sys
3	6	from collections import deque
4
5	6	import pyuv
		0 ignored issues – show Configuration introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The import `pyuv` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
6
7		from .base import BaseEventLoop
8
9
10		class UvEventLoop(BaseEventLoop):
11
12		"""`BaseEventLoop` subclass that uses `pvuv` as a backend."""
13
14		def _init(self):
15		self._loop = pyuv.Loop()
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_loop` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
16		self._async = pyuv.Async(self._loop, self._on_async)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_async` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
17		self._connection_error = None
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connection_error` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
18		self._error_stream = None
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_error_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
19		self._callbacks = deque()
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_callbacks` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
20
21		def _on_connect(self, stream, error):
22		self.stop()
23		if error:
24		msg = 'Cannot connect to {0}: {1}'.format(
25		self._connect_address, pyuv.errno.strerror(error))
26		self._connection_error = IOError(msg)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connection_error` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
27		return
28		self._read_stream = self._write_stream = stream
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_write_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history... Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_read_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
29
30		def _on_read(self, handle, data, error):
31		if error or not data:
32		msg = pyuv.errno.strerror(error) if error else 'EOF'
33		self._on_error(msg)
34		return
35		if handle == self._error_stream:
36		return
37		self._on_data(data)
38
39		def _on_write(self, handle, error):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `handle` seems to be unused. Loading history...
40		if error:
41		msg = pyuv.errno.strerror(error)
42		self._on_error(msg)
43
44		def _on_exit(self, handle, exit_status, term_signal):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `handle` seems to be unused. Loading history... Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `exit_status` seems to be unused. Loading history... Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `term_signal` seems to be unused. Loading history...
45		self._on_error('EOF')
46
47		def _disconnected(self, *args):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `args` seems to be unused. Loading history... Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
48		raise IOError('Not connected to Nvim')
49
50		def _connect_tcp(self, address, port):
51		stream = pyuv.TCP(self._loop)
52		self._connect_address = '{0}:{1}'.format(address, port)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connect_address` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
53		stream.connect((address, port), self._on_connect)
54
55		def _connect_socket(self, path):
56		stream = pyuv.Pipe(self._loop)
57		self._connect_address = path
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_connect_address` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
58		stream.connect(path, self._on_connect)
59
60		def _connect_stdio(self):
61		self._read_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_read_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
62		self._read_stream.open(sys.stdin.fileno())
63		self._write_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_write_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
64		self._write_stream.open(sys.stdout.fileno())
		0 ignored issues – show Bug introduced 2016-03-06 13:35 UTC by Report Bug Copy Issue Report The Instance of `RedirectStream` does not seem to have a member named `fileno`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
65
66		def _connect_child(self, argv):
67		self._write_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_write_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
68		self._read_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_read_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
69		self._error_stream = pyuv.Pipe(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_error_stream` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
70		stdin = pyuv.StdIO(self._write_stream,
71		flags=pyuv.UV_CREATE_PIPE + pyuv.UV_READABLE_PIPE)
72		stdout = pyuv.StdIO(self._read_stream,
73		flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
74		stderr = pyuv.StdIO(self._error_stream,
75		flags=pyuv.UV_CREATE_PIPE + pyuv.UV_WRITABLE_PIPE)
76		self._process = pyuv.Process(self._loop)
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_process` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
77		self._process.spawn(file=argv[0],
78		exit_callback=self._on_exit,
79		args=argv[1:],
80		flags=pyuv.UV_PROCESS_WINDOWS_HIDE,
81		stdio=(stdin, stdout, stderr,))
82		self._error_stream.start_read(self._on_read)
83
84		def _start_reading(self):
85		if self._transport_type in ['tcp', 'socket']:
86		self._loop.run()
87		if self._connection_error:
88		self.run = self.send = self._disconnected
89		raise self._connection_error
		0 ignored issues – show Bug introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report Raising NoneType while only classes or instances are allowed Loading history...
90		self._read_stream.start_read(self._on_read)
91
92		def _send(self, data):
93		self._write_stream.write(data, self._on_write)
94
95		def _run(self):
96		self._loop.run(pyuv.UV_RUN_DEFAULT)
97
98		def _stop(self):
99		self._loop.stop()
100
101		def _threadsafe_call(self, fn):
102		self._callbacks.append(fn)
103		self._async.send()
104
105		def _on_async(self, handle):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `handle` seems to be unused. Loading history...
106		while self._callbacks:
107		self._callbacks.popleft()()
108
109		def _poll_fd(self, fd, on_readable, on_writable):
110		poll = pyuv.Poll(self._loop, fd)
111		events = 0
112		if on_readable is not None:
113		events \|= pyuv.UV_READABLE
114		if on_writable is not None:
115		events \|= pyuv.UV_WRITABLE
116		def callback(poll_handle, evts, errorno):
		0 ignored issues – show Unused Code introduced 2016-08-28 19:44 UTC by Report Bug Copy Issue Report The argument `poll_handle` seems to be unused. Loading history... Unused Code introduced 2016-08-28 19:44 UTC by Report Bug Copy Issue Report The argument `errorno` seems to be unused. Loading history...
117		if evts & pyuv.UV_READABLE:
118		on_readable()
119		if evts & pyuv.UV_WRITABLE:
120		on_writable()
121
122		poll.start(events, callback)
123		return poll.stop
124
125		def _setup_signals(self, signals):
126		self._signal_handles = []
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The attribute `_signal_handles` was defined outside `__init__`. It is generally a good practice to initialize all attributes to default values in the `__init__` method: class Foo: def __init__(self, x=None): self.x = x Loading history...
127
128		def handler(h, signum):
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The argument `h` seems to be unused. Loading history...
129		self._on_signal(signum)
130
131		for signum in signals:
132		handle = pyuv.Signal(self._loop)
133		handle.start(handler, signum)
134		self._signal_handles.append(handle)
135
136		def _teardown_signals(self):
137		for handle in self._signal_handles:
138		handle.stop()
139

neovim / python-client

Pull Request — master (#72)

UvEventLoop.callback() A

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files