Merge pull request #66 from zurk/master

Refactor server.py

Author: vmarkovtsev

lookout/core/__init__.py

@@ -1,2 +1,2 @@
 """Lookout SDK for stateful analyzers - root package."""
-__version__ = "0.14.0"
+__version__ = "0.15.0"

lookout/core/test_helpers/__init__.py lookout/core/helpers/__init__.py

@@ -0,0 +1,97 @@
+from argparse import Namespace
+import subprocess
+from typing import Optional, Type
+
+from lookout.core.analyzer import Analyzer
+from lookout.core.cmdline import create_model_repo_from_args
+from lookout.core.data_requests import DataService
+from lookout.core.event_listener import EventListener
+from lookout.core.helpers.server import check_port_free, find_port, LookoutSDK
+from lookout.core.manager import AnalyzerManager
+
+
+class AnalyzerContextManager:
+    """Context manager for launching analyzer."""
+
+    def __init__(self, analyzer: Type[Analyzer], db: str, fs: str,
+                 init: bool = True, data_request_address: str = "localhost:10301"):
+        """
+        Initialization.
+
+        :param db: path to an SQLite database with model metadata.
+        :param fs: location where to store the trained model.
+        :param analyzer: analyzer class to use.
+        :param init: Value indicating whether to run the destructive database initialization \
+                     or not. If you want to reuse an existing database set False.
+        :param data_request_address: DataService GRPC endpoint to use.
+        """
+        self.analyzer = analyzer
+        self.init = init
+        self._port = find_port()
+        self.data_request_address = data_request_address
+        self._sql_alchemy_model_args = Namespace(
+            db="sqlite:///%s" % db,
+            fs=fs,
+            cache_size="1G",
+            cache_ttl="6h",
+            db_kwargs={},
+        )
+        self._lookout_sdk = None
+
+    def __enter__(self) -> "AnalyzerContextManager":
+        """
+        Create the context and run the events listener.
+        """
+        self.model_repository = create_model_repo_from_args(self._sql_alchemy_model_args)
+        if self.init:
+            self.model_repository.init()
+        self.data_service = DataService(self.data_request_address)
+        self.manager = AnalyzerManager(analyzers=[self.analyzer],
+                                       model_repository=self.model_repository,
+                                       data_service=self.data_service)
+        if not check_port_free(self._port):
+            self._port = find_port()
+        self.listener = EventListener(address="0.0.0.0:%d" % self._port, handlers=self.manager,
+                                      n_workers=1)
+        self.listener.start()
+        self._lookout_sdk = LookoutSDK()
+        return self
+
+    def __exit__(self, exc_type=None, exc_val=None, exc_tb=None):
+        """
+        Stop the events listener and shutdown the context.
+        """
+        self._lookout_sdk = None
+        self.listener.stop()
+        self.model_repository.shutdown()
+        self.data_service.shutdown()
+
+    def review(self, fr: str, to: str, *, git_dir: str, bblfsh: Optional[str]=None,
+               log_level: Optional[str]=None, config_json: Optional[dict]=None) \
+            -> subprocess.CompletedProcess:
+        """
+        Proxy for LookoutSDK.review().
+
+        Triggers a review event and effectively calls the underlying analyzer's `analyze()`.
+        Read parameters description in `LookoutSDK.review()`
+        """
+        if not self._lookout_sdk:
+            raise AttributeError(
+                "AnalyzerContextManager.review() is available only inside `with`")
+        return self._lookout_sdk.review(fr, to, self._port, git_dir=git_dir, bblfsh=bblfsh,
+                                        log_level=log_level, config_json=config_json)
+
+    def push(self, fr: str, to: str, *, git_dir: str, bblfsh: Optional[str]=None,
+             log_level: Optional[str]=None, config_json: Optional[dict]=None) \
+            -> subprocess.CompletedProcess:
+        """
+        Proxy for LookoutSDK.push().
+
+        Triggers a push event and effectively calls the underlying analyzer's `train()`.
+        Read parameters description in `LookoutSDK.push()`
+        """
+        if not self._lookout_sdk:
+            raise AttributeError(
+                "AnalyzerContextManager.push() is available only inside `with` statement")
+        return self._lookout_sdk.push(fr, to, self._port, git_dir=git_dir, bblfsh=bblfsh,
+                                      log_level=log_level, config_json=config_json)

lookout/core/helpers/analyzer_context_manager.py

@@ -0,0 +1,167 @@
+"""Utils to work with lookout-sdk binary."""
+import io
+import json
+import logging
+import os
+import pathlib
+import random
+from shutil import copyfileobj
+import socket
+import subprocess
+import sys
+import tarfile
+import tempfile
+from typing import Optional
+from urllib.error import HTTPError
+from urllib.request import urlopen
+
+from lookout.core.api.version import __version__ as binver
+
+
+class LookoutSDK:
+    """
+    Wrapper class for `lookout-sdk` executable.
+
+    Allows you to query analyzers the same way lookout server do.
+    About lookout-sdk read https://github.com/src-d/lookout-sdk
+    """
+
+    _log = logging.getLogger("LookoutSDK")
+
+    def __init__(self):
+        """
+        Fetch lookout-sdk executable if it is missing.
+        """
+        self._version = binver
+        self._exefile = (pathlib.Path(tempfile.gettempdir()) /
+                         "lookout-sdk-ml" / ("lookout-sdk-%s" % self._version))
+        if not self._exefile.exists():
+            self.fetch()
+
+    version = property(lambda self: self._version)
+
+    def fetch(self):
+        """
+        Download the lookout-sdk executable from GitHub Releases.
+        """
+        self._exefile.parent.mkdir(exist_ok=True)
+        platform = sys.platform
+        try:
+            buffer = io.BytesIO()
+            with urlopen("https://github.com/src-d/lookout/releases/download/"
+                         "%s/lookout-sdk_%s_%s_amd64.tar.gz" % (binver, binver, platform),
+                         ) as response:
+                copyfileobj(response, buffer)
+            buffer.seek(0)
+            with tarfile.open(fileobj=buffer, mode="r:gz") as tar, \
+                    self._exefile.open("wb") as fout:
+                copyfileobj(tar.extractfile("lookout-sdk_%s_amd64/lookout-sdk" % platform), fout)
+            os.chmod(str(self._exefile), 0o775)
+        except HTTPError as e:
+            if e.code == 404:
+                self._log.error("Release %s for %s platform is missing." % (binver, platform))
+            raise e from None
+        except Exception as e:
+            if self._exefile.exists():
+                os.remove(str(self._exefile))
+            raise e from None
+
+    def push(self, fr: str, to: str, port: int, *, git_dir: str, bblfsh: Optional[str]=None,
+             log_level: Optional[str]=None, config_json: Optional[dict]=None) \
+            -> subprocess.CompletedProcess:
+        """
+        Provide a simple data server and triggers an analyzer push event.
+
+        :param fr: Corresponds to --from flag.
+        :param to: Corresponds to --to flag.
+        :param port: Running analyzer port on localhost.
+        :param git_dir: Corresponds to --git-dir flag.
+        :param log_level: Corresponds to --log-level flag.
+        :param bblfsh: Corresponds to --bblfshd flag.
+        :param config_json: Corresponds to --config-json flag.
+        :return: CompletedProcess with return code.
+        """
+        return self._run("push", fr, to, port, git_dir, bblfsh, log_level, config_json)
+
+    def review(self, fr: str, to: str, port: int, *, git_dir: str, bblfsh: Optional[str]=None,
+               log_level: Optional[str]=None, config_json: Optional[dict]=None) \
+            -> subprocess.CompletedProcess:
+        """
+        Provide a simple data server and triggers an analyzer review event.
+
+        :param fr: Corresponds to --from flag.
+        :param to: Corresponds to --to flag.
+        :param port: Running analyzer port on localhost.
+        :param git_dir: Corresponds to --git-dir flag.
+        :param log_level: Corresponds to --log-level flag.
+        :param bblfsh: Corresponds to --bblfshd flag.
+        :param config_json: Corresponds to --config-json flag.
+        :return: CompletedProcess with return code.
+        """
+        return self._run("review", fr, to, port, git_dir, bblfsh, log_level, config_json)
+
+    def _run(self, cmd: str, fr: str, to: str, port: int, git_dir: str, bblfsh: Optional[str],
+             log_level: Optional[str], config_json: Optional[dict]) -> subprocess.CompletedProcess:
+        """
+        Run lookout-sdk executable. If you do not have it please fetch first.
+
+        :param cmd: Sub-command to run.
+        :param fr: Corresponds to --from flag.
+        :param to: Corresponds to --to flag.
+        :param port: Running analyzer port on localhost.
+        :param git_dir: Corresponds to --git-dir flag.
+        :param log_level: Corresponds to --log-level flag.
+        :param bblfsh: Corresponds to --bblfshd flag.
+        :param config_json: Corresponds to --config-json flag.
+        :return: CompletedProcess with return code.
+        """
+        command = [
+            str(self._exefile), cmd, "ipv4://localhost:%d" % port,
+            "--from", fr,
+            "--to", to,
+            "--git-dir", git_dir,
+        ]
+        if log_level:
+            command.extend(("--log-level", log_level))
+        if bblfsh:
+            command.extend(("--bblfshd", "ipv4://" + bblfsh))
+        if config_json:
+            command.extend(("--config-json", json.dumps(config_json)))
+        return subprocess.run(command, stdout=sys.stdout, stderr=sys.stderr, check=True)
+
+
+def check_port_free(port: int) -> bool:
+    """
+    Check if the port is not taken on localhost.
+
+    :param port: Port number.
+    :return: True if available else False.
+    """
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        s.connect(("localhost", port))
+        return False
+    except ConnectionRefusedError:
+        return True
+    finally:
+        try:
+            s.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
+        s.close()
+
+
+def find_port(attempts: int = 100) -> int:
+    """
+    Find a free port on localhost.
+
+    :param attempts: Number of random search attempts.
+    :return: Found free port number.
+    """
+    while True:
+        attempts -= 1
+        if attempts == 0:
+            raise ConnectionError("cannot find an open port")
+        port = random.randint(1024, 32768)
+        if check_port_free(port):
+            return port

lookout/core/helpers/server.py

@@ -1,4 +0,0 @@
-from lookout.core.test_helpers import server
-
-if not server.exefile.exists():
-    server.fetch()