Introduce ViewSetContext

This context should be used for view sets that are not attached to a
model.

Co-authored-by: Grant Gainey <[email protected]>

CHANGES/+viewset_context.devel

@@ -0,0 +1 @@
+Added `pass_view_set_context` decorator to lookup `PulpViewSetContext` objects.

CHANGES/pulp-glue/+viewset_context.devel

@@ -0,0 +1,2 @@
+Added a `PulpViewSetContext` to represent a view set not attached to a specific type of entity.
+Accordingly, `PulpEntityContext` should only be used when that API is defined by a `NamedModelViewset`.

pulp-glue/pulp_glue/common/context.py

@@ -533,15 +533,76 @@ def needs_plugin(
             self._needed_plugins.append(plugin_requirement)
 
 
-class PulpEntityContext:
+class PulpViewSetContext:
+    """
+    Base class to interact with a generic viewset.
+
+    Parameters:
+        pulp_ctx: The server context to attach this viewset context to.
+    """
+
+    # Subclasses should provide appropriate values here
+    ID_PREFIX: t.ClassVar[str]
+    """Common prefix for the operations of this entity."""
+    NEEDS_PLUGINS: t.ClassVar[t.List[PluginRequirement]] = []
+    """List of plugin requirements to operate such an entity on the server."""
+
+    def __init__(self, pulp_ctx: PulpContext) -> None:
+        self.pulp_ctx: PulpContext = pulp_ctx
+
+        # Add requirements to the lazy evaluated list
+        for plugin_requirement in self.NEEDS_PLUGINS:
+            self.pulp_ctx.needs_plugin(plugin_requirement)
+
+    def call(
+        self,
+        operation: str,
+        non_blocking: bool = False,
+        parameters: t.Optional[t.Dict[str, t.Any]] = None,
+        body: t.Optional[EntityDefinition] = None,
+        validate_body: bool = True,
+    ) -> t.Any:
+        """
+        Perform an API call for operation.
+        Wait for triggered tasks to finish if not background.
+        Returns the operation result, or the finished task.
+
+        Parameters:
+            operation: The operation to be performed on the entity. Usually the openapi
+                operation_id is constructed by concatenating with the `ID_PREFIX`.
+            non_blocking: returns unfinished tasks if `True`.
+            parameters: Arguments that are to be sent as headers, querystrings or part of the URI.
+            body: Body payload for POST, PUT, PATCH calls.
+            validate_body: Indicate whether the body should be validated.
+
+        Returns:
+            The body of the response, or the task or task group if one was issued.
+
+        Raises:
+            PulpNoWait: in case the context has `background_tasks` set or a task (group) timed out.
+        """
+        operation_id: str = (
+            getattr(self, operation.upper() + "_ID", None) or self.ID_PREFIX + "_" + operation
+        )
+        return self.pulp_ctx.call(
+            operation_id,
+            non_blocking=non_blocking,
+            parameters=parameters,
+            body=body,
+            validate_body=validate_body,
+        )
+
+
+class PulpEntityContext(PulpViewSetContext):
     """
     Base class for entity specific contexts.
     This class provides the basic CRUD commands and ties its instances to the global
     PulpContext for api access.
+    It typically corresponds to a NamedModelViewset.
     Mostly specification is achieved by defining / extending the class attributes below.
 
     Parameters:
-        pulp_ctx: The server context to attach this entity to.
+        pulp_ctx: The server context to attach this entity context to.
         pulp_href: Specifying this is equivalent to assinging to `pulp_href` later.
         entity: Specifying this is equivalent to assinging to `entity` later.
     """
@@ -553,12 +614,8 @@ class PulpEntityContext:
     """Translatable plural of `ENTITY`."""
     HREF: t.ClassVar[str]
     """Name of the href parameter in the url patterns."""
-    ID_PREFIX: t.ClassVar[str]
-    """Common prefix for the operations of this entity."""
     NULLABLES: t.ClassVar[t.Set[str]] = set()
     """Set of fields that can be cleared by sending 'null'."""
-    NEEDS_PLUGINS: t.ClassVar[t.List[PluginRequirement]] = []
-    """List of plugin requirements to operate such an entity on the server."""
     CAPABILITIES: t.ClassVar[t.Dict[str, t.List[PluginRequirement]]] = {}
     """
     List of capabilities this entity provides.
@@ -657,56 +714,14 @@ def __init__(
     ) -> None:
         assert pulp_href is None or entity is None
 
+        super().__init__(pulp_ctx)
         self.meta: t.Dict[str, str] = {}
-        self.pulp_ctx: PulpContext = pulp_ctx
-
-        # Add requirements to the lazy evaluated list
-        for plugin_requirement in self.NEEDS_PLUGINS:
-            self.pulp_ctx.needs_plugin(plugin_requirement)
 
         self._entity = None
         self._entity_lookup = entity or {}
         if pulp_href is not None:
             self.pulp_href = pulp_href
 
-    def call(
-        self,
-        operation: str,
-        non_blocking: bool = False,
-        parameters: t.Optional[t.Dict[str, t.Any]] = None,
-        body: t.Optional[EntityDefinition] = None,
-        validate_body: bool = True,
-    ) -> t.Any:
-        """
-        Perform an API call for operation.
-        Wait for triggered tasks to finish if not background.
-        Returns the operation result, or the finished task.
-
-        Parameters:
-            operation: The operation to be performed on the entity. Usually the openapi
-                operation_id is constructed by concatenating with the `ID_PREFIX`.
-            non_blocking: returns unfinished tasks if `True`.
-            parameters: Arguments that are to be sent as headers, querystrings or part of the URI.
-            body: Body payload for POST, PUT, PATCH calls.
-            validate_body: Indicate whether the body should be validated.
-
-        Returns:
-            The body of the response, or the task or task group if one was issued.
-
-        Raises:
-            PulpNoWait: in case the context has `background_tasks` set or a task (group) timed out.
-        """
-        operation_id: str = (
-            getattr(self, operation.upper() + "_ID", None) or self.ID_PREFIX + "_" + operation
-        )
-        return self.pulp_ctx.call(
-            operation_id,
-            non_blocking=non_blocking,
-            parameters=parameters,
-            body=body,
-            validate_body=validate_body,
-        )
-
     @classmethod
     def _preprocess_value(cls, key: str, value: t.Any) -> t.Any:
         if key in cls.NULLABLES and value == "":

pulp-glue/pulp_glue/core/context.py

@@ -11,6 +11,8 @@
     PulpContext,
     PulpEntityContext,
     PulpException,
+    PulpViewSetContext,
+    preprocess_payload,
 )
 from pulp_glue.common.i18n import get_translation
 
@@ -277,16 +279,18 @@ class PulpImporterContext(PulpEntityContext):
     ID_PREFIX = "importers_core_pulp"
 
 
-class PulpOrphanContext(PulpEntityContext):
+class PulpOrphanContext(PulpViewSetContext):
+    ID_PREFIX = "orphans_cleanup"
+
     def cleanup(self, body: t.Optional[t.Dict[str, t.Any]] = None) -> t.Any:
         if body is not None:
-            body = self.preprocess_entity(body)
+            body = preprocess_payload(body)
             if "orphan_protection_time" in body:
                 self.pulp_ctx.needs_plugin(PluginRequirement("core", specifier=">=3.15.0"))
         else:
             body = {}
         if self.pulp_ctx.has_plugin(PluginRequirement("core", specifier=">=3.14.0")):
-            result = self.pulp_ctx.call("orphans_cleanup_cleanup", body=body)
+            result = self.call("cleanup", body=body)
         else:
             if body:
                 self.pulp_ctx.needs_plugin(PluginRequirement("core", specifier=">=3.14.0"))

pulpcore/cli/common/generic.py

@@ -27,6 +27,7 @@
     PulpRemoteContext,
     PulpRepositoryContext,
     PulpRepositoryVersionContext,
+    PulpViewSetContext,
 )
 from pulp_glue.common.i18n import get_translation
 from pulp_glue.common.openapi import AuthProviderBase
@@ -237,6 +238,8 @@ def basic_auth(self) -> t.Optional[t.Union[t.Tuple[str, str], requests.auth.Auth
 
 pass_pulp_context = click.make_pass_decorator(PulpCLIContext)
 """Decorator to make the Pulp context available to a command."""
+pass_view_set_context = click.make_pass_decorator(PulpViewSetContext)
+"""Decorator to make the nearest view set context available to a command."""
 pass_entity_context = click.make_pass_decorator(PulpEntityContext)
 """Decorator to make the nearest entity context available to a command."""
 pass_acs_context = click.make_pass_decorator(PulpACSContext)

pulpcore/cli/core/orphan.py

@@ -1,15 +1,15 @@
 import typing as t
 
 import click
-from pulp_glue.common.context import PluginRequirement, PulpEntityContext
+from pulp_glue.common.context import PluginRequirement, PulpViewSetContext
 from pulp_glue.common.i18n import get_translation
 from pulp_glue.core.context import PulpOrphanContext
 
 from pulpcore.cli.common.generic import (
     PulpCLIContext,
     load_json_callback,
-    pass_entity_context,
     pass_pulp_context,
+    pass_view_set_context,
     pulp_group,
     pulp_option,
 )
@@ -47,9 +47,9 @@ def orphan(ctx: click.Context, pulp_ctx: PulpCLIContext) -> None:
     ),
     needs_plugins=[PluginRequirement("core", specifier=">=3.15.0")],
 )
-@pass_entity_context
+@pass_view_set_context
 @pass_pulp_context
-def cleanup(pulp_ctx: PulpCLIContext, orphan_ctx: PulpEntityContext, **kwargs: t.Any) -> None:
+def cleanup(pulp_ctx: PulpCLIContext, orphan_ctx: PulpViewSetContext, **kwargs: t.Any) -> None:
     """
     Cleanup orphaned content.
     """