add the full implementation

Signed-off-by: zethson <[email protected]>

.flake8

@@ -1,7 +1,7 @@
 [flake8]
-select = B,B9,C,D,DAR,E,F,N,RST,S,W
-ignore = E203,E501,RST201,RST203,RST301,W503,D100
+select = B,B9,C,D,DAR,E,F,N,RST,W
+ignore = E203,E501,RST201,RST203,RST301,W503,D100,B950
 max-line-length = 120
-max-complexity = 10
+max-complexity = 15
 docstring-convention = google
 per-file-ignores = tests/*:S101

README.rst

@@ -35,7 +35,8 @@ pypi-latest
 Features
 --------
 
-* TODO
+* Check whether the locally installed version of a Python package is the most recent version on PyPI
+* Prompt to update to the latest version if required
 
 
 Installation
@@ -51,7 +52,7 @@ You can install *pypi-latest* via pip_ from PyPI_:
 Usage
 -----
 
-Please see the `Command-line Reference <Usage_>`_ for details.
+Please see the `Usage Reference <Usage_>`_ for details.
 
 
 Credits

docs/usage.rst

@@ -1,6 +1,11 @@
 Usage
 =====
 
-.. click:: pypi_latest.__main__:main
-   :prog: pypi-latest
-   :nested: full
+Import the PypiLatest class as follows:
+
+.. code:: python
+
+   from pypi_latest import PypiLatest
+
+.. automodule:: pypi_latest
+   :members:

pypi_latest/__init__.py

@@ -3,3 +3,111 @@
 __author__ = "Lukas Heumos"
 __email__ = "[email protected]"
 __version__ = "0.1.0"
+
+import json
+import logging
+import sys
+import urllib.request
+from logging import Logger
+from subprocess import PIPE, Popen, check_call
+from urllib.error import HTTPError, URLError
+
+from pkg_resources import parse_version
+from rich import print
+
+from pypi_latest.questionary import custom_questionary
+
+log: Logger = logging.getLogger(__name__)
+
+
+class PypiLatest:
+    """Responsible for checking for newer versions and upgrading it if required."""
+
+    def __init__(self, package_name: str, latest_local_version: str):
+        """Constructor for PypiLatest."""
+        self.package_name = package_name
+        self.latest_local_version = latest_local_version
+
+    def check_upgrade(self) -> None:
+        """Checks whether the locally installed version of the package is the latest.
+
+        If not it prompts whether to upgrade and runs the upgrade command if desired.
+        """
+        if not PypiLatest.check_latest(self):
+            if custom_questionary(function="confirm", question="Do you want to upgrade?", default="y"):
+                PypiLatest.upgrade(self)
+
+    def check_latest(self) -> bool:
+        """Checks whether the locally installed version of the package is the latest available on PyPi.
+
+        Returns:
+            True if locally version is the latest or PyPI is inaccessible, False otherwise
+        """
+        sliced_local_version = (
+            self.latest_local_version[:-9]
+            if self.latest_local_version.endswith("-SNAPSHOT")
+            else self.latest_local_version
+        )
+        log.debug(f"Latest local {self.package_name} version is: {self.latest_local_version}.")
+        log.debug(f"Checking whether a new {self.package_name} version exists on PyPI.")
+        try:
+            # Retrieve info on latest version
+            # Adding nosec (bandit) here, since we have a hardcoded https request
+            # It is impossible to access file:// or ftp://
+            # See: https://stackoverflow.com/questions/48779202/audit-url-open-for-permitted-schemes-allowing-use-of-file-or-custom-schemes
+            req = urllib.request.Request(f"https://pypi.org/pypi/{self.package_name}/json")  # nosec
+            with urllib.request.urlopen(req, timeout=1) as response:  # nosec
+                contents = response.read()
+                data = json.loads(contents)
+                latest_pypi_version = data["info"]["version"]
+        except (HTTPError, TimeoutError, URLError):
+            print(
+                f"[bold red]Unable to contact PyPI to check for the latest {self.package_name} version. "
+                "Do you have an internet connection?"
+            )
+            # Returning true by default, since this is not a serious issue
+            return True
+
+        if parse_version(sliced_local_version) > parse_version(latest_pypi_version):
+            print(
+                f"[bold yellow]Installed version {self.latest_local_version} of {self.package_name} is newer than the latest release {latest_pypi_version}!"
+                f" You are running a nightly version and features may break!"
+            )
+        elif parse_version(sliced_local_version) == parse_version(latest_pypi_version):
+            return True
+        else:
+            print(
+                f"[bold red]Installed version {self.latest_local_version} of {self.package_name} is outdated. Newest version is {latest_pypi_version}!"
+            )
+            return False
+
+        return False
+
+    def upgrade(self) -> None:
+        """Calls pip as a subprocess with the --upgrade flag to upgrade the package to the latest version."""
+        log.debug(f"Attempting to upgrade {self.package_name} via   pip install --upgrade {self.package_name}   .")
+        if not PypiLatest.is_pip_accessible():
+            sys.exit(1)
+        try:
+            check_call([sys.executable, "-m", "pip", "install", "--upgrade", self.package_name])
+        except Exception as e:
+            print(f"[bold red]Unable to upgrade {self.package_name}")
+            print(f"[bold red]Exception: {e}")
+
+    @classmethod
+    def is_pip_accessible(cls) -> bool:
+        """Verifies that pip is accessible and in the PATH.
+
+        Returns:
+            True if accessible, False if not
+        """
+        log.debug("Verifying that pip is accessible.")
+        pip_installed = Popen(["pip", "--version"], stdout=PIPE, stderr=PIPE, universal_newlines=True)
+        (git_installed_stdout, git_installed_stderr) = pip_installed.communicate()
+        if pip_installed.returncode != 0:
+            log.debug("Pip was not accessible! Attempted to test via   pip --version   .")
+            print("[bold red]Unable to find 'pip' in the PATH. Is it installed?")
+            print("[bold red]Run command was [green]'pip --version '")
+            return False
+
+        return True

pypi_latest/questionary.py

@@ -0,0 +1,98 @@
+import logging
+import os
+import sys
+from logging import Logger
+from typing import List, Optional, Union
+
+import questionary
+from prompt_toolkit.styles import Style  # type: ignore
+from rich.console import Console
+
+
+def force_terminal_in_github_action() -> Console:
+    """Check, whether the GITHUB_ACTIONS environment variable is set or not.
+
+    If it is set, the process runs in a workflow file and we need to tell rich, in order to get colored output as well.
+
+    Returns:
+        Rich Console object
+    """
+    if "GITHUB_ACTIONS" in os.environ:
+        return Console(file=sys.stderr, force_terminal=True)
+    else:
+        return Console(file=sys.stderr)
+
+
+log: Logger = logging.getLogger(__name__)
+
+ehrapy_style = Style(
+    [
+        ("qmark", "fg:#0000FF bold"),  # token in front of the question
+        ("question", "bold"),  # question text
+        ("answer", "fg:#008000 bold"),  # submitted answer text behind the question
+        ("pointer", "fg:#0000FF bold"),  # pointer used in select and checkbox prompts
+        ("highlighted", "fg:#0000FF bold"),  # pointed-at choice in select and checkbox prompts
+        ("selected", "fg:#008000"),  # style for a selected item of a checkbox
+        ("separator", "fg:#cc5454"),  # separator in lists
+        ("instruction", ""),  # user instructions for select, rawselect, checkbox
+        ("text", ""),  # plain text
+        ("disabled", "fg:#FF0000 italic"),  # disabled choices for select and checkbox prompts
+    ]
+)
+
+# the console used for printing with rich
+console = force_terminal_in_github_action()
+
+
+def custom_questionary(
+    function: str,
+    question: str,
+    choices: Optional[List[str]] = None,
+    default: Optional[str] = None,
+) -> Union[str, bool]:
+    """Custom selection based on Questionary. Handles keyboard interrupts and default values.
+
+    Args:
+        function: The function of questionary to call (e.g. select or text).
+                  See https://github.com/tmbo/questionary for all available functions.
+        question: List of all possible choices.
+        choices: The question to prompt for. Should not include default values or colons.
+        default: A set default value, which will be chosen if the user does not enter anything.
+
+    Returns:
+        The chosen answer.
+    """
+    answer: Optional[str] = ""
+    try:
+        if function == "select":
+            if default not in choices:  # type: ignore
+                log.debug(f"Default value {default} is not in the set of choices!")
+            answer = getattr(questionary, function)(f"{question}: ", choices=choices, style=ehrapy_style).unsafe_ask()
+        elif function == "password":
+            while not answer or answer == "":
+                answer = getattr(questionary, function)(f"{question}: ", style=ehrapy_style).unsafe_ask()
+        elif function == "text":
+            if not default:
+                log.debug(
+                    "Tried to utilize default value in questionary prompt, but is None! Please set a default value."
+                )
+                default = ""
+            answer = getattr(questionary, function)(f"{question} [{default}]: ", style=ehrapy_style).unsafe_ask()
+        elif function == "confirm":
+            default_value_bool = True if default == "Yes" or default == "yes" else False
+            answer = getattr(questionary, function)(
+                f"{question} [{default}]: ", style=ehrapy_style, default=default_value_bool
+            ).unsafe_ask()
+        else:
+            log.debug(f"Unsupported questionary function {function} used!")
+
+    except KeyboardInterrupt:
+        console.print("[bold red] Aborted!")
+        sys.exit(1)
+    if answer is None or answer == "":
+        answer = default
+
+    log.debug(f"User was asked the question: ||{question}|| as: {function}")
+    log.debug(f"User selected {answer}")
+
+    return answer  # type: ignore

pyproject.toml

@@ -20,10 +20,11 @@ classifiers = [
 
 
 [tool.poetry.dependencies]
-python = "^3.6.1"
+python = ">=3.6.1,<3.10"
 click = "^8.0.0"
 rich = "^10.4.0"
 PyYAML = "^5.4.1"
+questionary = "^1.9.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^6.2.3"