tmp 2

Author: jobselko

pulp_python/app/utils.py

@@ -1,5 +1,6 @@
 import hashlib
 import logging
+import os
 import pkginfo
 import re
 import shutil
@@ -270,25 +271,42 @@ def artifact_to_metadata_artifact(
     if not filename.endswith(".whl"):
         return None
 
-    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()
+    temp_wheel_path = None
+    temp_metadata_path = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            "wb", dir=tmp_dir, suffix=filename, delete=False
+        ) as temp_file:
+            temp_wheel_path = temp_file.name
+            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:
+        with tempfile.NamedTemporaryFile(
+            "wb", dir=tmp_dir, suffix=".metadata", delete=False
+        ) as temp_md:
+            temp_metadata_path = temp_md.name
             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
+
+        metadata_artifact = Artifact.init_and_validate(temp_metadata_path)
+        try:
+            metadata_artifact.save()
+        except IntegrityError:
+            metadata_artifact = Artifact.objects.get(
+                sha256=metadata_artifact.sha256, pulp_domain=get_domain()
+            )
+        return metadata_artifact
+
+    finally:
+        if temp_wheel_path and os.path.exists(temp_wheel_path):
+            os.unlink(temp_wheel_path)
+        if temp_metadata_path and os.path.exists(temp_metadata_path):
+            os.unlink(temp_metadata_path)
 
 
 def fetch_json_release_metadata(name: str, version: str, remotes: set[Remote]) -> dict: