Add flake8 code quality check (#78)

* add flake8 code quality check

* add codacy grade

.flake8

@@ -0,0 +1,12 @@
+[flake8]
+max-complexity = 10
+max-line-length = 127
+exclude =
+    *_pb2.py
+    __init__.py
+    __pycache__
+    mars/lib/enum.py
+    mars/lib/six.py
+    mars/lib/functools32/*
+    mars/lib/futures/*
+    mars/lib/uhashring/*

.travis.yml

@@ -53,6 +53,11 @@ install:
   - if [ -z "$DEFAULT_VENV" ]; then deactivate; else source $DEFAULT_VENV/bin/activate; fi
   - python setup.py build_ext -i
 
+before_script:
+  - pip install flake8
+  - flake8 mars --count --select=E111,E901,E999,F401,F821,F822,F823,F841 --show-source --statistics
+  - flake8 mars --count --exit-zero --statistics
+
 # command to run tests
 script:
   mkdir -p build &&

README.rst

@@ -1,7 +1,7 @@
 Mars
 ====
 
-|PyPI version| |Build| |Coverage| |Docs| |License| |Implementation|
+|PyPI version| |Docs| |Build| |Coverage| |Quality| |License|
 
 Mars is a tensor-based unified framework for large-scale data computation.
 
@@ -174,13 +174,14 @@ Thank you in advance for your contributions!
    :target: https://travis-ci.org/mars-project/mars
 .. |Coverage| image:: https://img.shields.io/coveralls/github/mars-project/mars.svg?style=flat-square
    :target: https://coveralls.io/github/mars-project/mars
+.. |Quality| image:: https://img.shields.io/codacy/grade/4e15343492d14335847d67630bb3c319.svg?style=flat-square
+   :target: https://app.codacy.com/project/mars-project/mars/dashboard
 .. |PyPI version| image:: https://img.shields.io/pypi/v/pymars.svg?style=flat-square
    :target: https://pypi.python.org/pypi/pymars
 .. |Docs| image:: https://img.shields.io/badge/docs-latest-brightgreen.svg?style=flat-square
    :target: http://mars-project.readthedocs.org/
 .. |License| image:: https://img.shields.io/pypi/l/pymars.svg?style=flat-square
    :target: https://github.com/mars-project/mars/blob/master/LICENSE
-.. |Implementation| image:: https://img.shields.io/pypi/implementation/pymars.svg?style=flat-square
 .. _`mars-dev@googlegroups.com`: https://groups.google.com/forum/#!forum/mars-dev
 .. _`GitHub issue`: https://github.com/mars-project/mars/issues
 .. _`pull requests`: https://github.com/mars-project/mars/pulls

docs/source/conf.py

@@ -171,4 +171,4 @@
 # -- Extension configuration -------------------------------------------------
 
 locale_dirs = ['locale/']   # path is example but recommended.
-gettext_compact = False     # optional.
+gettext_compact = False     # optional.

docs/source/norm_zh.py

@@ -86,7 +86,7 @@ def _normalize(string, prefix='', width=76):
                     buf = []
                     size = 2
                     while chunks:
-                        l = _zh_len(escape(chunks[-1])) - 2 + prefixlen
+                        l = _zh_len(escape(chunks[-1])) - 2 + prefixlen  # noqa: E741
                         if size + l < width:
                             buf.append(chunks.pop())
                             size += l
@@ -114,7 +114,7 @@ def _normalize(string, prefix='', width=76):
 
 def main():
     try:
-        import jieba
+        import jieba  # noqa: F401
     except ImportError:
         return
 

mars/_version.py

@@ -12,35 +12,36 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import subprocess
+import os
+import sys
+
 version_info = (0, 1, 0, 'a2')
 _num_index = max(idx if isinstance(v, int) else 0
                  for idx, v in enumerate(version_info))
 __version__ = '.'.join(map(str, version_info[:_num_index + 1])) + \
               ''.join(version_info[_num_index + 1:])
 
 
-def get_git_info():
-    import subprocess
-    import os
-    import sys
+def _get_cmd_results(pkg_root, cmd):
+    proc = subprocess.Popen(cmd, cwd=pkg_root, stdout=subprocess.PIPE)
+    proc.wait()
+    if proc.returncode == 0:
+        s = proc.stdout.read()
+        if sys.version_info[0] >= 3:
+            s = s.decode()
+        return s
+
 
+def get_git_info():
     pkg_root = os.path.dirname(os.path.abspath(__file__))
     git_root = os.path.join(os.path.dirname(pkg_root), '.git')
 
-    def get_cmd_results(cmd):
-        proc = subprocess.Popen(cmd, cwd=pkg_root, stdout=subprocess.PIPE)
-        proc.wait()
-        if proc.returncode == 0:
-            s = proc.stdout.read()
-            if sys.version_info[0] >= 3:
-                s = s.decode()
-            return s
-
     if os.path.exists(git_root):
-        commit_hash = get_cmd_results(['git', 'rev-parse', 'HEAD']).strip()
+        commit_hash = _get_cmd_results(pkg_root, ['git', 'rev-parse', 'HEAD']).strip()
         if not commit_hash:
             return
-        branches = get_cmd_results(['git', 'branch']).splitlines(False)
+        branches = _get_cmd_results(pkg_root, ['git', 'branch']).splitlines(False)
         commit_ref = None
         for l in branches:
             if not l.startswith('*'):

mars/actors/pool/tests/test_gevent_pool.py

@@ -65,7 +65,7 @@ def __init__(self, value):
             raise ValueError('value < 0')
         self.value = value
 
-    def on_receive(self, message):
+    def on_receive(self, message):  # noqa: C901
         if message[0] == 'add':
             if not isinstance(message[1], six.integer_types):
                 raise TypeError('add number must be int')
@@ -568,7 +568,7 @@ def testRemoteConnections(self):
                 with create_actor_pool(address='127.0.0.1:12347', n_process=2, backend='gevent') as pool3:
                     addr3 = pool3.cluster_info.address
 
-                    _ = Connections(addr3)
+                    conns3 = Connections(addr3)
 
                     ps = list()
                     ps.append(gevent.spawn(connections1.connect))
@@ -577,6 +577,8 @@ def testRemoteConnections(self):
 
                     self.assertEqual(len(connections1.conn), 66)
 
+                    del conns3
+
     def testRemotePostCreatePreDestroy(self):
         with create_actor_pool(address=True, n_process=1, backend='gevent') as pool:
             addr = pool.cluster_info.address

mars/base_app.py

@@ -94,8 +94,7 @@ def _main(self, argv=None):
         endpoint = args.endpoint
         host = args.host
         port = args.port
-        if args.kv_store:
-            options.kv_store = args.kv_store
+        options.kv_store = args.kv_store if args.kv_store else options.kv_store
 
         load_modules = []
         for mod in args.load_modules or ():

mars/compat/numpy_compat.py

@@ -48,7 +48,7 @@ def divide(x1, x2, out=None, dtype=None):
         return x
 
 # functions copied from numpy
-try:
+try:  # noqa: C901
     from numpy import broadcast_to, nanprod, nancumsum, nancumprod
 except ImportError:  # pragma: no cover
     # these functions should arrive in numpy v1.10 to v1.12.  Until then,

mars/config.py

@@ -14,10 +14,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from copy import deepcopy
 import contextlib
+import functools
+import operator
 import warnings
 import threading
+from copy import deepcopy
 
 from .compat import six
 
@@ -264,13 +266,19 @@ def validate(x):
     return validate
 
 
-is_null = lambda x: x is None
-is_bool = lambda x: isinstance(x, bool)
-is_integer = lambda x: isinstance(x, six.integer_types)
-is_float = lambda x: isinstance(x, float)
-is_string = lambda x: isinstance(x, six.string_types)
-is_dict = lambda x: isinstance(x, dict)
-is_list = lambda x: isinstance(x, list)
+def _instance_check(typ, v):
+    return isinstance(v, typ)
+
+
+is_null = functools.partial(operator.is_, None)
+is_bool = functools.partial(_instance_check, bool)
+is_integer = functools.partial(_instance_check, six.integer_types)
+is_float = functools.partial(_instance_check, float)
+is_string = functools.partial(_instance_check, six.string_types)
+is_dict = functools.partial(_instance_check, dict)
+is_list = functools.partial(_instance_check, list)
+
+
 def is_in(vals):
     def validate(x):
         return x in vals

mars/deploy/local/core.py

@@ -164,6 +164,27 @@ def _start_cluster(endpoint, event, n_process=None, shared_memory=None, **kw):
         cluster.stop_service()
 
 
+def _start_cluster_process(endpoint, n_process, shared_memory, **kw):
+    event = multiprocessing.Event()
+
+    kw = kw.copy()
+    kw['n_process'] = n_process
+    kw['shared_memory'] = shared_memory or options.worker.cache_memory_limit or '20%'
+    process = gipc.start_process(_start_cluster, args=(endpoint, event), kwargs=kw)
+
+    while True:
+        event.wait(5)
+        if not event.is_set():
+            # service not started yet
+            continue
+        if not process.is_alive():
+            raise SystemError('New local cluster failed')
+        else:
+            break
+
+    return process
+
+
 def _start_web(scheduler_address, ui_port, event):
     import gevent.monkey
     gevent.monkey.patch_all(thread=False)
@@ -177,6 +198,25 @@ def _start_web(scheduler_address, ui_port, event):
         web.stop()
 
 
+def _start_web_process(scheduler_endpoint, web_endpoint):
+    web_event = multiprocessing.Event()
+    ui_port = int(web_endpoint.rsplit(':', 1)[1])
+    web_process = gipc.start_process(
+        _start_web, args=(scheduler_endpoint, ui_port, web_event), daemon=True)
+
+    while True:
+        web_event.wait(5)
+        if not web_event.is_set():
+            # web not started yet
+            continue
+        if not web_process.is_alive():
+            raise SystemError('New web interface failed')
+        else:
+            break
+
+    return web_process
+
+
 class LocalDistributedClusterClient(object):
     def __init__(self, endpoint, web_endpoint, cluster_process, web_process):
         self._cluster_process = cluster_process
@@ -227,35 +267,10 @@ def new_cluster(address='0.0.0.0', web=False, n_process=None, shared_memory=None
         else:
             web_endpoint = gen_endpoint(web)
 
-    event = multiprocessing.Event()
-    kw['n_process'] = n_process
-    kw['shared_memory'] = shared_memory or options.worker.cache_memory_limit or '20%'
-    process = gipc.start_process(_start_cluster, args=(endpoint, event), kwargs=kw)
-
-    while True:
-        event.wait(5)
-        if not event.is_set():
-            # service not started yet
-            continue
-        if not process.is_alive():
-            raise SystemError('New local cluster failed')
-        else:
-            break
+    process = _start_cluster_process(endpoint, n_process, shared_memory, **kw)
 
     web_process = None
     if web_endpoint:
-        web_event = multiprocessing.Event()
-        ui_port = int(web_endpoint.rsplit(':', 1)[1])
-        web_process = gipc.start_process(_start_web, args=(endpoint, ui_port, web_event), daemon=True)
-
-        while True:
-            web_event.wait(5)
-            if not web_event.is_set():
-                # web not started yet
-                continue
-            if not web_process.is_alive():
-                raise SystemError('New web interface failed')
-            else:
-                break
+        web_process = _start_web_process(endpoint, web_endpoint)
 
     return LocalDistributedClusterClient(endpoint, web_endpoint, process, web_process)

mars/deploy/local/session.py

@@ -19,6 +19,7 @@
 import time
 
 from ...api import MarsAPI
+from ...compat import TimeoutError
 from ...graph import DirectedGraph
 from ...scheduler.graph import GraphState
 from ...serialize import dataserializer

mars/deploy/local/tests/test_cluster.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import functools
 import unittest
 import sys
 
@@ -87,7 +88,7 @@ def testLocalClusterWithWeb(self):
                 np.testing.assert_array_equal(result, np.ones((3, 3)))
 
     def testNSchedulersNWorkers(self):
-        calc_cpu_cnt = lambda: 4
+        calc_cpu_cnt = functools.partial(lambda: 4)
 
         self.assertEqual(LocalDistributedCluster._calc_scheduler_worker_n_process(
             None, None, None, calc_cpu_count=calc_cpu_cnt), (2, 4))