From 736b017215d22639d846600ac7a94d7d7eed6985 Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Mon, 22 Sep 2025 15:11:36 +0530 Subject: [PATCH] registry provider functions --- examples/registry_provider_individual.py | 305 ++++++++++++++++++++++ src/tfe/client.py | 2 + src/tfe/errors.py | 1 + src/tfe/models/__init__.py | 21 ++ src/tfe/models/registry_provider_types.py | 100 +++++++ src/tfe/resources/_base.py | 6 +- src/tfe/resources/registry_provider.py | 194 ++++++++++++++ 7 files changed, 627 insertions(+), 2 deletions(-) create mode 100644 examples/registry_provider_individual.py create mode 100644 src/tfe/models/registry_provider_types.py create mode 100644 src/tfe/resources/registry_provider.py diff --git a/examples/registry_provider_individual.py b/examples/registry_provider_individual.py new file mode 100644 index 0000000..bf26faf --- /dev/null +++ b/examples/registry_provider_individual.py @@ -0,0 +1,305 @@ +#!/usr/bin/env python3 +""" +Registry Provider Individual Function Tests + +This file provides individual test functions for each registry provider operation. +You can run specific functions to test individual parts of the API. + +Functions available: +- test_list_simple() - Basic list test +- test_create_private() - Create a private provider +- test_create_public() - Create a public provider +- test_read_with_id() - Read a provider by ID +- test_delete_by_id() - Delete a provider by ID + +Usage: + python registry_provider_individual.py +""" + +import os +import random +import sys + +# Add the src directory to the path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from tfe import TFEClient +from tfe.models.registry_provider_types import ( + RegistryName, + RegistryProviderCreateOptions, + RegistryProviderID, + RegistryProviderIncludeOps, + RegistryProviderListOptions, + RegistryProviderReadOptions, +) + + +def get_client_and_org(): + """Initialize client and get organization name.""" + client = TFEClient() + organization_name = os.getenv("TFE_ORGANIZATION", "aayush-test") + return client, organization_name + + +def test_list_simple(): + """Test 1: Simple list of registry providers.""" + print("=== Test 1: List Registry Providers ===") + + client, org = get_client_and_org() + + try: + providers = list(client.registry_providers.list(org)) + print(f"✓ Found {len(providers)} providers in organization '{org}'") + + for i, provider in enumerate(providers[:5], 1): + print(f" {i}. {provider.name}") + print(f" Namespace: {provider.namespace}") + print(f" Registry: {provider.registry_name.value}") + print(f" ID: {provider.id}") + print(f" Can Delete: {provider.permissions.can_delete}") + print() + + return providers + + except Exception as e: + print(f"✗ Error: {e}") + return [] + + +def test_list_with_options(): + """Test 2: List with filtering options.""" + print("=== Test 2: List with Options ===") + + client, org = get_client_and_org() + + try: + # Test with search + options = RegistryProviderListOptions( + search="test", registry_name=RegistryName.PRIVATE, page_size=5 + ) + + providers = list(client.registry_providers.list(org, options)) + print(f"✓ Found {len(providers)} providers matching search 'test'") + + # Test with include + include_options = RegistryProviderListOptions( + include=[RegistryProviderIncludeOps.REGISTRY_PROVIDER_VERSIONS] + ) + + detailed_providers = list(client.registry_providers.list(org, include_options)) + print(f"✓ Found {len(detailed_providers)} providers with version details") + + return providers + + except Exception as e: + print(f"✗ Error: {e}") + return [] + + +def test_create_private(): + """Test 3: Create a private registry provider.""" + print("=== Test 3: Create Private Provider ===") + + client, org = get_client_and_org() + + try: + provider_name = f"test-provider-{random.randint(100000, 999999)}" + + options = RegistryProviderCreateOptions( + name=provider_name, + namespace=org, # For private providers, namespace = org name + registry_name=RegistryName.PRIVATE, + ) + + provider = client.registry_providers.create(org, options) + print(f"✓ Created private provider: {provider.name}") + print(f" ID: {provider.id}") + print(f" Namespace: {provider.namespace}") + print(f" Registry: {provider.registry_name.value}") + print(f" Created: {provider.created_at}") + + return provider + + except Exception as e: + print(f"✗ Error creating private provider: {e}") + return None + + +def test_create_public(): + """Test 4: Create a public registry provider.""" + print("=== Test 4: Create Public Provider ===") + + client, org = get_client_and_org() + + try: + provider_name = f"test-provider-{random.randint(100000, 999999)}" + namespace_name = f"test-namespace-{random.randint(1000, 9999)}" + + options = RegistryProviderCreateOptions( + name=provider_name, + namespace=namespace_name, + registry_name=RegistryName.PUBLIC, + ) + + provider = client.registry_providers.create(org, options) + print(f"✓ Created public provider: {provider.name}") + print(f" ID: {provider.id}") + print(f" Namespace: {provider.namespace}") + print(f" Registry: {provider.registry_name.value}") + print(f" Created: {provider.created_at}") + + return provider + + except Exception as e: + print(f"✗ Error creating public provider: {e}") + return None + + +def test_read_with_id(provider_data): + """Test 5: Read a provider by ID.""" + print("=== Test 5: Read Provider by ID ===") + + client, org = get_client_and_org() + + if not provider_data: + print("⚠️ No provider data provided") + return None + + try: + provider_id = RegistryProviderID( + organization_name=org, + registry_name=provider_data.registry_name, + namespace=provider_data.namespace, + name=provider_data.name, + ) + + # Basic read + provider = client.registry_providers.read(provider_id) + print(f"✓ Read provider: {provider.name}") + print(f" ID: {provider.id}") + print(f" Namespace: {provider.namespace}") + print(f" Registry: {provider.registry_name.value}") + print(f" Created: {provider.created_at}") + print(f" Updated: {provider.updated_at}") + print(f" Can Delete: {provider.permissions.can_delete}") + + # Read with options + options = RegistryProviderReadOptions( + include=[RegistryProviderIncludeOps.REGISTRY_PROVIDER_VERSIONS] + ) + + detailed_provider = client.registry_providers.read(provider_id, options) + print(f"✓ Read with options: {detailed_provider.name}") + + if detailed_provider.registry_provider_versions: + print( + f" Found {len(detailed_provider.registry_provider_versions)} versions" + ) + else: + print(" No versions found") + + return provider + + except Exception as e: + print(f"✗ Error reading provider: {e}") + return None + + +def test_delete_by_id(provider_data): + """Test 6: Delete a provider by ID.""" + print("=== Test 6: Delete Provider by ID ===") + + client, org = get_client_and_org() + + if not provider_data: + print("⚠️ No provider data provided") + return False + + try: + provider_id = RegistryProviderID( + organization_name=org, + registry_name=provider_data.registry_name, + namespace=provider_data.namespace, + name=provider_data.name, + ) + + # Verify provider exists + provider = client.registry_providers.read(provider_id) + print(f"✓ Found provider to delete: {provider.name}") + + # Delete the provider + client.registry_providers.delete(provider_id) + print("✓ Successfully called delete() for provider") + + # Verify deletion (optional - may take time) + import time + + time.sleep(2) + + try: + client.registry_providers.read(provider_id) + print("⚠️ Provider still exists (deletion may take time)") + except Exception: + print("✓ Provider successfully deleted") + + return True + + except Exception as e: + print(f"✗ Error deleting provider: {e}") + return False + + +def main(): + """Run all tests in sequence.""" + print("🚀 REGISTRY PROVIDER INDIVIDUAL TESTS") + print("=" * 50) + + # Test 1: List providers + providers = test_list_simple() + print() + + # Test 2: List with options + test_list_with_options() + print() + + # ⚠️ WARNING: Uncomment the following tests to create/delete providers + print("⚠️ WARNING: Creation and deletion tests are commented out for safety") + print("⚠️ Uncomment them in the code to test creation and deletion") + print() + + # UNCOMMENT TO TEST CREATION: + # Test 3: Create private provider + private_provider = test_create_private() + print() + + # Test 4: Create public provider + public_provider = test_create_public() + print() + + # Test 5: Read provider + if private_provider: + test_read_with_id(private_provider) + print() + + # Test 6: Delete provider (UNCOMMENT TO TEST DELETION) + if private_provider: + test_delete_by_id(private_provider) + print() + + if public_provider: + test_delete_by_id(public_provider) + print() + + # Test with existing provider if available + if providers: + print("=== Testing with Existing Provider ===") + existing_provider = providers[0] + test_read_with_id(existing_provider) + print() + + print("✅ Individual tests completed!") + print("💡 To test creation/deletion, uncomment the relevant sections in the code") + + +if __name__ == "__main__": + main() diff --git a/src/tfe/client.py b/src/tfe/client.py index 1bea139..e5c8cef 100644 --- a/src/tfe/client.py +++ b/src/tfe/client.py @@ -5,6 +5,7 @@ from .resources.organizations import Organizations from .resources.projects import Projects from .resources.registry_module import RegistryModules +from .resources.registry_provider import RegistryProviders from .resources.run_task import RunTasks from .resources.run_trigger import RunTriggers from .resources.state_version_outputs import StateVersionOutputs @@ -35,6 +36,7 @@ def __init__(self, config: TFEConfig | None = None): self.variables = Variables(self._transport) self.workspaces = Workspaces(self._transport) self.registry_modules = RegistryModules(self._transport) + self.registry_providers = RegistryProviders(self._transport) self.state_versions = StateVersions(self._transport) self.state_version_outputs = StateVersionOutputs(self._transport) diff --git a/src/tfe/errors.py b/src/tfe/errors.py index 8b6d82c..84eaf3c 100644 --- a/src/tfe/errors.py +++ b/src/tfe/errors.py @@ -66,6 +66,7 @@ class ErrStateVersionUploadNotSupported(TFEError): ... ERR_INVALID_PROVIDER = "invalid value for provider" ERR_REQUIRED_VERSION = "version is required" ERR_INVALID_VERSION = "invalid value for version" +ERR_INVALID_NAMESPACE = "invalid value for namespace" ERR_REQUIRED_NAMESPACE = "namespace is required" ERR_INVALID_REGISTRY_NAME = "invalid registry name" ERR_UNSUPPORTED_BOTH_NAMESPACE_AND_PRIVATE_REGISTRY_NAME = ( diff --git a/src/tfe/models/__init__.py b/src/tfe/models/__init__.py index f176a1f..15fe592 100644 --- a/src/tfe/models/__init__.py +++ b/src/tfe/models/__init__.py @@ -37,6 +37,18 @@ TestConfig, ) +# Re-export all registry provider types +from .registry_provider_types import ( + RegistryProvider, + RegistryProviderCreateOptions, + RegistryProviderID, + RegistryProviderIncludeOps, + RegistryProviderList, + RegistryProviderListOptions, + RegistryProviderPermissions, + RegistryProviderReadOptions, +) + # Define what should be available when importing with * __all__ = [ # Registry module types @@ -69,6 +81,15 @@ "Root", "TestConfig", "TerraformRegistryModule", + # Registry provider types + "RegistryProvider", + "RegistryProviderCreateOptions", + "RegistryProviderID", + "RegistryProviderIncludeOps", + "RegistryProviderList", + "RegistryProviderListOptions", + "RegistryProviderPermissions", + "RegistryProviderReadOptions", # Main types from types.py (will be dynamically added below) "Capacity", "DataRetentionPolicy", diff --git a/src/tfe/models/registry_provider_types.py b/src/tfe/models/registry_provider_types.py new file mode 100644 index 0000000..3ac3140 --- /dev/null +++ b/src/tfe/models/registry_provider_types.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import Any + +from pydantic import BaseModel, Field + + +class RegistryName(Enum): + """Registry name enumeration.""" + + PRIVATE = "private" + PUBLIC = "public" + + +class RegistryProviderIncludeOps(Enum): + """Registry provider include operations.""" + + REGISTRY_PROVIDER_VERSIONS = "registry-provider-versions" + + +class RegistryProviderPermissions(BaseModel): + """Registry provider permissions.""" + + can_delete: bool = Field(alias="can-delete") + + model_config = {"populate_by_name": True} + + +class RegistryProvider(BaseModel): + """Registry provider model.""" + + id: str + name: str + namespace: str + created_at: datetime = Field(alias="created-at") + updated_at: datetime = Field(alias="updated-at") + registry_name: RegistryName = Field(alias="registry-name") + permissions: RegistryProviderPermissions + + # Relations + organization: dict[str, Any] | None = None + registry_provider_versions: list[dict[str, Any]] | None = Field( + alias="registry-provider-versions", default=None + ) + + # Links + links: dict[str, Any] | None = None + + model_config = {"populate_by_name": True} + + +class RegistryProviderID(BaseModel): + """Registry provider identifier.""" + + organization_name: str + registry_name: RegistryName + namespace: str + name: str + + +class RegistryProviderCreateOptions(BaseModel): + """Options for creating a registry provider.""" + + name: str + namespace: str + registry_name: RegistryName = Field(alias="registry-name") + + model_config = {"populate_by_name": True} + + +class RegistryProviderReadOptions(BaseModel): + """Options for reading a registry provider.""" + + include: list[RegistryProviderIncludeOps] | None = None + + +class RegistryProviderListOptions(BaseModel): + """Options for listing registry providers.""" + + registry_name: RegistryName | None = Field( + alias="filter[registry_name]", default=None + ) + organization_name: str | None = Field( + alias="filter[organization_name]", default=None + ) + search: str | None = Field(alias="q", default=None) + include: list[RegistryProviderIncludeOps] | None = None + page_number: int | None = Field(alias="page[number]", default=None) + page_size: int | None = Field(alias="page[size]", default=None) + + model_config = {"populate_by_name": True} + + +class RegistryProviderList(BaseModel): + """Registry provider list response.""" + + items: list[RegistryProvider] + pagination: dict[str, Any] | None = None diff --git a/src/tfe/resources/_base.py b/src/tfe/resources/_base.py index c14ebba..a0eaa71 100644 --- a/src/tfe/resources/_base.py +++ b/src/tfe/resources/_base.py @@ -27,7 +27,8 @@ def _list( data = json_response.get("data", []) yield from data - if len(data) < int(p["page[size]"]): + page_size = int(p["page[size]"]) + if len(data) < page_size: break page += 1 @@ -53,6 +54,7 @@ async def _alist( data = r.json().get("data", []) for item in data: yield item - if len(data) < p["page[size]"]: + page_size = int(p["page[size]"]) + if len(data) < page_size: break page += 1 diff --git a/src/tfe/resources/registry_provider.py b/src/tfe/resources/registry_provider.py new file mode 100644 index 0000000..3c1fd22 --- /dev/null +++ b/src/tfe/resources/registry_provider.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..errors import ( + ERR_INVALID_ORG, +) +from ..models.registry_provider_types import ( + RegistryName, + RegistryProvider, + RegistryProviderCreateOptions, + RegistryProviderID, + RegistryProviderListOptions, + RegistryProviderPermissions, + RegistryProviderReadOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class RegistryProviders(_Service): + """Registry providers service for managing Terraform registry providers.""" + + def list( + self, organization: str, options: RegistryProviderListOptions | None = None + ) -> Iterator[RegistryProvider]: + """List all the registry providers within an organization.""" + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + path = f"/api/v2/organizations/{organization}/registry-providers" + params = {} + + if options: + if options.include: + params["include"] = ",".join([opt.value for opt in options.include]) + if options.search: + params["q"] = options.search + if options.registry_name: + params["filter[registry_name]"] = options.registry_name.value + if options.organization_name: + params["filter[organization_name]"] = options.organization_name + if options.page_number: + params["page[number]"] = str(options.page_number) + if options.page_size: + params["page[size]"] = str(options.page_size) + + for item in self._list(path, params=params): + if item is None: + continue # type: ignore[unreachable] # Skip None items + yield self._parse_registry_provider(item) + + def create( + self, organization: str, options: RegistryProviderCreateOptions + ) -> RegistryProvider: + """Create a registry provider.""" + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + if not self._validate_create_options(options): + raise ValueError("Invalid create options") + + path = f"/api/v2/organizations/{organization}/registry-providers" + + # Prepare the data payload + data = { + "data": { + "type": "registry-providers", + "attributes": { + "name": options.name, + "namespace": options.namespace, + "registry-name": options.registry_name.value, + }, + } + } + + response = self.t.request("POST", path, json_body=data) + response_data = response.json() + return self._parse_registry_provider(response_data["data"]) + + def read( + self, + provider_id: RegistryProviderID, + options: RegistryProviderReadOptions | None = None, + ) -> RegistryProvider: + """Read a specific registry provider.""" + if not self._validate_provider_id(provider_id): + raise ValueError("Invalid provider ID") + + path = ( + f"/api/v2/organizations/{provider_id.organization_name}/" + f"registry-providers/{provider_id.registry_name.value}/" + f"{provider_id.namespace}/{provider_id.name}" + ) + + params = {} + if options and options.include: + params["include"] = ",".join([opt.value for opt in options.include]) + + response = self.t.request("GET", path, params=params) + response_data = response.json() + return self._parse_registry_provider(response_data["data"]) + + def delete(self, provider_id: RegistryProviderID) -> None: + """Delete a registry provider.""" + if not self._validate_provider_id(provider_id): + raise ValueError("Invalid provider ID") + + path = ( + f"/api/v2/organizations/{provider_id.organization_name}/" + f"registry-providers/{provider_id.registry_name.value}/" + f"{provider_id.namespace}/{provider_id.name}" + ) + + self.t.request("DELETE", path) + + def _validate_provider_id(self, provider_id: RegistryProviderID) -> bool: + """Validate a registry provider ID.""" + if not valid_string_id(provider_id.organization_name): + return False + if not valid_string_id(provider_id.name): + return False + if not valid_string_id(provider_id.namespace): + return False + if provider_id.registry_name not in [RegistryName.PRIVATE, RegistryName.PUBLIC]: + return False + return True + + def _validate_create_options(self, options: RegistryProviderCreateOptions) -> bool: + """Validate create options.""" + if not valid_string_id(options.name): + return False + if not valid_string_id(options.namespace): + return False + if options.registry_name not in [RegistryName.PRIVATE, RegistryName.PUBLIC]: + return False + return True + + def _parse_registry_provider(self, data: dict[str, Any]) -> RegistryProvider: + """Parse a registry provider from API response data.""" + if data is None: + raise ValueError("Cannot parse registry provider: data is None") + + attributes = data.get("attributes", {}) + relationships = data.get("relationships", {}) + + # Parse timestamps + created_at = attributes.get("created-at") + updated_at = attributes.get("updated-at") + + # Parse permissions + permissions_data = attributes.get("permissions", {}) + permissions = RegistryProviderPermissions( + **{"can-delete": permissions_data.get("can-delete", False)} + ) + + # Parse relationships + organization = None + if "organization" in relationships: + org_data = relationships["organization"].get("data") + if org_data: + organization = {"id": org_data.get("id"), "type": org_data.get("type")} + + registry_provider_versions = None + if "registry-provider-versions" in relationships: + versions_data = relationships["registry-provider-versions"].get("data", []) + registry_provider_versions = [ + {"id": v.get("id"), "type": v.get("type")} for v in versions_data + ] + + # Parse registry name + registry_name_str = attributes.get("registry-name", "private") + registry_name = ( + RegistryName.PRIVATE + if registry_name_str == "private" + else RegistryName.PUBLIC + ) + + # Create the provider data dict with aliases + provider_data = { + "id": data.get("id", ""), + "name": attributes.get("name", ""), + "namespace": attributes.get("namespace", ""), + "created-at": created_at, + "updated-at": updated_at, + "registry-name": registry_name, + "permissions": permissions, + "organization": organization, + "registry-provider-versions": registry_provider_versions, + "links": data.get("links"), + } + + return RegistryProvider(**provider_data)