Refactor system validation

Author: eclipseo

pydepsdev/api.py

@@ -27,6 +27,12 @@
     DEFAULT_MAX_BACKOFF,
     DEFAULT_MAX_RETRIES,
     DEFAULT_TIMEOUT_DURATION,
+    SUPPORTED_SYSTEMS,
+    SUPPORTED_SYSTEMS_REQUIREMENTS,
+    SUPPORTED_SYSTEMS_DEPENDENCIES,
+    SUPPORTED_SYSTEMS_DEPENDENTS,
+    SUPPORTED_SYSTEMS_CAPABILITIES,
+    SUPPORTED_SYSTEMS_QUERY,
 )
 from .exceptions import APIError
 from .utils import encode_url_param, validate_hash, validate_system
@@ -362,8 +368,7 @@ async def get_requirements(
             ValueError: If system_name is not "NUGET".
             APIError: On request failure.
         """
-        if system_name.upper() != "NUGET":
-            raise ValueError("get_requirements is only for NuGet.")
+        validate_system(system_name, SUPPORTED_SYSTEMS_REQUIREMENTS)
         name_enc = encode_url_param(package_name)
         ver_enc = encode_url_param(version)
         return await self._get(
@@ -394,7 +399,7 @@ async def get_dependencies(
             ValueError: If system_name is invalid.
             APIError: On request failure.
         """
-        validate_system(system_name)
+        validate_system(system_name, SUPPORTED_SYSTEMS_DEPENDENCIES)
         name_enc = encode_url_param(package_name)
         ver_enc = encode_url_param(version)
         return await self._get(
@@ -429,7 +434,7 @@ async def get_dependents(
             ValueError: If system_name is invalid.
             APIError: On request failure.
         """
-        validate_system(system_name)
+        validate_system(system_name, SUPPORTED_SYSTEMS_DEPENDENTS)
         name_enc = encode_url_param(package_name)
         ver_enc = encode_url_param(version)
         return await self._get(
@@ -470,8 +475,7 @@ async def get_capabilities(
             ValueError: If system_name is not "GO".
             APIError: On request failure.
         """
-        if system_name.upper() != "GO":
-            raise ValueError("get_capabilities is only for GO.")
+        validate_system(system_name, SUPPORTED_SYSTEMS_CAPABILITIES)
         name_enc = encode_url_param(package_name)
         ver_enc = encode_url_param(version)
         return await self._get(
@@ -653,7 +657,7 @@ async def query_package_versions(
         if hash_type and hash_value:
             validate_hash(hash_type)
         if version_system:
-            validate_system(version_system)
+            validate_system(version_system, SUPPORTED_SYSTEMS_QUERY)
 
         params: Dict[str, str] = {}
         if hash_type and hash_value:

pydepsdev/constants.py

@@ -17,6 +17,11 @@
 
 BASE_URL = "https://api.deps.dev/v3alpha"
 SUPPORTED_SYSTEMS = ["GO", "RUBYGEMS", "NPM", "CARGO", "MAVEN", "PYPI", "NUGET"]
+SUPPORTED_SYSTEMS_REQUIREMENTS = ["RUBYGEMS", "NPM", "MAVEN", "NUGET"]
+SUPPORTED_SYSTEMS_DEPENDENCIES = ["NPM", "CARGO", "MAVEN", "PYPI"]
+SUPPORTED_SYSTEMS_DEPENDENTS = ["NPM", "CARGO", "MAVEN", "PYPI"]
+SUPPORTED_SYSTEMS_CAPABILITIES = ["GO"]
+SUPPORTED_SYSTEMS_QUERY = ["RUBYGEMS", "NPM", "CARGO", "MAVEN", "PYPI", "NUGET"]
 SUPPORTED_HASHES = ["MD5", "SHA1", "SHA256", "SHA512"]
 
 DEFAULT_MAX_RETRIES = 3

pydepsdev/utils.py

@@ -16,8 +16,15 @@
 #
 
 import urllib.parse
-from typing import NoReturn
-from .constants import SUPPORTED_SYSTEMS, SUPPORTED_HASHES
+from typing import Optional, Sequence
+from .constants import (
+    SUPPORTED_SYSTEMS,
+    SUPPORTED_SYSTEMS_REQUIREMENTS,
+    SUPPORTED_SYSTEMS_DEPENDENCIES,
+    SUPPORTED_SYSTEMS_DEPENDENTS,
+    SUPPORTED_SYSTEMS_CAPABILITIES,
+    SUPPORTED_SYSTEMS_QUERY,
+)
 
 
 def encode_url_param(param: str) -> str:
@@ -33,22 +40,26 @@ def encode_url_param(param: str) -> str:
     return urllib.parse.quote_plus(param)
 
 
-def validate_system(system: str) -> None:
+def validate_system(
+    system: str,
+    allowed_systems: Optional[Sequence[str]] = None,
+) -> None:
     """
     Validate that the given system identifier is supported.
+    If allowed_systems is None, we fall back to SUPPORTED_SYSTEMS.
 
     Args:
-        system (str): The package system name (e.g. "npm", "pypi") to validate.
+        system (str): e.g. "npm", "PYPI", etc.
+        allowed_systems (Optional[Sequence[str]]):
+             A specific SUPPORTED_* constant, or None for all.
 
     Raises:
-        ValueError: If `system` (case-insensitive) is not in SUPPORTED_SYSTEMS.
+        ValueError: if system.upper() not in allowed_systems.
     """
     normalized = system.upper()
-    if normalized not in SUPPORTED_SYSTEMS:
-        raise ValueError(
-            "This operation is currently only available for "
-            f"{', '.join(SUPPORTED_SYSTEMS)}."
-        )
+    allowed = allowed_systems or SUPPORTED_SYSTEMS
+    if normalized not in allowed:
+        raise ValueError(f"This operation is only available for: {', '.join(allowed)}")
 
 
 def validate_hash(hash_type: str) -> None: