tmp

Author: jobselko

pulp_python/app/serializers.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import tempfile
 from gettext import gettext as _
 from django.conf import settings
 from django.db.utils import IntegrityError
@@ -551,9 +552,12 @@ def validate(self, data):
             )
         # Create metadata artifact for wheel files
         if filename.endswith(".whl"):
-            if metadata_artifact := artifact_to_metadata_artifact(filename, artifact):
-                data["metadata_artifact"] = metadata_artifact
-                data["metadata_sha256"] = metadata_artifact.sha256
+            with tempfile.TemporaryDirectory(dir=settings.WORKING_DIRECTORY) as temp_dir:
+                if metadata_artifact := artifact_to_metadata_artifact(
+                    filename, artifact, tmp_dir=temp_dir
+                ):
+                    data["metadata_artifact"] = metadata_artifact
+                    data["metadata_sha256"] = metadata_artifact.sha256
 
         return data
 

pulp_python/app/utils.py

@@ -261,33 +261,34 @@ def artifact_to_python_content_data(filename, artifact, domain=None):
     return data
 
 
-def artifact_to_metadata_artifact(filename: str, artifact: Artifact) -> Artifact | None:
+def artifact_to_metadata_artifact(
+    filename: str, artifact: Artifact, tmp_dir: str = "."
+) -> Artifact | None:
     """
     Creates artifact for metadata from the provided wheel artifact.
     """
     if not filename.endswith(".whl"):
         return None
 
-    with tempfile.TemporaryDirectory(dir=settings.WORKING_DIRECTORY) as temp_dir_name:
-        with tempfile.NamedTemporaryFile(dir=temp_dir_name, suffix=filename) as temp_file:
-            artifact.file.seek(0)
-            shutil.copyfileobj(artifact.file, temp_file)
-            temp_file.flush()
-            metadata_content = extract_wheel_metadata(temp_file.name)
-            if not metadata_content:
-                return None
-
-            with tempfile.NamedTemporaryFile(dir=temp_dir_name, suffix=".metadata") as temp_md:
-                temp_md.write(metadata_content)
-                temp_md.flush()
-                metadata_artifact = Artifact.init_and_validate(temp_md.name)
-                try:
-                    metadata_artifact.save()
-                except IntegrityError:
-                    metadata_artifact = Artifact.objects.get(
-                        sha256=metadata_artifact.sha256, pulp_domain=get_domain()
-                    )
-                return metadata_artifact
+    with tempfile.NamedTemporaryFile("wb", dir=tmp_dir, suffix=filename) as temp_file:
+        artifact.file.seek(0)
+        shutil.copyfileobj(artifact.file, temp_file)
+        temp_file.flush()
+        metadata_content = extract_wheel_metadata(temp_file.name)
+        if not metadata_content:
+            return None
+
+        with tempfile.NamedTemporaryFile("wb", dir=tmp_dir, suffix=".metadata") as temp_md:
+            temp_md.write(metadata_content)
+            temp_md.flush()
+            metadata_artifact = Artifact.init_and_validate(temp_md.name)
+            try:
+                metadata_artifact.save()
+            except IntegrityError:
+                metadata_artifact = Artifact.objects.get(
+                    sha256=metadata_artifact.sha256, pulp_domain=get_domain()
+                )
+            return metadata_artifact
 
 
 def fetch_json_release_metadata(name: str, version: str, remotes: set[Remote]) -> dict: