Cast

CHANGES/pulp-glue/+cast.feature

@@ -0,0 +1 @@
+Added `detail_context` to master-detail contexts.

CHANGES/pulp-glue/+load_plugins.feature

@@ -0,0 +1 @@
+Added `load_plugins` to `pulp_glue.common` so plugins providing a "pulp_glue.plugins" entrypoint can be enumerated and loaded.

pulp-glue/docs/dev/reference/common.md

@@ -0,0 +1,3 @@
+# pulp_glue.common
+
+::: pulp_glue.common

pulp-glue/pulp_glue/common/__init__.py

@@ -1 +1,32 @@
+import sys
+import typing as t
+
+if sys.version_info >= (3, 10):
+    from importlib.metadata import entry_points
+else:
+    from importlib_metadata import entry_points
+
 __version__ = "0.30.0.dev"
+
+# Keep track to prevent loading plugins twice
+loaded_plugins: t.Set[str] = set()
+
+
+def load_plugins(enabled_plugins: t.Optional[t.List[str]] = None) -> None:
+    """
+    Load glue plugins that provide a `pulp_glue.plugins` entrypoint.
+    This may be needed when you rely on the `TYPE_REGISTRY` attributes but cannot load the modules
+    explicitely.
+
+    Parameters:
+        enabled_plugins: Optional list of plugins to consider for loading.
+    """
+    for entry_point in entry_points(group="pulp_glue.plugins"):
+        name = entry_point.name
+        if (
+            enabled_plugins is None or entry_point.name in enabled_plugins
+        ) and entry_point.name not in loaded_plugins:
+            plugin = entry_point.load()
+            if hasattr(plugin, "mount"):
+                plugin.mount()
+            loaded_plugins.add(name)

pulp-glue/pulp_glue/common/context.py

@@ -1304,7 +1304,36 @@ def needs_capability(self, capability: str) -> None:
             )
 
 
-class PulpRemoteContext(PulpEntityContext):
+class PulpMasterContext(PulpEntityContext):
+    TYPE_REGISTRY: t.Final[t.ClassVar[t.Dict[str, t.Type["t.Self"]]]]
+
+    def __init_subclass__(cls, **kwargs: t.Any) -> None:
+        super().__init_subclass__(**kwargs)
+        if not hasattr(cls, "RESOURCE_TYPE"):
+            cls.TYPE_REGISTRY = {}
+        elif hasattr(cls, "PLUGIN"):
+            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
+
+    def detail_context(self, pulp_href: str) -> "t.Self":
+        """
+        Provide a detail context for a matching href.
+        """
+        m = re.search(self.HREF_PATTERN, pulp_href)
+        if m is None:
+            raise PulpException(f"'{pulp_href}' is not an href for {self.ENTITY}.")
+        plugin = m.group("plugin")
+        resource_type = m.group("resource_type")
+        try:
+            detail_class = self.TYPE_REGISTRY[f"{plugin}:{resource_type}"]
+        except KeyError:
+            raise PulpException(
+                f"{self.ENTITY} with plugin '{plugin}' and"
+                f"resource type '{resource_type}' is unknown."
+            )
+        return detail_class(self.pulp_ctx, pulp_href=pulp_href)
+
+
+class PulpRemoteContext(PulpMasterContext):
     """
     Base class for remote contexts.
     """
@@ -1330,27 +1359,15 @@ class PulpRemoteContext(PulpEntityContext):
         "sock_read_timeout",
         "rate_limit",
     }
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpRemoteContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
 
-class PulpPublicationContext(PulpEntityContext):
+class PulpPublicationContext(PulpMasterContext):
     """Base class for publication contexts."""
 
     ENTITY = _("publication")
     ENTITIES = _("publications")
     ID_PREFIX = "publications"
     HREF_PATTERN = r"publications/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpPublicationContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def list(self, limit: int, offset: int, parameters: t.Dict[str, t.Any]) -> t.List[t.Any]:
         if parameters.get("repository") is not None:
@@ -1360,20 +1377,14 @@ def list(self, limit: int, offset: int, parameters: t.Dict[str, t.Any]) -> t.Lis
         return super().list(limit, offset, parameters)
 
 
-class PulpDistributionContext(PulpEntityContext):
+class PulpDistributionContext(PulpMasterContext):
     """Base class for distribution contexts."""
 
     ENTITY = _("distribution")
     ENTITIES = _("distributions")
     ID_PREFIX = "distributions"
     HREF_PATTERN = r"distributions/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     NULLABLES = {"content_guard", "publication", "remote", "repository", "repository_version"}
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpDistributionContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
 
 class PulpRepositoryVersionContext(PulpEntityContext):
@@ -1432,7 +1443,7 @@ def repair(self) -> t.Any:
         return self.call("repair", parameters={self.HREF: self.pulp_href}, body={})
 
 
