Enable selective repository sync for replication

CHANGES/+optional-request-body.feature

@@ -0,0 +1 @@
+Only mark request bodies as `required` in the OpenAPI schema when the serializer has at least one required field. Previously all request bodies were unconditionally marked required, which forced generated client bindings to demand a body object even for endpoints where every field is optional.

CHANGES/7617.feature

@@ -0,0 +1 @@
+Added support for passing ``q_select`` as a parameter to the replicate action, allowing users to selectively sync a subset of upstream distributions without modifying the stored upstream-pulp configuration.

pulpcore/app/serializers/__init__.py

@@ -125,7 +125,7 @@
     UserRoleSerializer,
     UserSerializer,
 )
-from .replica import UpstreamPulpSerializer
+from .replica import UpstreamPulpReplicateSerializer, UpstreamPulpSerializer
 from .vulnerability_report import VulnerabilityReportSerializer
 from .openpgp import (
     OpenPGPDistributionSerializer,

pulpcore/app/serializers/replica.py

@@ -124,3 +124,22 @@ class Meta:
             "last_replication",
             "policy",
         )
+
+
+class UpstreamPulpReplicateSerializer(serializers.Serializer):
+    q_select = serializers.CharField(
+        help_text=_(
+            "Filter distributions on the upstream Pulp using complex filtering. "
+            "When specified, overrides the stored q_select for this replication run only. "
+            'E.g. pulp_label_select="foo" OR pulp_label_select="key=val"',
+        ),
+        allow_null=True,
+        allow_blank=True,
+        required=False,
+    )
+
+    def validate_q_select(self, value):
+        from pulpcore.app.viewsets import DistributionFilter
+
+        DistributionFilter().filters["q"].field.clean(value)
+        return value

pulpcore/app/tasks/replica.py

@@ -28,7 +28,7 @@ def user_agent():
     return f"pulpcore/{pulp_version} ({python}, {system}) (pulp-glue {pulp_glue_version})"
 
 
-def replicate_distributions(server_pk):
+def replicate_distributions(server_pk, q_select=None):
     server = UpstreamPulp.objects.get(pk=server_pk)
 
     # Write out temporary files related to SSL
@@ -82,10 +82,11 @@ def replicate_distributions(server_pk):
                         replicator = replicator_class(ctx, task_group, tls_settings, server)
                         supported_replicators.append(replicator)
 
+        effective_q_select = q_select if q_select is not None else server.q_select
         distro_repo_pairs = []
         for replicator in supported_replicators:
             distro_names = []
-            distros = replicator.upstream_distributions(q=server.q_select)
+            distros = replicator.upstream_distributions(q=effective_q_select)
             for distro in distros:
                 # Create remote
                 remote = replicator.create_or_update_remote(upstream_distribution=distro)
@@ -111,7 +112,13 @@ def replicate_distributions(server_pk):
                 distro_names.append(distro["name"])
                 distro_repo_pairs.append((distro["name"], str(repository.pk)))
 
-            replicator.remove_missing(distro_names)
+            # When a per-request q_select override is used, this is a selective sync
+            # of a subset of distributions.  Skipping remove_missing avoids deleting
+            # distributions that simply weren't included in the filter — but it also
+            # means that distributions removed from upstream won't be cleaned up until
+            # a full (non-overridden) replication runs.
+            if q_select is None:
+                replicator.remove_missing(distro_names)
     except GluePulpException as e:
         raise ExternalServiceError(service_name=server.base_url, details=str(e))
 

pulpcore/app/viewsets/replica.py

@@ -8,7 +8,11 @@
 
 from pulpcore.app.models import TaskGroup, UpstreamPulp
 from pulpcore.app.response import TaskGroupOperationResponse
-from pulpcore.app.serializers import TaskGroupOperationResponseSerializer, UpstreamPulpSerializer
+from pulpcore.app.serializers import (
+    TaskGroupOperationResponseSerializer,
+    UpstreamPulpReplicateSerializer,
+    UpstreamPulpSerializer,
+)
 from pulpcore.app.tasks import replicate_distributions
 from pulpcore.app.viewsets import NamedModelViewSet, RolesMixin
 from pulpcore.app.viewsets.base import DATETIME_FILTER_OPTIONS, NAME_FILTER_OPTIONS
