hanlde missing CLI plugin extras when loading entry_point

Author: avaldebe

pyproject.toml

@@ -36,7 +36,7 @@ dependencies = [
 
 [project.optional-dependencies]
 mqtt = ["paho-mqtt >=2.1.0"]
-influxdb = ["influxdb >=5.2.0"]
+influxdb = ["influxdb >=5.2.0", "pytz >=2020 ; python_version < '3.10'"]
 
 [project.urls]
 Homepage = "https://avaldebe.github.io/PyPMS"
@@ -83,8 +83,9 @@ test = [
     "packaging >=25.0",
     "mock_serial",
     "logot[loguru]>=1.3.0",
-    "coverage >=7.8",
+    { include-group = "coverage" },
 ]
+coverage = ["coverage >=7.8"]
 docs = ["mkdocs>=1.2.3", "mkdocs-material>=9.4", "pymdown-extensions>=9.5"]
 
 [tool.hatch]
@@ -123,6 +124,9 @@ set_env =
     !report: COVERAGE_FILE={env:COVERAGE_FILE:.coverage/coverage.{envname}}
 commands =
     coverage run -m pytest -ra -q
+extras =
+    influxdb
+    mqtt
 dependency_groups =
     test
 
@@ -133,6 +137,8 @@ parallel_show_output = true
 commands =
     coverage combine --keep
     coverage report
+dependency_groups =
+    coverage
 depends =
     py39, py310, py311, py312, py313
 
@@ -160,10 +166,19 @@ dependency_groups =
 
 [tool.pytest.ini_options]
 minversion = "6.0"
-addopts = "--lf -Werror"
+addopts = "--ff"
 pythonpath = ["src"]
 testpaths = ["tests"]
 logot_capturer = "logot.loguru.LoguruCapturer"
+filterwarnings = [
+    # DeprecationWarning from pypms are errors
+    "error::DeprecationWarning:(pms|tests).*:",
+    # Python3.9 urlib3-1.21.1
+    "ignore:Using or importing the ABCs:DeprecationWarning:urllib3.*",
+    # Python3.12 influxdb-5.3.2
+    "ignore:datetime.datetime.utcfromtimestamp:DeprecationWarning:influxdb.*",
+]
+
 
 [tool.coverage.run]
 source = ["pms"]

src/pms/cli.py

@@ -1,14 +1,18 @@
+import re
 import sys
 from datetime import datetime
 from enum import Enum
+from functools import partial
 from pathlib import Path
+from textwrap import dedent
 from typing import Annotated, Optional, Union
 
 if sys.version_info >= (3, 10):
     from importlib import metadata
 else:
     import importlib_metadata as metadata
 
+
 import typer
 from loguru import logger
 
@@ -17,13 +21,50 @@
 
 main = typer.Typer(add_completion=False, no_args_is_help=True)
 
+
+def pluggin_missing_extras(*extras: str, package: str = "pypms"):  # pragma: no cover
+    green = partial(typer.style, fg=typer.colors.GREEN)
+    red = partial(typer.style, fg=typer.colors.RED)
+
+    required = metadata.requires(package)
+    assert required is not None
+    regex = re.compile(rf"extra == '({'|'.join(extras)})'")
+    missing = tuple(
+        dep for dep, _, markers in (req.partition(";") for req in required) if regex.search(markers)
+    )
+    extra = ", ".join(red(extra, bold=True) for extra in extras)
+    package = green(package, bold=True)
+
+    def command(ctx: typer.Context):
+        sub_cmd = ctx.command_path
+        dependencies = " ".join(f"'{red(dep, bold=True)}'" for dep in missing)
+        msg = f"""
+            {green(sub_cmd, bold=True)} require dependencies which are not installed.
+
+            You can install the required dependencies with
+                {green("python3 -m pip install --upgrade")} {package}[{extra}]
+            Or, if you installed {package} with {green("pipx")}
+                {green("pipx inject")} {package} {dependencies}
+            Or, if you installed {package} with {green("uv tool")}
+                {green("uv tool install")} {package}[{extra}]
+        """
+        typer.echo(dedent(msg))
+
+    command.__doc__ = f"needs {', '.join(missing)}"
+    return command
+
+
 """
 Extra cli commands from plugins
 
 additional Typer commands are loaded from plugins (entry points) advertized as `"pypms.extras"`
 """
+ep: metadata.EntryPoint
 for ep in metadata.entry_points(group="pypms.extras"):
-    main.command(name=ep.name)(ep.load())
+    try:
+        main.command(name=ep.name)(ep.load())
+    except ModuleNotFoundError:  # pragma: no cover
+        main.command(name=ep.name)(pluggin_missing_extras(*ep.extras))
 
 
 def version_callback(value: bool):

src/pms/extra/influxdb.py

@@ -2,46 +2,20 @@
 
 import json
 from dataclasses import fields
-from functools import partial
-from textwrap import dedent
 from typing import Annotated, Protocol
 
 import typer
+from influxdb import InfluxDBClient as Client
 
 from pms.core import exit_on_fail
 
 
-def __missing_influxdb() -> str:  # pragma: no cover
-    green = partial(typer.style, fg=typer.colors.GREEN)
-    red = partial(typer.style, fg=typer.colors.GREEN)
-    package = green("pypms", bold=True)
-    extra = module = red("influxdb", bold=True)
-    msg = f"""
-        {green(__name__, bold=True)} provides additional functionality to {package}.
-        This functionality requires the {module} module, which is not installed.
-
-        You can install this additional dependency with
-            {green("python3 -m pip install --upgrade")} {package}[{extra}]
-        Or, if you installed {package} with {green("pipx")}
-            {green("pipx inject")} {package} {module}
-        Or, if you installed {package} with {green("uv tool")}
-            {green("uv tool install")} {package}[{extra}]
-    """
-    return dedent(msg)
-
-
 class PubFunction(Protocol):
     def __call__(self, *, time: int, tags: dict[str, str], data: dict[str, float]) -> None: ...
 
 
 def client_pub(*, host: str, port: int, username: str, password: str, db_name: str) -> PubFunction:
-    try:
-        from influxdb import InfluxDBClient as client
-    except ModuleNotFoundError:  # pragma: no cover
-        typer.echo(__missing_influxdb())
-        raise typer.Abort()
-
-    c = client(host, port, username, password, None)
+    c = Client(host, port, username, password, None)
     if db_name not in {x["name"] for x in c.get_list_database()}:
         c.create_database(db_name)
     c.switch_database(db_name)

src/pms/extra/mqtt.py

@@ -2,46 +2,20 @@
 
 from dataclasses import fields
 from datetime import datetime
-from functools import partial
-from textwrap import dedent
 from typing import Annotated, Callable, NamedTuple
 
 import typer
 from loguru import logger
+from paho.mqtt.client import Client
 
 from pms.core import exit_on_fail
 from pms.sensors.base import ObsData
 
 
-def __missing_mqtt():  # pragma: no cover
-    green = partial(typer.style, fg=typer.colors.GREEN)
-    red = partial(typer.style, fg=typer.colors.GREEN)
-    package = green("pypms", bold=True)
-    extra = module = red("paho-mqtt", bold=True)
-    msg = f"""
-        {green(__name__, bold=True)} provides additional functionality to {package}.
-        This functionality requires the {module} module, which is not installed.
-
-        You can install this additional dependency with
-            {green("python3 -m pip install --upgrade")} {package}[{extra}]
-        Or, if you installed {package} with {green("pipx")}
-            {green("pipx inject")} {package} {module}
-        Or, if you installed {package} with {green("uv tool")}
-            {green("uv tool install")} {package}[{extra}]
-    """
-    return dedent(msg)
-
-
 def client_pub(
     *, topic: str, host: str, port: int, username: str, password: str
 ) -> Callable[[dict[str, int | str]], None]:
-    try:
-        from paho.mqtt import client
-    except ModuleNotFoundError:  # pragma: no cover
-        typer.echo(__missing_mqtt())
-        raise typer.Abort()
-
-    c = client.Client(client_id=topic)
+    c = Client(client_id=topic)
     c.enable_logger(logger)  # type:ignore[arg-type]
     if username:
         c.username_pw_set(username, password)
@@ -50,7 +24,7 @@ def client_pub(
         f"{topic}/$online", "true", 1, True
     )
     c.will_set(f"{topic}/$online", "false", 1, True)
-    c.connect(host, port, 60)
+    c.connect(host, port)
     c.loop_start()
 
     def pub(data: dict[str, int | str]) -> None:
@@ -106,28 +80,22 @@ def client_sub(
     *,
     on_sensordata: Callable[[Data], None],
 ) -> None:
-    def on_message(client, userdata, msg):
+    def on_message(client: Client, userdata, msg):
         try:
             data = Data.decode(msg.topic, msg.payload)
         except UserWarning as e:
             logger.debug(e)
         else:
             on_sensordata(data)
 
-    try:
-        from paho.mqtt import client
-    except ModuleNotFoundError:  # pragma: no cover
-        typer.echo(__missing_mqtt())
-        raise typer.Abort()
-
-    c = client.Client(client_id=topic)
+    c = Client(client_id=topic)
     c.enable_logger(logger)  # type:ignore[arg-type]
     if username:
         c.username_pw_set(username, password)
 
     c.on_connect = lambda client, userdata, flags, rc: client.subscribe(topic)
     c.on_message = on_message
-    c.connect(host, port, 60)
+    c.connect(host, port)
     c.loop_forever()