When no repo, use synchronous upload for rpm

Author: YasenT

pulp-glue/pulp_glue/rpm/context.py

@@ -1,3 +1,4 @@
+import os
 import typing as t
 
 from pulp_glue.common.context import (
@@ -154,6 +155,60 @@ def list_iterator(
             stats=stats,
         )
 
+    def upload(
+        self,
+        file: t.IO[bytes],
+        chunk_size: int,
+        repository: t.Optional[PulpRepositoryContext],
+        **kwargs: t.Any,
+    ) -> t.Any:
+        """
+        Create an RPM package by uploading a file.
+
+        This function is deprecated. The create call can handle the upload logic transparently.
+
+        Parameters:
+            file: A file like object that supports `os.path.getsize`.
+            chunk_size: Size of the chunks to upload independently.
+            repository: Repository context to add the newly created content to.
+            kwargs: Extra args specific to the content type, passed to the create call.
+
+        Returns:
+            The result of the create task.
+        """
+        self.needs_capability("upload")
+        size = os.path.getsize(file.name)
+        body: t.Dict[str, t.Any] = {**kwargs}
+
+        if not self.pulp_ctx.fake_mode:
+            if chunk_size > size:
+                # Small file: direct upload
+                body["file"] = file
+            else:
+                # Large file: chunked upload
+                if self.pulp_ctx.has_plugin(PluginRequirement("core", specifier=">=3.20.0")):
+                    from pulp_glue.core.context import PulpUploadContext
+
+                    upload_href = PulpUploadContext(self.pulp_ctx).upload_file(file, chunk_size)
+                    body["upload"] = upload_href
+                else:
+                    from pulp_glue.core.context import PulpArtifactContext
+
+                    artifact_href = PulpArtifactContext(self.pulp_ctx).upload(file, chunk_size)
+                    body["artifact"] = artifact_href
+
+        # For rpm plugin >= 3.32.5, use synchronous upload endpoint when no repository is provided
+        # For older versions, always use the create endpoint (backward compatibility)
+        if repository is None and self.pulp_ctx.has_plugin(
+            PluginRequirement("rpm", specifier=">=3.32.5")
+        ):
+            return self.call("upload", body=body)
+
+        # Repository is specified or older rpm version: use create endpoint (async path)
+        if repository is not None:
+            body["repository"] = repository
+        return self.create(body=body)
+
 
 class PulpRpmAdvisoryContext(PulpContentContext):
     PLUGIN = "rpm"

pulpcore/cli/rpm/content.py

@@ -389,14 +389,17 @@ def upload(
                 _("You must specify one (and only one) of --file or --directory.")
             )
 
-        # Sanity: If directory, repository required
+        # Sanity: If using temp repository, both directory and repository are required
         final_dest_repo_ctx = kwargs["repository"]
-        if directory and not final_dest_repo_ctx:
+        if kwargs["use_temp_repository"] and not (directory and final_dest_repo_ctx):
             raise click.ClickException(
-                _("You must specify a --repository to use --directory uploads.")
+                _(
+                    "You must specify both --directory and --repository "
+                    "to use --use-temp-repository."
+                )
             )
 
-        # Sanity: ignore publish|use_temp unless directory has been specified
+        # Sanity: ignore publish|use_temp unless destination-repository has been specified
         use_tmp = final_dest_repo_ctx and kwargs["use_temp_repository"]
         do_publish = final_dest_repo_ctx and kwargs["publish"]
 
@@ -416,10 +419,11 @@ def upload(
             )
         else:
             # Upload a directory-full of RPMs
+            dest_repo_ctx = None
             try:
                 dest_repo_ctx = _determine_upload_repository(final_dest_repo_ctx, pulp_ctx, use_tmp)
                 result = _upload_rpms(entity_ctx, dest_repo_ctx, directory, chunk_size)
-                if use_tmp:
+                if use_tmp and dest_repo_ctx:
                     result = _copy_to_final(dest_repo_ctx, final_dest_repo_ctx, pulp_ctx)
             finally:
                 if use_tmp and dest_repo_ctx:
@@ -488,23 +492,36 @@ def _copy_to_final(
 
 def _upload_rpms(
     entity_ctx: PulpContentContext,
-    dest_repo_ctx: PulpRpmRepositoryContext,
+    dest_repo_ctx: t.Optional[PulpRpmRepositoryContext],
     directory: t.Any,
     chunk_size: int,
 ) -> t.Any:
     rpms_path = f"{directory}/*.rpm"
     rpm_names = glob.glob(rpms_path)
     if not rpm_names:
         raise click.ClickException(_("Directory {} has no .rpm files in it.").format(directory))
-    click.echo(
-        _(
-            "About to upload {} files for {}.".format(
-                len(rpm_names),
-                dest_repo_ctx.entity["name"],
-            )
-        ),
-        err=True,
-    )
+
+    # Build message based on whether we have a repository or not
+    if dest_repo_ctx:
+        click.echo(
+            _(
+                "About to upload {} files for {}.".format(
+                    len(rpm_names),
+                    dest_repo_ctx.entity["name"],
+                )
+            ),
+            err=True,
+        )
+    else:
+        click.echo(
+            _(
+                "About to upload {} files from {}.".format(
+                    len(rpm_names),
+                    directory,
+                )
+            ),
+            err=True,
+        )
     # Upload all *.rpm into the destination
     successful_uploads = 0
     for name in rpm_names:
@@ -525,9 +542,11 @@ def _upload_rpms(
 
 
 def _determine_upload_repository(
-    final_dest_repo_ctx: PulpRpmRepositoryContext, pulp_ctx: PulpCLIContext, use_tmp: bool
-) -> PulpRpmRepositoryContext:
-    if use_tmp:
+    final_dest_repo_ctx: t.Optional[PulpRpmRepositoryContext],
+    pulp_ctx: PulpCLIContext,
+    use_tmp: bool,
+) -> t.Optional[PulpRpmRepositoryContext]:
+    if use_tmp and final_dest_repo_ctx:
         dest_repo_ctx = PulpRpmRepositoryContext(pulp_ctx)
         body: t.Dict[str, t.Any] = {
             "name": f"uploadtmp_{final_dest_repo_ctx.entity['name']}_{uuid4()}",