-class PulpRepositoryContext(PulpEntityContext):
+class PulpRepositoryContext(PulpMasterContext):
     """Base class for repository contexts."""
 
     ENTITY = _("repository")
@@ -1441,12 +1452,6 @@ class PulpRepositoryContext(PulpEntityContext):
     ID_PREFIX = "repositories"
     VERSION_CONTEXT: t.ClassVar[t.Type[PulpRepositoryVersionContext]] = PulpRepositoryVersionContext
     NULLABLES = {"description", "retain_repo_versions"}
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpRepositoryContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def get_version_context(
         self,
@@ -1555,19 +1560,14 @@ def reclaim(
         return self.call("reclaim_space_reclaim", body=body)
 
 
-class PulpContentContext(PulpEntityContext):
+class PulpContentContext(PulpMasterContext):
     """Base class for content contexts."""
 
     ENTITY = _("content")
     ENTITIES = _("content")
+    HREF_PATTERN = r"content/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     ID_PREFIX = "content"
     HREF_PATTERN = r"content/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpContentContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def __init__(
         self,
@@ -1669,38 +1669,26 @@ def upload(
         return self.create(body=body)
 
 
-class PulpACSContext(PulpEntityContext):
+class PulpACSContext(PulpMasterContext):
     """Base class for ACS contexts."""
 
     ENTITY = _("ACS")
     ENTITIES = _("ACSes")
     HREF_PATTERN = r"acs/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     ID_PREFIX = "acs"
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpACSContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
     def refresh(self, href: t.Optional[str] = None) -> t.Any:
         return self.call("refresh", parameters={self.HREF: href or self.pulp_href})
 
 
-class PulpContentGuardContext(PulpEntityContext):
+class PulpContentGuardContext(PulpMasterContext):
     """Base class for content guard contexts."""
 
     ENTITY = "content guard"
     ENTITIES = "content guards"
     ID_PREFIX = "contentguards"
     HREF_PATTERN = r"contentguards/(?P<plugin>[\w\-_]+)/(?P<resource_type>[\w\-_]+)/"
     NULLABLES = {"description"}
-    TYPE_REGISTRY: t.Final[t.Dict[str, t.Type["PulpContentGuardContext"]]] = {}
-
-    def __init_subclass__(cls, **kwargs: t.Any) -> None:
-        super().__init_subclass__(**kwargs)
-        if hasattr(cls, "PLUGIN") and hasattr(cls, "RESOURCE_TYPE"):
-            cls.TYPE_REGISTRY[f"{cls.PLUGIN}:{cls.RESOURCE_TYPE}"] = cls
 
 
 EntityFieldDefinition = t.Union[None, str, PulpEntityContext]

pulp-glue/pyproject.toml

@@ -33,6 +33,15 @@ documentation = "https://pulpproject.org/pulp-glue/docs/dev/"
 repository = "https://github.com/pulp/pulp-cli"
 changelog = "https://pulpproject.org/pulp-cli/changes/"
 
+[project.entry-points."pulp_glue.plugins"]
+ansible = "pulp_glue.ansible"
+certguard = "pulp_glue.certguard"
+container = "pulp_glue.container"
+core = "pulp_glue.core"
+file = "pulp_glue.file"
+python = "pulp_glue.python"
+rpm = "pulp_glue.rpm"
+
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["pulp_glue.*"]

pulp-glue/tests/test_entity_context.py

@@ -0,0 +1,35 @@
+import random
+import string
+import typing as t
+
+import pytest
+
+from pulp_glue.common import load_plugins, loaded_plugins
+from pulp_glue.common.context import PulpContext, PulpRepositoryContext
+from pulp_glue.file.context import PulpFileRepositoryContext
+
+pytestmark = pytest.mark.glue
+
+
+@pytest.fixture
+def file_repository(pulp_ctx: PulpContext) -> t.Dict[str, t.Any]:
+    name = "".join(random.choices(string.ascii_letters, k=8))
+    file_repository_ctx = PulpFileRepositoryContext(pulp_ctx)
+    yield file_repository_ctx.create(body={"name": name})
+    file_repository_ctx.delete()
+
+
+def test_plugin_loading() -> None:
+    load_plugins()
+    assert "core" in loaded_plugins
+
+
+def test_type_registry() -> None:
+    assert "file:file" in PulpRepositoryContext.TYPE_REGISTRY
+
+
+def test_detail_context(pulp_ctx: PulpContext, file_repository: t.Dict[str, t.Any]) -> None:
+    master_ctx = PulpRepositoryContext(pulp_ctx)
+    detail_ctx = master_ctx.detail_context(pulp_href=file_repository["pulp_href"])
+    assert isinstance(detail_ctx, PulpFileRepositoryContext)
+    assert detail_ctx.entity["name"] == file_repository["name"]