pulp · aKlimau · Apr 13, 2026
diff --git a/CHANGES/+add-pulp-exceptions.feature b/CHANGES/+add-pulp-exceptions.feature
@@ -0,0 +1 @@
+Add more Pulp Exceptions.
diff --git a/pulp_ansible/app/serializers.py b/pulp_ansible/app/serializers.py
@@ -11,7 +11,12 @@
 from drf_spectacular.utils import extend_schema_field
 
 from galaxy_importer.constants import NAME_REGEXP
+from pulpcore.plugin.exceptions import DigestValidationError
 from pulpcore.plugin.models import Artifact, Content, ContentArtifact, SigningService
+from pulp_ansible.exceptions import (
+    CollectionFilenameParseError,
+    MissingExpectedFieldsError,
+)
 from pulpcore.plugin.serializers import (
     DetailRelatedField,
     ContentChecksumSerializer,
@@ -484,19 +489,12 @@ def validate(self, data):
         fields = ("namespace", "name", "version")
         if not all((f"expected_{x}" in data for x in fields)):
             if not ("file" in data or "filename" in self.context):
-                raise ValidationError(
-                    _(
-                        "expected_namespace, expected_name, and expected_version must be "
-                        "specified when using artifact or upload objects"
-                    )
-                )
+                raise MissingExpectedFieldsError()
             filename = self.context.get("filename") or data["file"].name
             try:
                 collection = parse_collection_filename(filename)
             except ValueError:
-                raise ValidationError(
-                    _("Failed to parse Collection file upload '{}'").format(filename)
-                )
+                raise CollectionFilenameParseError(filename)
             data["expected_namespace"] = collection.namespace
             data["expected_name"] = collection.name
             data["expected_version"] = collection.version
@@ -511,7 +509,7 @@ def deferred_validate(self, data):
         artifact = data.get("artifact")
 
         if (sha256 := data.get("sha256")) and sha256 != artifact.sha256:
-            raise ValidationError(_("Expected sha256 did not match uploaded artifact's sha256"))
+            raise DigestValidationError(actual=artifact.sha256, expected=sha256)
 
         collection_info = process_collection_artifact(
             artifact=artifact,

diff --git a/pulp_ansible/app/tasks/collections.py b/pulp_ansible/app/tasks/collections.py
@@ -59,7 +59,12 @@
 from semantic_version.base import Always
 
 from pulp_ansible.app.constants import PAGE_SIZE
-from pulp_ansible.exceptions import CollectionNotFound
+from pulp_ansible.exceptions import (
+    AvailableVersionsNotFoundError,
+    RemoteURLRequiredError,
+    UnsupportedAPIVersionError,
+    CollectionNotFound,
+)
 from pulp_ansible.app.models import (
     AnsibleCollectionDeprecated,
     AnsibleNamespace,
@@ -228,7 +233,7 @@ def sync(remote_pk, repository_pk, mirror, optimize):
     is_repo_remote = repository.remote is not None and remote.pk == repository.remote.pk
 
     if not remote.url:
-        raise ValueError(_("A CollectionRemote must have a 'url' specified to synchronize."))
+        raise RemoteURLRequiredError()
 
     first_stage = CollectionSyncFirstStage(remote, repository, is_repo_remote, optimize)
     if first_stage.should_sync:
@@ -594,14 +599,14 @@ async def _get_root_api(self, root):
             api_data = parse_metadata(await downloader.run())
 
         if "available_versions" not in api_data:
-            raise RuntimeError(_("Could not find 'available_versions' at {}").format(root))
+            raise AvailableVersionsNotFoundError(root)
 
         if "v3" in api_data.get("available_versions", {}):
             api_version = 3
         elif "v2" in api_data.get("available_versions", {}):
             api_version = 2
         else:
-            raise RuntimeError(_("Unsupported API versions at {}").format(root))
+            raise UnsupportedAPIVersionError(root)
 
         endpoint = f"{root}v{api_version}"
 

diff --git a/pulp_ansible/app/tasks/git.py b/pulp_ansible/app/tasks/git.py
@@ -12,6 +12,7 @@
     ArtifactSaver,
     RemoteArtifactSaver,
 )
+from pulp_ansible.exceptions import RemoteURLRequiredError
 from pulp_ansible.app.models import AnsibleRepository, GitRemote
 from pulp_ansible.app.tasks.collections import (
     declarative_content_from_git_repo,
@@ -46,7 +47,7 @@ def synchronize(remote_pk, repository_pk, mirror=False):
     repository = AnsibleRepository.objects.get(pk=repository_pk)
 
     if not remote.url:
-        raise ValueError(_("A remote must have a url specified to synchronize."))
+        raise RemoteURLRequiredError()
 
     log.info(
         _("Synchronizing: repository=%(r)s remote=%(p)s"), {"r": repository.name, "p": remote.name}

diff --git a/pulp_ansible/app/tasks/roles.py b/pulp_ansible/app/tasks/roles.py
@@ -13,6 +13,7 @@
     Stage,
 )
 from pulp_ansible.app.constants import PAGE_SIZE
+from pulp_ansible.exceptions import RemoteURLRequiredError
 from pulp_ansible.app.models import AnsibleRepository, RoleRemote, Role
 from pulp_ansible.app.tasks.utils import get_api_version, get_page_url, parse_metadata
 
@@ -48,7 +49,7 @@ def synchronize(remote_pk, repository_pk, mirror=False):
     repository = AnsibleRepository.objects.get(pk=repository_pk)
 
     if not remote.url:
-        raise ValueError(_("A remote must have a url specified to synchronize."))
+        raise RemoteURLRequiredError()
 
     log.info(
         _("Synchronizing: repository=%(r)s remote=%(p)s"), {"r": repository.name, "p": remote.name}

diff --git a/pulp_ansible/app/tasks/signature.py b/pulp_ansible/app/tasks/signature.py
@@ -26,7 +26,7 @@
 from pulpcore.plugin.util import gpg_verify
 from pulpcore.plugin.exceptions import InvalidSignatureError
 from pulp_ansible.app.tasks.utils import get_file_obj_from_tarball
-from rest_framework import serializers
+from pulp_ansible.exceptions import SignatureVerificationError
 
 log = logging.getLogger(__name__)
 
@@ -50,13 +50,9 @@ def verify_signature_upload(data):
                 verified = gpg_verify(gpgkey, file, manifest_file.name)
             except InvalidSignatureError as e:
                 if gpgkey:
-                    raise serializers.ValidationError(
-                        _("Signature verification failed: {}").format(e.verified.status)
-                    )
+                    raise SignatureVerificationError(str(e.verified.status))
                 elif settings.ANSIBLE_SIGNATURE_REQUIRE_VERIFICATION:
-                    raise serializers.ValidationError(
-                        _("Signature verification failed: No key available.")
-                    )
+                    raise SignatureVerificationError("No key available.")
                 else:
                     # We have no key configured. So we simply accept the signature as is
                     verified = e.verified

diff --git a/pulp_ansible/app/tasks/utils.py b/pulp_ansible/app/tasks/utils.py
@@ -5,11 +5,20 @@
 import re
 import yaml
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
-from rest_framework.serializers import ValidationError
 from yaml.error import YAMLError
 
 from galaxy_importer.schema import MAX_LENGTH_NAME, MAX_LENGTH_VERSION
 from pulp_ansible.app.constants import PAGE_SIZE
+from pulp_ansible.exceptions import (
+    APIVersionNotFoundError,
+    CollectionFieldTooLongError,
+    CollectionNameRequiredError,
+    InvalidCollectionFilenameError,
+    InvalidCollectionNameFormatError,
+    InvalidCollectionVersionError,
+    InvalidRequirementsFormatError,
+    RequirementsFileParseError,
+)
 
 log = logging.getLogger(__name__)
 
@@ -46,25 +55,20 @@ def parse_collection_filename(filename):
     match = FILENAME_REGEXP.match(filename)
 
     if not match:
-        msg = _("Invalid filename {filename}. Expected format: namespace-name-version.tar.gz")
-        raise ValueError(msg.format(filename=filename))
+        raise InvalidCollectionFilenameError(filename)
 
     namespace, name, version = match.groups()
 
     match = VERSION_REGEXP.match(version)
     if not match:
-        msg = _(
-            "Invalid version string {version} from filename {filename}. "
-            "Expected semantic version format."
-        )
-        raise ValueError(msg.format(version=version, filename=filename))
+        raise InvalidCollectionVersionError(version, filename)
 
     if len(namespace) > MAX_LENGTH_NAME:
-        raise ValueError(_("Expected namespace to be max length of %s") % MAX_LENGTH_NAME)
+        raise CollectionFieldTooLongError("namespace", MAX_LENGTH_NAME)
     if len(name) > MAX_LENGTH_NAME:
-        raise ValueError(_("Expected name to be max length of %s") % MAX_LENGTH_NAME)
+        raise CollectionFieldTooLongError("name", MAX_LENGTH_NAME)
     if len(version) > MAX_LENGTH_VERSION:
-        raise ValueError(_("Expected version to be max length of %s") % MAX_LENGTH_VERSION)
+        raise CollectionFieldTooLongError("version", MAX_LENGTH_VERSION)
 
     return CollectionFilename(namespace, name, version)
 
@@ -73,7 +77,7 @@ def get_api_version(url):
     """Get API version."""
     result = re.findall(r"/v(\d)/", url)
     if len(result) == 0:
-        raise RuntimeError(f"Could not determine API version for: {url}")
+        raise APIVersionNotFoundError(url)
     return int(result[0])
 
 
@@ -132,27 +136,20 @@ def parse_collections_requirements_file(requirements_file_string):
             try:
                 requirements = yaml.safe_load(requirements_file_string)
             except YAMLError as err:
-                raise ValidationError(
-                    _(
-                        "Failed to parse the collection requirements yml: {file} "
-                        "with the following error: {error}".format(
-                            file=requirements_file_string, error=err
-                        )
-                    )
-                )
+                raise RequirementsFileParseError(requirements_file_string, str(err))
         else:
             requirements = requirements_file_string
 
         if not isinstance(requirements, dict) or "collections" not in requirements:
-            raise ValidationError(
+            raise InvalidRequirementsFormatError(
                 _(
                     "Expecting collections requirements file to be a dict with the key "
                     "collections that contains a list of collections to install."
                 )
             )
 
         if not isinstance(requirements["collections"], list):
-            raise ValidationError(
+            raise InvalidRequirementsFormatError(
                 _(
                     "Expecting collections requirements file to be a dict with the key "
                     "collections that contains a list of collections to install."
@@ -163,9 +160,7 @@ def parse_collections_requirements_file(requirements_file_string):
             if isinstance(collection_req, dict):
                 req_name = collection_req.get("name", None)
                 if req_name is None:
-                    raise ValidationError(
-                        _("Collections requirement entry should contain the key name.")
-                    )
+                    raise CollectionNameRequiredError()
 
                 req_version = collection_req.get("version", "*")
                 req_source = collection_req.get("source", None)
@@ -174,12 +169,7 @@ def parse_collections_requirements_file(requirements_file_string):
             else:
                 entry = RequirementsFileEntry(name=collection_req, version="*", source=None)
             if "." not in entry.name:
-                raise ValidationError(
-                    _(
-                        "Collections requirement entry should contain the collection name in the "
-                        "format namespace.name"
-                    )
-                )
+                raise InvalidCollectionNameFormatError()
             collection_info.append(entry)
 
     return collection_info