pylint

Author: luminousmining

.github/workflows/lint.yml

@@ -0,0 +1,21 @@
+name: lint
+
+on:
+  push:
+    branches: [main]
+
+jobs:
+  pylint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install dev requirements
+        run: pip install -r requirements-dev.txt
+
+      - name: Run pylint
+        run: pylint main.py core/ stratums/ utils/ scripts/

.pylintrc

@@ -0,0 +1,23 @@
+[MAIN]
+py-version = 3.11
+
+[MESSAGES CONTROL]
+disable =
+    missing-module-docstring,
+    missing-class-docstring,
+    missing-function-docstring,
+    broad-exception-caught,
+    logging-fstring-interpolation,
+    duplicate-code,
+    cyclic-import,
+    too-few-public-methods,
+    too-many-instance-attributes,
+    too-many-arguments,
+    too-many-positional-arguments,
+    too-many-nested-blocks
+
+[FORMAT]
+max-line-length = 120
+
+[BASIC]
+good-names = i,j,k,_

core/pool.py

@@ -22,10 +22,10 @@ def __init__(self, algo: str, hostname: str, port: int,
         self.algo = str(algo)
         self.hostname = str(hostname)
         self.port = int(port)
-        self.__clients = dict()
+        self.__clients = {}
         self.__socket = None
         self.alive = False
-        self.threadAccept = None
+        self.thread_accept = None
         self.stratum = None
 
         if algo == ALGORITHM.SMART_MINING:
@@ -57,8 +57,8 @@ def bind(self) -> None:
         self.__socket.bind((self.hostname, self.port))
         self.__socket.listen(1)
 
-        self.threadAccept = threading.Thread(target=self.__accept, args=())
-        self.threadAccept.start()
+        self.thread_accept = threading.Thread(target=self.__accept, args=())
+        self.thread_accept.start()
 
     def __accept(self) -> None:
         try:
@@ -74,8 +74,8 @@ def __accept(self) -> None:
                                        args=(addr[1], sock))
         thread_loop.start()
 
-        self.threadAccept = threading.Thread(target=self.__accept, args=())
-        self.threadAccept.start()
+        self.thread_accept = threading.Thread(target=self.__accept, args=())
+        self.thread_accept.start()
 
     def remove_client(self, by: str, addr: int) -> None:
         logging.warning(f'Remove client {addr} - {by}')
@@ -112,11 +112,9 @@ def __on_client(self, addr, sock) -> None:
                             self.stratum.on_message(sock, data)
             except TimeoutError:
                 pass
-            except socket.timeout:
-                pass
             except ConnectionAbortedError:
                 self.remove_client('ConnectionAbortedError', addr)
                 return
-            except Exception as e:
-                self.remove_client(f'Exception[{e}]', addr)
+            except Exception as exc:
+                self.remove_client(f'Exception[{exc}]', addr)
                 return

main.py

@@ -7,12 +7,12 @@
 from core import Pool
 
 
-log_level = logging.DEBUG
+LOG_LEVEL = logging.DEBUG
 logging.basicConfig(
     format='[%(levelname)s][%(asctime)s]: %(message)s',
     datefmt='%m/%d/%Y %I:%M:%S %p',
     encoding='utf-8',
-    level=log_level)
+    level=LOG_LEVEL)
 
 
 parser = argparse.ArgumentParser(description='')

requirements-dev.txt

@@ -0,0 +1 @@
+pylint

scripts/test_workflow_client.py

@@ -13,8 +13,8 @@ def wait_for_server(timeout: int) -> None:
     deadline = time.monotonic() + timeout
     while time.monotonic() < deadline:
         try:
-            s = socket.create_connection((HOST, PORT), timeout=1)
-            s.close()
+            sock = socket.create_connection((HOST, PORT), timeout=1)
+            sock.close()
             return
         except OSError:
             time.sleep(0.2)

stratums/autolykos_v2.py

@@ -6,16 +6,13 @@
 
 class StratumAutolykosv2(Stratum):
 
-    def __init__(self):
-        super(StratumAutolykosv2, self).__init__()
-
     def on_message(self, sock, data: dict):
         if 'method' in data:
             self.__on_method(sock, data)
         else:
             self.__on_response(sock, data)
 
-    def __on_response(self, sock, data: dict):
+    def __on_response(self, _sock, data: dict):
         logging.info(f'response => {data}')
 
     def __on_method(self, sock, data: dict):
@@ -36,7 +33,7 @@ def __on_mining_subscribe(self, sock,  request_id: Union[int, str]):
         size = 6
 
         key_id = f'"id":{request_id}'
-        if type(request_id) is str:
+        if isinstance(request_id, str):
             key_id = f'"id":"{request_id}"'
 
         body = '{' \
@@ -48,7 +45,7 @@ def __on_mining_subscribe(self, sock,  request_id: Union[int, str]):
 
     def __on_mining_authorize(self, sock, request_id: Union[int, str]):
         key_id = f'"id":{request_id}'
-        if type(request_id) is str:
+        if isinstance(request_id, str):
             key_id = f'"id":"{request_id}"'
 
         body = '{' \
@@ -84,7 +81,7 @@ def __on_mining_authorize(self, sock, request_id: Union[int, str]):
 
     def __on_mining_submit(self, sock, request_id: Union[int, str], params: list):
         key_id = request_id
-        if type(request_id) is str:
+        if isinstance(request_id, str):
             key_id = f'"{request_id}"'
 
         logging.info(f'Nonce: {params}')

stratums/blake3.py

@@ -1,20 +1,18 @@
+# pylint: disable=line-too-long
 import logging
 
 from stratums import Stratum
 
 
 class StratumBlake3(Stratum):
 
-    def __init__(self):
-        super(StratumBlake3, self).__init__()
-
     def on_message(self, sock, data: dict):
         if 'method' in data:
             self.__on_method(sock, data)
         else:
             self.__on_response(sock, data)
 
-    def __on_response(self, sock, data: dict):
+    def __on_response(self, _sock, data: dict):
         logging.info(f'response => {data}')
 
     def __on_method(self, sock, data: dict):

stratums/ethash.py

@@ -5,16 +5,13 @@
 
 class StratumEthash(Stratum):
 
-    def __init__(self):
-        super(StratumEthash, self).__init__()
-
     def on_message(self, sock, data: dict):
         if 'method' in data:
             self.__on_method(sock, data)
         else:
             self.__on_response(sock, data)
 
-    def __on_response(self, sock, data: dict):
+    def __on_response(self, _sock, data: dict):
         logging.info(f'response => {data}')
 
     def __on_method(self, sock, data: dict):

stratums/kawpow.py

@@ -5,16 +5,13 @@
 
 class StratumKawpow(Stratum):
 
-    def __init__(self):
-        super(StratumKawpow, self).__init__()
-
     def on_message(self, sock, data: dict):
         if 'method' in data:
             self.__on_method(sock, data)
         else:
             self.__on_response(sock, data)
 
-    def __on_response(self, sock, data: dict):
+    def __on_response(self, _sock, data: dict):
         logging.info(f'response => {data}')
 
     def __on_method(self, sock, data: dict):