Create relations in bulk

CHANGES/6940.bugfix

@@ -0,0 +1 @@
+Improved the performance of the synchronization

pulp_container/app/tasks/sync_stages.py

@@ -459,18 +459,41 @@ async def run(self):
         """
         Relate each item in the input queue to objects specified on the DeclarativeContent.
         """
-        async for dc in self.items():
-
-            if dc.extra_data.get('relation'):
-                self.relate_manifest_to_list(dc)
-            elif dc.extra_data.get('blob_relation'):
-                self.relate_blob(dc)
-            elif dc.extra_data.get('config_relation'):
-                self.relate_config_blob(dc)
-            elif dc.extra_data.get('man_relation'):
-                self.relate_manifest_tag(dc)
+        async for batch in self.batches():
+            manifest_to_list_list = []
+            blob_list = []
+            config_blob_list = []
+            for dc in batch:
+                if dc.extra_data.get("relation"):
+                    manifest_to_list = self.relate_manifest_to_list(dc)
+                    manifest_to_list_list.append(manifest_to_list)
+                elif dc.extra_data.get("blob_relation"):
+                    blob = self.relate_blob(dc)
+                    blob_list.append(blob)
+                elif dc.extra_data.get("config_relation"):
+                    config_blob = self.relate_config_blob(dc)
+                    config_blob_list.append(config_blob)
+                elif dc.extra_data.get("man_relation"):
+                    self.relate_manifest_tag(dc)
+
+            ManifestListManifest.objects.bulk_create(
+                objs=manifest_to_list_list,
+                ignore_conflicts=True,
+                batch_size=1000
+            )
+            BlobManifest.objects.bulk_create(
+                objs=blob_list,
+                ignore_conflicts=True,
+                batch_size=1000
+            )
+            Manifest.objects.bulk_update(
+                objs=config_blob_list,
+                fields=["config_blob"],
+                batch_size=1000
+            )
 
-            await self.put(dc)
+            for dc in batch:
+                await self.put(dc)
 
     def relate_config_blob(self, dc):
         """
@@ -479,10 +502,14 @@ def relate_config_blob(self, dc):
         Args:
             dc (pulpcore.plugin.stages.DeclarativeContent): dc for a Blob
 
+        Returns:
+            pulp_container.app.models.Manifest: An existing Manifest object with the updated
+                config layer
+
         """
-        configured_dc = dc.extra_data.get('config_relation')
+        configured_dc = dc.extra_data.get("config_relation")
         configured_dc.content.config_blob = dc.content
-        configured_dc.content.save()
+        return configured_dc.content
 
     def relate_blob(self, dc):
         """
@@ -491,13 +518,13 @@ def relate_blob(self, dc):
         Args:
             dc (pulpcore.plugin.stages.DeclarativeContent): dc for a Blob
 
+        Returns:
+            pulp_container.app.models.BlobManifest: A new BlobManifest object with the added
+                reference to a Manifest object
+
         """
-        related_dc = dc.extra_data.get('blob_relation')
-        thru = BlobManifest(manifest=related_dc.content, manifest_blob=dc.content)
-        try:
-            thru.save()
-        except IntegrityError:
-            pass
+        related_dc = dc.extra_data.get("blob_relation")
+        return BlobManifest(manifest=related_dc.content, manifest_blob=dc.content)
 
     def relate_manifest_tag(self, dc):
         """
@@ -507,13 +534,15 @@ def relate_manifest_tag(self, dc):
             dc (pulpcore.plugin.stages.DeclarativeContent): dc for a Tag
 
         """
-        related_dc = dc.extra_data.get('man_relation')
+        related_dc = dc.extra_data.get("man_relation")
         dc.content.tagged_manifest = related_dc.content
         try:
             dc.content.save()
         except IntegrityError:
-            existing_tag = Tag.objects.get(name=dc.content.name,
-                                           tagged_manifest=related_dc.content)
+            existing_tag = Tag.objects.get(
+                name=dc.content.name,
+                tagged_manifest=related_dc.content
+            )
             dc.content.delete()
             dc.content = existing_tag
 
@@ -524,19 +553,20 @@ def relate_manifest_to_list(self, dc):
         Args:
             dc (pulpcore.plugin.stages.DeclarativeContent): dc for a ImageManifest
 
-        """
-        related_dc = dc.extra_data.get('relation')
-        platform = dc.extra_data.get('platform')
-        thru = ManifestListManifest(manifest_list=dc.content, image_manifest=related_dc.content,
-                                    architecture=platform['architecture'],
-                                    os=platform['os'],
-                                    features=platform.get('features'),
-                                    variant=platform.get('variant'),
-                                    os_version=platform.get('os.version'),
-                                    os_features=platform.get('os.features')
-                                    )
-
-        try:
-            thru.save()
-        except IntegrityError:
-            pass
+        Returns:
+            pulp_container.app.models.ManifestListManifest: A new ManifestListManifest object
+                with the associated ImageManifest object and corresponding platform information
+
+        """
+        related_dc = dc.extra_data.get("relation")
+        platform = dc.extra_data.get("platform")
+        return ManifestListManifest(
+            manifest_list=dc.content,
+            image_manifest=related_dc.content,
+            architecture=platform["architecture"],
+            os=platform["os"],
+            features=platform.get("features"),
+            variant=platform.get("variant"),
+            os_version=platform.get("os.version"),
+            os_features=platform.get("os.features")
+        )