@@ -118,24 +122,31 @@ class UpstreamPulpViewSet(
         summary="Replicate",
         description="Trigger an asynchronous repository replication task group. This API is "
         "provided as a tech preview.",
-        request=None,
+        request=UpstreamPulpReplicateSerializer,
         responses={202: TaskGroupOperationResponseSerializer},
     )
-    @action(detail=True, methods=["post"])
+    @action(detail=True, methods=["post"], serializer_class=UpstreamPulpReplicateSerializer)
     def replicate(self, request, pk):
         """
         Triggers an asynchronous repository replication operation.
         """
+        serializer = UpstreamPulpReplicateSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
         server = UpstreamPulp.objects.get(pk=pk)
         task_group = TaskGroup.objects.create(description=f"Replication of {server.name}")
 
         exclusive_resources = [f"pdrn:{request.pulp_domain.pulp_id}:servers"]
 
+        task_kwargs = {"server_pk": pk}
+        if q_select := serializer.validated_data.get("q_select"):
+            task_kwargs["q_select"] = q_select
+
         dispatch(
             replicate_distributions,
             exclusive_resources=exclusive_resources,
             shared_resources=[server],
-            kwargs={"server_pk": pk},
+            kwargs=task_kwargs,
             task_group=task_group,
         )
 

pulpcore/openapi/__init__.py

@@ -241,7 +241,12 @@ def _get_request_body(self):
         """Get request body."""
         request_body = super()._get_request_body()
         if request_body:
-            request_body["required"] = True
+            serializer = force_instance(self.get_request_serializer())
+            has_required_fields = any(
+                field.required for field in getattr(serializer, "fields", {}).values()
+            )
+            if has_required_fields:
+                request_body["required"] = True
         return request_body
 
     def _get_response_bodies(self):

pulpcore/tests/functional/api/test_replication.py

@@ -779,6 +779,67 @@ def test_replicate_with_basic_q_select(
     assert result.results[0].name == "0"
 
 
+@pytest.mark.parallel
+def test_replicate_with_per_request_q_select(
+    domain_factory,
+    populate_upstream,
+    bindings_cfg,
+    pulpcore_bindings,
+    monitor_task_group,
+    pulp_settings,
+    gen_object_with_cleanup,
+    add_domain_objects_to_cleanup,
+):
+    """Test that q_select can be passed per-request to the replicate action."""
+    source_domain = populate_upstream(6)
+    dest_domain = domain_factory()
+    add_domain_objects_to_cleanup(dest_domain)
+
+    # Create upstream pulp with NO stored q_select
+    upstream = gen_object_with_cleanup(
+        pulpcore_bindings.UpstreamPulpsApi,
+        {
+            "name": str(uuid.uuid4()),
+            "base_url": bindings_cfg.host,
+            "api_root": pulp_settings.API_ROOT,
+            "domain": source_domain.name,
+            "username": bindings_cfg.username,
+            "password": bindings_cfg.password,
+        },
+        pulp_domain=dest_domain.name,
+    )
+
+    # Selective replicate: only sync 'even' distributions
+    replicate_body = pulpcore_bindings.module.UpstreamPulpReplicate(
+        q_select="pulp_label_select='even'"
+    )
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(
+        upstream.pulp_href, upstream_pulp_replicate=replicate_body
+    )
+    monitor_task_group(response.task_group)
+    result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
+    assert result.count == 3
+    assert {d.name for d in result.results} == {"0", "2", "4"}
+
+    # Selective replicate of 'odd' should NOT delete the 'even' ones (remove_missing skipped)
+    replicate_body = pulpcore_bindings.module.UpstreamPulpReplicate(
+        q_select="pulp_label_select='odd'"
+    )
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(
+        upstream.pulp_href, upstream_pulp_replicate=replicate_body
+    )
+    monitor_task_group(response.task_group)
+    result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
+    assert result.count == 6
+    assert {d.name for d in result.results} == {"0", "1", "2", "3", "4", "5"}
+
+    # Full replicate (no per-request q_select) should still work and run remove_missing
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(upstream.pulp_href)
+    monitor_task_group(response.task_group)
+    result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
+    assert result.count == 6
+
+
 @pytest.mark.parallel
 def test_replicate_with_complex_q_select(
     domain_factory,