use dedicated classes as input for add registry credentials method

Author: tamirse

datacrunch/containers/__init__.py

@@ -21,4 +21,10 @@
     Secret,
     RegistryCredential,
     ContainersService,
+    BaseRegistryCredentials,
+    DockerHubCredentials,
+    GithubCredentials,
+    GCRCredentials,
+    AWSECRCredentials,
+    CustomRegistryCredentials,
 )

datacrunch/containers/containers.py

@@ -1,5 +1,5 @@
-from dataclasses import dataclass, field
-from dataclasses_json import dataclass_json, config, Undefined  # type: ignore
+from dataclasses import dataclass
+from dataclasses_json import dataclass_json, Undefined  # type: ignore
 from typing import List, Optional, Dict
 from enum import Enum
 
@@ -275,6 +275,79 @@ class RegistryCredential:
     created_at: str
 
 
+@dataclass_json
+@dataclass
+class BaseRegistryCredentials:
+    """Base class for registry credentials"""
+    name: str
+    type: ContainerRegistryType
+
+
+@dataclass_json
+@dataclass
+class DockerHubCredentials(BaseRegistryCredentials):
+    """Credentials for DockerHub registry"""
+    username: str
+    access_token: str
+
+    def __init__(self, name: str, username: str, access_token: str):
+        super().__init__(name=name, type=ContainerRegistryType.DOCKERHUB)
+        self.username = username
+        self.access_token = access_token
+
+
+@dataclass_json
+@dataclass
+class GithubCredentials(BaseRegistryCredentials):
+    """Credentials for GitHub Container Registry"""
+    username: str
+    access_token: str
+
+    def __init__(self, name: str, username: str, access_token: str):
+        super().__init__(name=name, type=ContainerRegistryType.GITHUB)
+        self.username = username
+        self.access_token = access_token
+
+
+@dataclass_json
+@dataclass
+class GCRCredentials(BaseRegistryCredentials):
+    """Credentials for Google Container Registry"""
+    service_account_key: str
+
+    def __init__(self, name: str, service_account_key: str):
+        super().__init__(name=name, type=ContainerRegistryType.GCR)
+        self.service_account_key = service_account_key
+
+
+@dataclass_json
+@dataclass
+class AWSECRCredentials(BaseRegistryCredentials):
+    """Credentials for AWS Elastic Container Registry"""
+    access_key_id: str
+    secret_access_key: str
+    region: str
+    ecr_repo: str
+
+    def __init__(self, name: str, access_key_id: str, secret_access_key: str, region: str, ecr_repo: str):
+        super().__init__(name=name, type=ContainerRegistryType.AWS_ECR)
+        self.access_key_id = access_key_id
+        self.secret_access_key = secret_access_key
+        self.region = region
+        self.ecr_repo = ecr_repo
+
+
+@dataclass_json
+@dataclass
+class CustomRegistryCredentials(BaseRegistryCredentials):
+    """Credentials for custom container registries"""
+    docker_config_json: str
+
+    def __init__(self, name: str, docker_config_json: str):
+        super().__init__(name=name, type=ContainerRegistryType.CUSTOM)
+        self.docker_config_json = docker_config_json
+
+
 class ContainersService:
     """Service for managing container deployments"""
 
@@ -581,73 +654,13 @@ def get_registry_credentials(self) -> List[RegistryCredential]:
         response = self.client.get(CONTAINER_REGISTRY_CREDENTIALS_ENDPOINT)
         return [RegistryCredential.from_dict(credential) for credential in response.json()]
 
-    def add_registry_credentials(
-        self,
-        name: str,
-        registry_type: ContainerRegistryType,
-        username: str = None,
-        access_token: str = None,
-        service_account_key: str = None,
-        docker_config_json: str = None,
-        access_key_id: str = None,
-        secret_access_key: str = None,
-        region: str = None,
-        ecr_repo: str = None
-    ) -> None:
+    def add_registry_credentials(self, credentials: BaseRegistryCredentials) -> None:
         """Add registry credentials
 
-        :param name: name of the credentials
-        :type name: str
-        :param registry_type: type of registry (e.g. ContainerRegistryType.DOCKERHUB)
-        :type registry_type: ContainerRegistryType
-        :param username: registry username (required for DOCKERHUB and GITHUB)
-        :type username: str
-        :param access_token: registry access token (required for DOCKERHUB and GITHUB)
-        :type access_token: str
-        :param service_account_key: service account key JSON string (required for GCR)
-        :type service_account_key: str
-        :param docker_config_json: docker config JSON string (required for CUSTOM)
-        :type docker_config_json: str
-        :param access_key_id: AWS access key ID (required for AWS_ECR)
-        :type access_key_id: str
-        :param secret_access_key: AWS secret access key (required for AWS_ECR)
-        :type secret_access_key: str
-        :param region: AWS region (required for AWS_ECR)
-        :type region: str
-        :param ecr_repo: ECR repository URL (required for AWS_ECR)
-        :type ecr_repo: str
+        :param credentials: Registry credentials object
+        :type credentials: BaseRegistryCredentials
         """
-        data = {
-            "name": name,
-            "type": registry_type.value
-        }
-
-        # Add specific parameters based on registry type
-        if registry_type == ContainerRegistryType.DOCKERHUB or registry_type == ContainerRegistryType.GITHUB:
-            if not username or not access_token:
-                raise ValueError(
-                    f"Username and access_token are required for {registry_type.value} registry type")
-            data["username"] = username
-            data["access_token"] = access_token
-        elif registry_type == ContainerRegistryType.GCR:
-            if not service_account_key:
-                raise ValueError(
-                    "service_account_key is required for GCR registry type")
-            data["service_account_key"] = service_account_key
-        elif registry_type == ContainerRegistryType.AWS_ECR:
-            if not all([access_key_id, secret_access_key, region, ecr_repo]):
-                raise ValueError(
-                    "access_key_id, secret_access_key, region, and ecr_repo are required for AWS_ECR registry type")
-            data["access_key_id"] = access_key_id
-            data["secret_access_key"] = secret_access_key
-            data["region"] = region
-            data["ecr_repo"] = ecr_repo
-        elif registry_type == ContainerRegistryType.CUSTOM:
-            if not docker_config_json:
-                raise ValueError(
-                    "docker_config_json is required for CUSTOM registry type")
-            data["docker_config_json"] = docker_config_json
-
+        data = credentials.to_dict()
         self.client.post(CONTAINER_REGISTRY_CREDENTIALS_ENDPOINT, data)
 
     def delete_registry_credentials(self, credentials_name: str) -> None:

examples/containers/registry_credentials_example.py

@@ -1,6 +1,12 @@
 import os
 from datacrunch import DataCrunchClient
-from datacrunch.containers.containers import ContainerRegistryType
+from datacrunch.containers import (
+    DockerHubCredentials,
+    GithubCredentials,
+    GCRCredentials,
+    AWSECRCredentials,
+    CustomRegistryCredentials
+)
 
 # Environment variables
 DATACRUNCH_CLIENT_ID = os.environ.get('DATACRUNCH_CLIENT_ID')
@@ -11,21 +17,21 @@
                                      client_secret=DATACRUNCH_CLIENT_SECRET)
 
 # Example 1: DockerHub Credentials
-datacrunch_client.containers.add_registry_credentials(
+dockerhub_creds = DockerHubCredentials(
     name="my-dockerhub-creds",
-    registry_type=ContainerRegistryType.DOCKERHUB,
     username="your-dockerhub-username",
     access_token="your-dockerhub-access-token"
 )
+datacrunch_client.containers.add_registry_credentials(dockerhub_creds)
 print("Created DockerHub credentials")
 
 # Example 2: GitHub Container Registry Credentials
-datacrunch_client.containers.add_registry_credentials(
+github_creds = GithubCredentials(
     name="my-github-creds",
-    registry_type=ContainerRegistryType.GITHUB,
     username="your-github-username",
     access_token="your-github-token"
 )
+datacrunch_client.containers.add_registry_credentials(github_creds)
 print("Created GitHub credentials")
 
 # Example 3: Google Container Registry (GCR) Credentials
@@ -43,22 +49,22 @@
   "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/your-service-account%40your-project.iam.gserviceaccount.com"
 }"""
 
-datacrunch_client.containers.add_registry_credentials(
+gcr_creds = GCRCredentials(
     name="my-gcr-creds",
-    registry_type=ContainerRegistryType.GCR,
     service_account_key=gcr_service_account_key
 )
+datacrunch_client.containers.add_registry_credentials(gcr_creds)
 print("Created GCR credentials")
 
 # Example 4: AWS ECR Credentials
-datacrunch_client.containers.add_registry_credentials(
+aws_creds = AWSECRCredentials(
     name="my-aws-ecr-creds",
-    registry_type=ContainerRegistryType.AWS_ECR,
     access_key_id="AKIAEXAMPLE123456",
     secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
     region="eu-north-1",
     ecr_repo="887841266746.dkr.ecr.eu-north-1.amazonaws.com"
 )
+datacrunch_client.containers.add_registry_credentials(aws_creds)
 print("Created AWS ECR credentials")
 
 # Example 5: Custom Registry Credentials
@@ -70,11 +76,11 @@
   }
 }"""
 
-datacrunch_client.containers.add_registry_credentials(
+custom_creds = CustomRegistryCredentials(
     name="my-custom-registry-creds",
-    registry_type=ContainerRegistryType.CUSTOM,
     docker_config_json=custom_docker_config
 )
+datacrunch_client.containers.add_registry_credentials(custom_creds)
 print("Created Custom registry credentials")
 
 # Delete all registry credentials

tests/unit_tests/containers/test_containers.py

@@ -27,6 +27,10 @@
     ScalingTriggers,
     QueueLoadScalingTrigger,
     UtilizationScalingTrigger,
+    DockerHubCredentials,
+    GCRCredentials,
+    AWSECRCredentials,
+    CustomRegistryCredentials,
 )
 from datacrunch.exceptions import APIException
 
@@ -740,27 +744,17 @@ def test_add_registry_credentials(self, containers_service, registry_credentials
         )
 
         # act
-        containers_service.add_registry_credentials(
-            REGISTRY_CREDENTIAL_NAME,
-            ContainerRegistryType.DOCKERHUB,
-            "username",
-            "token"
+        creds = DockerHubCredentials(
+            name=REGISTRY_CREDENTIAL_NAME,
+            username="username",
+            access_token="token"
         )
+        containers_service.add_registry_credentials(creds)
 
         # assert
         assert responses.assert_call_count(
             registry_credentials_endpoint, 1) is True
-
-    @responses.activate
-    def test_add_registry_credentials_validation_error(self, containers_service):
-        # act & assert
-        with pytest.raises(ValueError) as excinfo:
-            containers_service.add_registry_credentials(
-                REGISTRY_CREDENTIAL_NAME,
-                ContainerRegistryType.DOCKERHUB,
-                # Missing username and token
-            )
-        assert "Username and access_token are required" in str(excinfo.value)
+        assert responses.calls[0].request.body == '{"name": "test-credential", "registry_type": "dockerhub", "username": "username", "access_token": "token"}'
 
     @responses.activate
     def test_add_registry_credentials_gcr(self, containers_service, registry_credentials_endpoint):
@@ -773,15 +767,16 @@ def test_add_registry_credentials_gcr(self, containers_service, registry_credent
 
         # act
         service_account_key = '{"key": "value"}'
-        containers_service.add_registry_credentials(
-            REGISTRY_CREDENTIAL_NAME,
-            ContainerRegistryType.GCR,
+        creds = GCRCredentials(
+            name=REGISTRY_CREDENTIAL_NAME,
             service_account_key=service_account_key
         )
+        containers_service.add_registry_credentials(creds)
 
         # assert
         assert responses.assert_call_count(
             registry_credentials_endpoint, 1) is True
+        assert responses.calls[0].request.body == '{"name": "test-credential", "registry_type": "gcr", "service_account_key": {"key": "value"}}'
 
     @responses.activate
     def test_add_registry_credentials_aws_ecr(self, containers_service, registry_credentials_endpoint):
@@ -793,18 +788,19 @@ def test_add_registry_credentials_aws_ecr(self, containers_service, registry_cre
         )
 
         # act
-        containers_service.add_registry_credentials(
-            REGISTRY_CREDENTIAL_NAME,
-            ContainerRegistryType.AWS_ECR,
+        creds = AWSECRCredentials(
+            name=REGISTRY_CREDENTIAL_NAME,
             access_key_id="test-key",
             secret_access_key="test-secret",
             region="us-west-2",
             ecr_repo="test.ecr.aws.com"
         )
+        containers_service.add_registry_credentials(creds)
 
         # assert
         assert responses.assert_call_count(
             registry_credentials_endpoint, 1) is True
+        assert responses.calls[0].request.body == '{"name": "test-credential", "registry_type": "aws-ecr", "access_key_id": "test-key", "secret_access_key": "test-secret", "region": "us-west-2", "ecr_repo": "test.ecr.aws.com"}'
 
     @responses.activate
     def test_add_registry_credentials_custom(self, containers_service, registry_credentials_endpoint):
@@ -817,15 +813,16 @@ def test_add_registry_credentials_custom(self, containers_service, registry_cred
 
         # act
         docker_config = '{"auths": {"registry.example.com": {"auth": "base64-encoded"}}}'
-        containers_service.add_registry_credentials(
-            REGISTRY_CREDENTIAL_NAME,
-            ContainerRegistryType.CUSTOM,
+        creds = CustomRegistryCredentials(
+            name=REGISTRY_CREDENTIAL_NAME,
             docker_config_json=docker_config
         )
+        containers_service.add_registry_credentials(creds)
 
         # assert
         assert responses.assert_call_count(
             registry_credentials_endpoint, 1) is True
+        assert responses.calls[0].request.body == '{"name": "test-credential", "registry_type": "custom", "docker_config_json": {"auths": {"registry.example.com": {"auth": "base64-encoded"}}}}'
 
     @responses.activate
     def test_delete_registry_credentials(self, containers_service, registry_credentials_endpoint):