Add more Pulp Exceptions.

  Assisted-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: aKlimau

CHANGES/+add-pulp-exceptions.feature

@@ -0,0 +1 @@
+Add more Pulp Exceptions.

pulp_python/app/exceptions.py

@@ -0,0 +1,169 @@
+from gettext import gettext as _
+
+from pulpcore.plugin.exceptions import PulpException
+
+
+class ProvenanceVerificationError(PulpException):
+    """
+    Raised when provenance verification fails.
+    """
+
+    error_code = "PYT0001"
+
+    def __init__(self, message):
+        """
+        :param message: Description of the provenance verification error
+        :type message: str
+        """
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("Provenance verification failed: {message}").format(
+            message=self.message
+        )
+
+
+class AttestationVerificationError(PulpException):
+    """
+    Raised when attestation verification fails.
+    """
+
+    error_code = "PYT0002"
+
+    def __init__(self, message):
+        """
+        :param message: Description of the attestation verification error
+        :type message: str
+        """
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("Attestation verification failed: {message}").format(
+            message=self.message
+        )
+
+
+class PackageSubstitutionError(PulpException):
+    """
+    Raised when packages with the same filename but different checksums are being added.
+    """
+
+    error_code = "PYT0003"
+
+    def __init__(self, duplicates):
+        """
+        :param duplicates: Description of duplicate packages
+        :type duplicates: str
+        """
+        self.duplicates = duplicates
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _(
+            "Found duplicate packages being added with the same filename but different "
+            "checksums. To allow this, set 'allow_package_substitution' to True on the "
+            "repository. Conflicting packages: {duplicates}"
+        ).format(duplicates=self.duplicates)
+
+
+class UnsupportedProtocolError(PulpException):
+    """
+    Raised when an unsupported protocol is used for syncing.
+    """
+
+    error_code = "PYT0004"
+
+    def __init__(self, protocol):
+        """
+        :param protocol: The unsupported protocol
+        :type protocol: str
+        """
+        self.protocol = protocol
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _(
+            "Only HTTP(S) is supported for python syncing, got: {protocol}"
+        ).format(protocol=self.protocol)
+
+
+class MissingRelativePathError(PulpException):
+    """
+    Raised when relative_path field is missing during package upload.
+    """
+
+    error_code = "PYT0005"
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("This field is required: relative_path")
+
+
+class InvalidPythonExtensionError(PulpException):
+    """
+    Raised when a file has an invalid Python package extension.
+    """
+
+    error_code = "PYT0006"
+
+    def __init__(self, filename):
+        """
+        :param filename: The filename with invalid extension
+        :type filename: str
+        """
+        self.filename = filename
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _(
+            "Extension on {filename} is not a valid python extension "
+            "(.whl, .exe, .egg, .tar.gz, .tar.bz2, .zip)"
+        ).format(filename=self.filename)
+
+
+class InvalidProvenanceError(PulpException):
+    """
+    Raised when uploaded provenance data is invalid.
+    """
+
+    error_code = "PYT0007"
+
+    def __init__(self, message):
+        """
+        :param message: Description of the provenance validation error
+        :type message: str
+        """
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _(
+            "The uploaded provenance is not valid: {message}"
+        ).format(message=self.message)
+
+
+class RemoteFetchError(PulpException):
+    """
+    Raised when fetching metadata from all remotes fails.
+    """
+
+    error_code = "PYT0008"
+
+    def __init__(self, url):
+        self.url = url
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("Failed to fetch {url} from any remote.").format(
+            url=self.url
+        )
+
+
+class InvalidAttestationsError(PulpException):
+    """
+    Raised when attestation data cannot be validated.
+    """
+
+    error_code = "PYT0009"
+
+    def __init__(self, message):
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("Invalid attestations: {message}").format(
+            message=self.message
+        )

pulp_python/app/models.py

@@ -11,7 +11,6 @@
     BEFORE_SAVE,
     hook,
 )
-from rest_framework.serializers import ValidationError
 from pulpcore.plugin.models import (
     AutoAddObjPermsMixin,
     Content,
@@ -23,6 +22,7 @@
 from pulpcore.plugin.responses import ArtifactResponse
 
 from pathlib import PurePath
+from .exceptions import PackageSubstitutionError
 from .provenance import Provenance
 from .utils import (
     artifact_to_python_content_data,
@@ -407,14 +407,10 @@ def finalize_new_version(self, new_version):
 
     def _check_for_package_substitution(self, new_version):
         """
-        Raise a ValidationError if newly added packages would replace existing packages that have
-        the same filename but a different sha256 checksum.
+        Raise a PackageSubstitutionError if newly added packages would replace existing packages
+        that have the same filename but a different sha256 checksum.
         """
         qs = PythonPackageContent.objects.filter(pk__in=new_version.content)
         duplicates = collect_duplicates(qs, ("filename",))
         if duplicates:
-            raise ValidationError(
-                "Found duplicate packages being added with the same filename but different checksums. "  # noqa: E501
-                "To allow this, set 'allow_package_substitution' to True on the repository. "
-                f"Conflicting packages: {duplicates}"
-            )
+            raise PackageSubstitutionError(duplicates)

pulp_python/app/serializers.py

@@ -8,14 +8,22 @@
 from packaging.requirements import Requirement
 from rest_framework import serializers
 from pypi_attestations import AttestationError
-from pydantic import TypeAdapter, ValidationError
+from pydantic import TypeAdapter, ValidationError as PydanticValidationError
 from urllib.parse import urljoin
 
+from pulpcore.plugin.exceptions import DigestValidationError
 from pulpcore.plugin import models as core_models
 from pulpcore.plugin import serializers as core_serializers
 from pulpcore.plugin.util import get_domain, get_prn, get_current_authenticated_user
 
 from pulp_python.app import models as python_models
+from pulp_python.app.exceptions import (
+    AttestationVerificationError,
+    InvalidProvenanceError,
+    InvalidPythonExtensionError,
+    MissingRelativePathError,
+    ProvenanceVerificationError,
+)
 from pulp_python.app.provenance import (
     Attestation,
     Provenance,
@@ -374,7 +382,7 @@ def validate_attestations(self, value):
                 attestations = TypeAdapter(list[Attestation]).validate_json(value)
             else:
                 attestations = TypeAdapter(list[Attestation]).validate_python(value)
-        except ValidationError as e:
+        except PydanticValidationError as e:
             raise serializers.ValidationError(_("Invalid attestations: {}".format(e)))
         return attestations
 
@@ -387,9 +395,7 @@ def handle_attestations(self, filename, sha256, attestations, offline=True):
         try:
             verify_provenance(filename, sha256, provenance, offline=offline)
         except AttestationError as e:
-            raise serializers.ValidationError(
-                {"attestations": _("Attestations failed verification: {}".format(e))}
-            )
+            raise AttestationVerificationError(str(e))
         return provenance.model_dump(mode="json")
 
     def deferred_validate(self, data):
@@ -408,26 +414,18 @@ def deferred_validate(self, data):
         try:
             filename = data["relative_path"]
         except KeyError:
-            raise serializers.ValidationError(detail={"relative_path": _("This field is required")})
+            raise MissingRelativePathError()
 
         artifact = data["artifact"]
         try:
             _data = artifact_to_python_content_data(filename, artifact, domain=get_domain())
         except ValueError:
-            raise serializers.ValidationError(
-                _(
-                    "Extension on {} is not a valid python extension "
-                    "(.whl, .exe, .egg, .tar.gz, .tar.bz2, .zip)"
-                ).format(filename)
-            )
+            raise InvalidPythonExtensionError(filename)
 
         if data.get("sha256") and data["sha256"] != artifact.sha256:
-            raise serializers.ValidationError(
-                detail={
-                    "sha256": _(
-                        "The uploaded artifact's sha256 checksum does not match the one provided"
-                    )
-                }
+            raise DigestValidationError(
+                actual=artifact.sha256,
+                expected=data["sha256"],
             )
 
         data.update(_data)
@@ -641,15 +639,13 @@ def deferred_validate(self, data):
         try:
             provenance = Provenance.model_validate_json(data["file"].read())
             data["provenance"] = provenance.model_dump(mode="json")
-        except ValidationError as e:
-            raise serializers.ValidationError(
-                _("The uploaded provenance is not valid: {}".format(e))
-            )
+        except PydanticValidationError as e:
+            raise InvalidProvenanceError(str(e))
         if data.pop("verify"):
             try:
                 verify_provenance(data["package"].filename, data["package"].sha256, provenance)
             except AttestationError as e:
-                raise serializers.ValidationError(_("Provenance verification failed: {}".format(e)))
+                raise ProvenanceVerificationError(str(e))
         return data
 
     def retrieve(self, validated_data):

pulp_python/app/tasks/sync.py

@@ -3,11 +3,9 @@
 
 from aiohttp import ClientResponseError, ClientError
 from lxml.etree import LxmlError
-from gettext import gettext as _
 from functools import partial
 
-from rest_framework import serializers
-
+from pulpcore.plugin.exceptions import SyncError
 from pulpcore.plugin.download import HttpDownloader
 from pulpcore.plugin.models import Artifact, ProgressReport, Remote, Repository
 from pulpcore.plugin.stages import (
@@ -17,6 +15,7 @@
     Stage,
 )
 
+from pulp_python.app.exceptions import UnsupportedProtocolError
 from pulp_python.app.models import (
     PythonPackageContent,
     PythonRemote,
@@ -54,7 +53,7 @@ def sync(remote_pk, repository_pk, mirror):
     repository = Repository.objects.get(pk=repository_pk)
 
     if not remote.url:
-        raise serializers.ValidationError(detail=_("A remote must have a url attribute to sync."))
+        raise SyncError("A remote must have a url attribute to sync.")
 
     first_stage = PythonBanderStage(remote)
     DeclarativeVersion(first_stage, repository, mirror).create()
@@ -117,7 +116,8 @@ async def run(self):
         url = self.remote.url.rstrip("/")
         downloader = self.remote.get_downloader(url=url)
         if not isinstance(downloader, HttpDownloader):
-            raise ValueError("Only HTTP(S) is supported for python syncing")
+            protocol = type(downloader).__name__
+            raise UnsupportedProtocolError(protocol)
 
         async with Master(url, allow_non_https=True) as master:
             # Replace the session with the remote's downloader session

pulp_python/app/tasks/upload.py

@@ -3,10 +3,12 @@
 from datetime import datetime, timezone
 from django.db import transaction
 from django.contrib.sessions.models import Session
-from pydantic import TypeAdapter
+from pydantic import TypeAdapter, ValidationError as PydanticValidationError
+from pypi_attestations import AttestationError
 from pulpcore.plugin.models import Artifact, CreatedResource, Content, ContentArtifact
 from pulpcore.plugin.util import get_domain, get_current_authenticated_user, get_prn
 
+from pulp_python.app.exceptions import AttestationVerificationError, InvalidAttestationsError
 from pulp_python.app.models import PythonPackageContent, PythonRepository, PackageProvenance
 from pulp_python.app.provenance import (
     Attestation,
@@ -122,13 +124,19 @@ def create_provenance(package, attestations, domain):
     Returns:
         the newly created PackageProvenance
     """
-    attestations = TypeAdapter(list[Attestation]).validate_python(attestations)
+    try:
+        attestations = TypeAdapter(list[Attestation]).validate_python(attestations)
+    except PydanticValidationError as e:
+        raise InvalidAttestationsError(str(e))
 
     user = get_current_authenticated_user()
     publisher = AnyPublisher(kind="Pulp User", prn=get_prn(user))
     att_bundle = AttestationBundle(publisher=publisher, attestations=attestations)
     provenance = Provenance(attestation_bundles=[att_bundle])
-    verify_provenance(package.filename, package.sha256, provenance)
+    try:
+        verify_provenance(package.filename, package.sha256, provenance)
+    except AttestationError as e:
+        raise AttestationVerificationError(str(e))
     provenance_json = provenance.model_dump(mode="json")
 
     prov_sha256 = PackageProvenance.calculate_sha256(provenance_json)

pulp_python/app/utils.py

@@ -20,6 +20,8 @@
 from pulpcore.plugin.exceptions import TimeoutException
 from pulpcore.plugin.util import get_domain
 
+from pulp_python.app.exceptions import RemoteFetchError
+
 log = logging.getLogger(__name__)
 
 
@@ -325,7 +327,7 @@ def fetch_json_release_metadata(name: str, version: str, remotes: set[Remote]) -
             json_data = json.load(file)
         return json_data
     else:
-        raise Exception(f"Failed to fetch {url} from any remote.")
+        raise RemoteFetchError(url=url)
 
 
 def python_content_to_json(base_path, content_query, version=None, domain=None):