Add support for jmp pkg show command

Author: kirkbrauer

packages/jumpstarter-cli-pkg/jumpstarter_cli_pkg/__init__.py

@@ -2,6 +2,7 @@
 from jumpstarter_cli_common import AliasedGroup
 
 from .commands.list import list
+from .commands.show import show
 
 
 @click.group(cls=AliasedGroup)
@@ -10,6 +11,7 @@ def pkg():
 
 
 pkg.add_command(list)
+pkg.add_command(show)
 
 if __name__ == "__main__":
     pkg()

packages/jumpstarter-cli-pkg/jumpstarter_cli_pkg/commands/show.py

@@ -0,0 +1,36 @@
+import asyncclick as click
+from jumpstarter_cli_common import OutputType, opt_output_all
+from jumpstarter_cli_common.exceptions import handle_exceptions
+from jumpstarter_cli_pkg.repository import LocalDriverRepository
+
+
+@click.command("show")
+@click.argument("package")
+@click.option("--drivers", is_flag=True, help="Print drivers only.")
+@click.option("--driver-clients", is_flag=True, help="Print driver clients only.")
+@click.option("--adapters", is_flag=True, help="Print adapters only.")
+@opt_output_all
+@handle_exceptions
+def show(package: str, drivers: bool, driver_clients: bool, adapters: bool, output: OutputType):
+    local_repo = LocalDriverRepository.from_venv()
+    local_package = local_repo.get_package(package)
+    click.echo("Name: " + local_package.name)
+    click.echo("Version: " + local_package.version)
+    click.echo("Summary: " + (local_package.summary if local_package.summary else ""))
+    click.echo("Categories: " + ", ".join(local_package.categories))
+    click.echo("License: " + (local_package.license if local_package.license else ""))
+    click.echo("Drivers:")
+    for driver in local_package.drivers.items:
+        click.echo(f"  {driver.name}:")
+        click.echo(f"    Name: {driver.name}")
+        click.echo(f"    Type: {driver.type}")
+    click.echo("Driver Clients:")
+    for driver_client in local_package.driver_clients.items:
+        click.echo(f"  {driver_client.name}:")
+        click.echo(f"    Name: {driver_client.name}")
+        click.echo(f"    Type: {driver_client.type}")
+    click.echo("Adapters:")
+    for adapter in local_package.adapters.items:
+        click.echo(f"  {driver_client.name}:")
+        click.echo(f"    Name: {adapter.name}")
+        click.echo(f"    Type: {adapter.type}")

packages/jumpstarter-cli-pkg/jumpstarter_cli_pkg/repository/base.py

@@ -1,18 +1,26 @@
 from abc import ABC, abstractmethod
 
 from .package import (
+    V1Alpha1DriverPackage,
     V1Alpha1DriverPackageList,
 )
 
 
 class DriverRepository(ABC):
     """
-    A repository of driver packages.
+    A repository of Jumpstarter plugin packages.
     """
 
     @abstractmethod
     def list_packages(self) -> V1Alpha1DriverPackageList:
         """
-        List all available driver packages.
+        List all available packages.
+        """
+        pass
+
+    @abstractmethod
+    def get_package(self, name: str) -> V1Alpha1DriverPackage:
+        """
+        Get a package by name.
         """
         pass

packages/jumpstarter-cli-pkg/jumpstarter_cli_pkg/repository/entry_points.py

@@ -0,0 +1,4 @@
+class EntryPointGroups:
+    DRIVER_ENTRY_POINT_GROUP = "jumpstarter.drivers"
+    DRIVER_CLIENT_ENTRY_POINT_GROUP = "jumpstarter.clients"
+    ADAPTER_ENTRY_POINT_GROUP = "jumpstarter.adapters"

packages/jumpstarter-cli-pkg/jumpstarter_cli_pkg/repository/local.py

@@ -1,77 +1,52 @@
-from importlib.metadata import entry_points
+from importlib.metadata import Distribution, PackageNotFoundError, distribution, distributions
 
 from .base import DriverRepository
+from .entry_points import EntryPointGroups
 from .package import (
-    V1Alpha1AdapterEntryPoint,
-    V1Alpha1DriverClientEntryPoint,
-    V1Alpha1DriverEntryPoint,
     V1Alpha1DriverPackage,
     V1Alpha1DriverPackageList,
 )
+from jumpstarter.common.exceptions import JumpstarterException
 
 
 class LocalDriverRepository(DriverRepository):
     """
     A local repository of driver packages from the current venv.
     """
 
-    DRIVER_ENTRY_POINT_GROUP = "jumpstarter.drivers"
-    DRIVER_CLIENT_ENTRY_POINT_GROUP = "jumpstarter.clients"
-    ADAPTER_ENTRY_POINT_GROUP = "jumpstarter.adapters"
-
     @staticmethod
     def from_venv():
         """
         Create a `LocalDriverRepository` from the current venv.
         """
         return LocalDriverRepository()
 
-    def _get_driver_packages_from_entry_points(self) -> list[V1Alpha1DriverPackage]:
-        # Create a dict of driver packages to collect entry points
-        driver_packages: dict[str, V1Alpha1DriverPackage] = {}
-
-        # Closure to process entry points for a specific entry point group
-        def _process_entry_points(group: str):
-            """Process entry points for a specific group and add them to driver_packages."""
-            for entry_point in list(entry_points(group=group)):
-                package_id = f"{entry_point.dist.name}=={entry_point.dist.version}"
-                # Check if the package is in the driver packages
-                if package_id not in driver_packages:
-                    # Create a new package
-                    if entry_point.dist is not None:
-                        # Create the package from the entry point distribution metadata
-                        driver_packages[package_id] = V1Alpha1DriverPackage.from_distribution(entry_point.dist)
-                    else:
-                        # Skip this entry point if the distribution metadata is not available
-                        continue
-                # Add the driver/client to the package
-                match group:
-                    case LocalDriverRepository.DRIVER_ENTRY_POINT_GROUP:
-                        driver_packages[package_id].drivers.items.append(
-                            V1Alpha1DriverEntryPoint.from_entry_point(entry_point)
-                        )
-                    case LocalDriverRepository.DRIVER_CLIENT_ENTRY_POINT_GROUP:
-                        driver_packages[package_id].driver_clients.items.append(
-                            V1Alpha1DriverClientEntryPoint.from_entry_point(entry_point)
-                        )
-                    case LocalDriverRepository.ADAPTER_ENTRY_POINT_GROUP:
-                        driver_packages[package_id].adapters.items.append(
-                            V1Alpha1AdapterEntryPoint.from_entry_point(entry_point)
-                        )
-
-        # Process driver entry points
-        _process_entry_points(LocalDriverRepository.DRIVER_ENTRY_POINT_GROUP)
-
-        # Process client entry points
-        _process_entry_points(LocalDriverRepository.DRIVER_CLIENT_ENTRY_POINT_GROUP)
-
-        # Process adapter entry points
-        _process_entry_points(LocalDriverRepository.DRIVER_CLIENT_ENTRY_POINT_GROUP)
-
-        # Return the assembled driver packages list
-        return list(driver_packages.values())
+    def _is_jumpstarter_package(self, dist: Distribution) -> bool:
+        for entry_point in list(dist.entry_points):
+            match entry_point.group:
+                case (
+                    EntryPointGroups.DRIVER_ENTRY_POINT_GROUP
+                    | EntryPointGroups.DRIVER_CLIENT_ENTRY_POINT_GROUP
+                    | EntryPointGroups.ADAPTER_ENTRY_POINT_GROUP
+                ):
+                    return True
+                case _:
+                    return False
+        return False
 
     def list_packages(self) -> V1Alpha1DriverPackageList:
         # Get the local drivers using the Jumpstarter drivers entry point
-        driver_packages = self._get_driver_packages_from_entry_points()
+        driver_packages = []
+        # Iterate through the local package distributions
+        for dist in distributions():
+            # Check if the distribution is a Jumpstarter package
+            if self._is_jumpstarter_package(dist):
+                driver_packages.append(V1Alpha1DriverPackage.from_distribution(dist))
         return V1Alpha1DriverPackageList(items=driver_packages)
+
+    def get_package(self, name: str) -> V1Alpha1DriverPackage:
+        try:
+            # Convert the distribution to a driver package object
+            return V1Alpha1DriverPackage.from_distribution(distribution(name))
+        except PackageNotFoundError as e:
+            raise JumpstarterException(f"Package '{name}' metadata could not be found") from e

packages/jumpstarter-cli-pkg/jumpstarter_cli_pkg/repository/package.py

@@ -3,6 +3,8 @@
 
 from pydantic import Field
 
+from .entry_points import EntryPointGroups
+from jumpstarter.common.exceptions import JumpstarterException
 from jumpstarter.models import JsonBaseModel, ListBaseModel
 
 
@@ -57,7 +59,7 @@ def from_entry_point(ep: EntryPoint):
         return V1Alpha1DriverEntryPoint(name=ep.name, type=ep.value.replace(":", "."), package=ep.dist.name)
 
 
-class V1Alpha1AdapterEntryPointList(ListBaseModel[V1Alpha1DriverEntryPoint]):
+class V1Alpha1AdapterEntryPointList(ListBaseModel[V1Alpha1AdapterEntryPoint]):
     """
     A list of Jumpstarter adapter list models.
     """
@@ -73,7 +75,7 @@ class V1Alpha1DriverEntryPointList(ListBaseModel[V1Alpha1DriverEntryPoint]):
     kind: Literal["DriverEntryPointList"] = Field(default="DriverEntryPointList")
 
 
-class V1Alpha1DriverClientEntryPointList(ListBaseModel[V1Alpha1DriverEntryPoint]):
+class V1Alpha1DriverClientEntryPointList(ListBaseModel[V1Alpha1DriverClientEntryPoint]):
     """
     A list of Jumpstarter driver client classes.
     """
@@ -107,26 +109,39 @@ def requires_dist_to_categories(name: str, requires_dist: list[str]) -> list[str
         Convert the `Requires-Dist` metadata to Jumpstarter driver categories.
         """
         categories = []
-        # Check the package name
-        match name:
-            case "jumpstarter-driver-composite":
-                categories.append("composite")
-            case "jumpstarter-driver-network":
-                categories.append("network")
-            case "jumpstarter-driver-opendal":
-                categories.append("storage")
-            case "jumpstarter-driver-power":
-                categories.append("power")
+
+        # Define constants for root driver package names
+        DRIVER_COMPOSITE = "jumpstarter-driver-composite"
+        DRIVER_NETWORK = "jumpstarter-driver-network"
+        DRIVER_OPENDAL = "jumpstarter-driver-storage"
+        DRIVER_POWER = "jumpstarter-driver-power"
+
+        # Define constants for category names
+        CATEGORY_COMPOSITE = "composite"
+        CATEGORY_NETWORK = "network"
+        CATEGORY_STORAGE = "storage"
+        CATEGORY_POWER = "power"
+
+        # Check package name
+        if name == DRIVER_COMPOSITE:
+            categories.append(CATEGORY_COMPOSITE)
+        elif name == DRIVER_NETWORK:
+            categories.append(CATEGORY_NETWORK)
+        elif name == DRIVER_OPENDAL:
+            categories.append(CATEGORY_STORAGE)
+        elif name == DRIVER_POWER:
+            categories.append(CATEGORY_POWER)
+
         # Check package dependencies
         for dist in requires_dist:
-            if "jumpstarter-driver-composite" in dist and "composite" not in categories:
-                categories.append("composite")
-            elif "jumpstarter-driver-network" in dist and "network" not in categories:
-                categories.append("network")
-            elif "jumpstarter-driver-opendal" in dist and "storage" not in categories:
-                categories.append("storage")
-            elif "jumpstarter-driver-power" in dist and "power" not in categories:
-                categories.append("power")
+            if DRIVER_COMPOSITE in dist and CATEGORY_COMPOSITE not in categories:
+                categories.append(CATEGORY_COMPOSITE)
+            elif DRIVER_NETWORK in dist and CATEGORY_NETWORK not in categories:
+                categories.append(CATEGORY_NETWORK)
+            elif DRIVER_OPENDAL in dist and CATEGORY_STORAGE not in categories:
+                categories.append(CATEGORY_STORAGE)
+            elif DRIVER_POWER in dist and CATEGORY_POWER not in categories:
+                categories.append(CATEGORY_POWER)
 
         return categories
 
@@ -135,6 +150,23 @@ def from_distribution(dist: Distribution):
         """
         Create a `DriverPackage` from an `importlib.metadata.EntryPoint`.
         """
+        # Collect entry points for each type
+        drivers = []
+        driver_clients = []
+        adapters = []
+        # Iterate through the entry points
+        for ep in list(dist.entry_points):
+            match ep.group:
+                case EntryPointGroups.DRIVER_ENTRY_POINT_GROUP:
+                    drivers.append(V1Alpha1DriverEntryPoint.from_entry_point(ep))
+                case EntryPointGroups.DRIVER_CLIENT_ENTRY_POINT_GROUP:
+                    driver_clients.append(V1Alpha1DriverClientEntryPoint.from_entry_point(ep))
+                case EntryPointGroups.ADAPTER_ENTRY_POINT_GROUP:
+                    adapters.append(V1Alpha1AdapterEntryPoint.from_entry_point(ep))
+        #  Check if any entry points were found
+        if len(drivers) + len(driver_clients) + len(adapters) == 0:
+            raise JumpstarterException(f"No valid Jumpstarter entry points found for package '{dist.name}'")
+        # Return the completed driver package
         return V1Alpha1DriverPackage(
             name=dist.name,
             categories=V1Alpha1DriverPackage.requires_dist_to_categories(
@@ -143,6 +175,9 @@ def from_distribution(dist: Distribution):
             version=dist.version,
             summary=dist.metadata.get("Summary"),
             license=dist.metadata.get("License"),
+            drivers=V1Alpha1DriverEntryPointList(items=drivers),
+            driver_clients=V1Alpha1DriverClientEntryPointList(items=driver_clients),
+            adapters=V1Alpha1AdapterEntryPointList(items=adapters),
         )
 
     def list_drivers(self) -> V1Alpha1DriverEntryPointList: