diff --git a/CHANGELOG.md b/CHANGELOG.md index 4db63b0..62a56dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Unreleased +# Released + +# v0.1.4 + +## Enhancements +* Standardize Notification Configuration option models on Pydantic [#132](https://github.com/hashicorp/python-tfe/pull/132) + # v0.1.3 ## Enhancements diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index 07e1b85..3e33322 100644 --- a/examples/notification_configuration.py +++ b/examples/notification_configuration.py @@ -10,13 +10,9 @@ """ import os -import sys - -# Add the src directory to the Python path so we can import the tfe module -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) from pytfe.client import TFEClient -from pytfe.models.notification_configuration import ( +from pytfe.models import ( NotificationConfigurationCreateOptions, NotificationConfigurationListOptions, NotificationConfigurationSubscribableChoice, diff --git a/pyproject.toml b/pyproject.toml index c8986fb..c50cdc5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "pytfe" -version = "0.1.3" +version = "0.1.4" description = "Official Python SDK for HashiCorp Terraform Cloud / Terraform Enterprise (TFE) API v2" readme = "README.md" license = { text = "MPL-2.0" } diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 0f1435d..7aa184f 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -59,6 +59,19 @@ DataRetentionPolicySetOptions, ) +# ── Notification Configurations ─────────────────────────────────────────────── +from .notification_configuration import ( + DeliveryResponse, + NotificationConfiguration, + NotificationConfigurationCreateOptions, + NotificationConfigurationList, + NotificationConfigurationListOptions, + NotificationConfigurationSubscribableChoice, + NotificationConfigurationUpdateOptions, + NotificationDestinationType, + NotificationTriggerType, +) + # ── OAuth ───────────────────────────────────────────────────────────────────── from .oauth_client import ( OAuthClient, @@ -376,6 +389,16 @@ # ── Public surface ──────────────────────────────────────────────────────────── __all__ = [ + # Notification configurations + "DeliveryResponse", + "NotificationConfiguration", + "NotificationConfigurationCreateOptions", + "NotificationConfigurationList", + "NotificationConfigurationListOptions", + "NotificationConfigurationSubscribableChoice", + "NotificationConfigurationUpdateOptions", + "NotificationDestinationType", + "NotificationTriggerType", # OAuth "OAuthClient", "OAuthClientAddProjectsOptions", diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index e1af877..be9f880 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -13,6 +13,8 @@ from enum import Enum from typing import Any +from pydantic import BaseModel, ConfigDict, Field + class NotificationTriggerType(Enum): """Represents the different TFE notifications that can be sent as a run's progress transitions between different states.""" @@ -187,69 +189,57 @@ def __repr__(self) -> str: return f"NotificationConfiguration(id='{self.id}', name='{self.name}', enabled={self.enabled})" -class NotificationConfigurationListOptions: +def _serialize_triggers( + triggers: list[NotificationTriggerType | str], +) -> list[str]: + """Serialize trigger enums or raw strings to their wire value.""" + return [t.value if isinstance(t, NotificationTriggerType) else t for t in triggers] + + +def _validate_triggers( + triggers: list[NotificationTriggerType | str], +) -> list[str]: + """Collect errors for any non-enum, non-known-string trigger entries.""" + errors: list[str] = [] + for trigger in triggers: + if isinstance(trigger, NotificationTriggerType): + continue + try: + NotificationTriggerType(trigger) + except ValueError: + errors.append(f"Invalid trigger type: {trigger}") + return errors + + +class NotificationConfigurationListOptions(BaseModel): """Represents the options for listing notification configurations.""" - # Type annotations for instance attributes - page_size: int | None - subscribable_choice: NotificationConfigurationSubscribableChoice | None + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) - def __init__( - self, - page_size: int | None = None, - subscribable_choice: NotificationConfigurationSubscribableChoice | None = None, - ): - self.page_size = page_size - self.subscribable_choice = subscribable_choice + page_size: int | None = Field(default=None, alias="page[size]") + subscribable_choice: NotificationConfigurationSubscribableChoice | None = Field( + default=None, exclude=True + ) def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" - params = {} - - if self.page_size is not None: - params["page[size]"] = self.page_size - - return params + return self.model_dump(by_alias=True, exclude_none=True) -class NotificationConfigurationCreateOptions: +class NotificationConfigurationCreateOptions(BaseModel): """Represents the options for creating a new notification configuration.""" - # Type annotations for instance attributes + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) + destination_type: NotificationDestinationType enabled: bool name: str - token: str | None - triggers: list[NotificationTriggerType] - url: str | None - email_addresses: list[str] - email_users: list[Any] - subscribable_choice: NotificationConfigurationSubscribableChoice | None - - def __init__( - self, - destination_type: NotificationDestinationType, - enabled: bool, - name: str, - token: str | None = None, - triggers: list[NotificationTriggerType] | None = None, - url: str | None = None, - email_addresses: list[str] | None = None, - email_users: list[Any] | None = None, - subscribable_choice: NotificationConfigurationSubscribableChoice | None = None, - ): - # Required fields - self.destination_type = destination_type - self.enabled = enabled - self.name = name - - # Optional fields - self.token = token - self.triggers = triggers or [] - self.url = url - self.email_addresses = email_addresses or [] - self.email_users = email_users or [] - self.subscribable_choice = subscribable_choice + token: str | None = None + triggers: list[NotificationTriggerType | str] = Field(default_factory=list) + url: str | None = None + email_addresses: list[str] = Field(default_factory=list) + email_users: list[Any] = Field(default_factory=list) + subscribable_choice: NotificationConfigurationSubscribableChoice | None = None def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" @@ -262,14 +252,11 @@ def to_dict(self) -> dict[str, Any]: }, } - # Add optional attributes if self.token is not None: data["attributes"]["token"] = self.token if self.triggers: - data["attributes"]["triggers"] = [ - trigger.value for trigger in self.triggers - ] + data["attributes"]["triggers"] = _serialize_triggers(self.triggers) if self.url is not None: data["attributes"]["url"] = self.url @@ -277,84 +264,58 @@ def to_dict(self) -> dict[str, Any]: if self.email_addresses: data["attributes"]["email-addresses"] = self.email_addresses - # Handle relationships if self.email_users: - data["relationships"] = data.get("relationships", {}) - data["relationships"]["users"] = { - "data": [ - { - "type": "users", - "id": user.id if hasattr(user, "id") else str(user), - } - for user in self.email_users - ] + data["relationships"] = { + "users": { + "data": [ + { + "type": "users", + "id": user.id if hasattr(user, "id") else str(user), + } + for user in self.email_users + ] + } } return data - def validate(self) -> list[str]: + def validate(self) -> list[str]: # type: ignore[override] """Validate the create options and return any errors.""" - errors = [] + errors: list[str] = [] - # Required field validation if not self.name or not self.name.strip(): errors.append("Name is required") - if not isinstance(self.enabled, bool): - errors.append("Enabled must be a boolean") # type: ignore[unreachable] - - # URL validation for certain destination types - if self.destination_type in [ + if self.destination_type in ( NotificationDestinationType.GENERIC, NotificationDestinationType.SLACK, NotificationDestinationType.MICROSOFT_TEAMS, - ]: + ): if not self.url: errors.append("URL is required for this destination type") - # Trigger validation - for trigger in self.triggers: - if not isinstance(trigger, NotificationTriggerType): - errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] + errors.extend(_validate_triggers(self.triggers)) return errors -class NotificationConfigurationUpdateOptions: +class NotificationConfigurationUpdateOptions(BaseModel): """Represents the options for updating an existing notification configuration.""" - # Type annotations for instance attributes - enabled: bool | None - name: str | None - token: str | None - triggers: list[NotificationTriggerType] | None - url: str | None - email_addresses: list[str] | None - email_users: list[Any] | None - - def __init__( - self, - enabled: bool | None = None, - name: str | None = None, - token: str | None = None, - triggers: list[NotificationTriggerType] | None = None, - url: str | None = None, - email_addresses: list[str] | None = None, - email_users: list[Any] | None = None, - ): - self.enabled = enabled - self.name = name - self.token = token - self.triggers = triggers - self.url = url - self.email_addresses = email_addresses - self.email_users = email_users + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) + + enabled: bool | None = None + name: str | None = None + token: str | None = None + triggers: list[NotificationTriggerType | str] | None = None + url: str | None = None + email_addresses: list[str] | None = None + email_users: list[Any] | None = None def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" data: dict[str, Any] = {"type": "notification-configurations", "attributes": {}} - # Add only specified attributes if self.enabled is not None: data["attributes"]["enabled"] = self.enabled @@ -365,9 +326,7 @@ def to_dict(self) -> dict[str, Any]: data["attributes"]["token"] = self.token if self.triggers is not None: - data["attributes"]["triggers"] = [ - trigger.value for trigger in self.triggers - ] + data["attributes"]["triggers"] = _serialize_triggers(self.triggers) if self.url is not None: data["attributes"]["url"] = self.url @@ -375,34 +334,30 @@ def to_dict(self) -> dict[str, Any]: if self.email_addresses is not None: data["attributes"]["email-addresses"] = self.email_addresses - # Handle relationships if self.email_users is not None: - data["relationships"] = data.get("relationships", {}) - data["relationships"]["users"] = { - "data": [ - { - "type": "users", - "id": user.id if hasattr(user, "id") else str(user), - } - for user in self.email_users - ] + data["relationships"] = { + "users": { + "data": [ + { + "type": "users", + "id": user.id if hasattr(user, "id") else str(user), + } + for user in self.email_users + ] + } } return data - def validate(self) -> list[str]: + def validate(self) -> list[str]: # type: ignore[override] """Validate the update options and return any errors.""" - errors = [] + errors: list[str] = [] - # Name validation (if provided) if self.name is not None and (not self.name or not self.name.strip()): errors.append("Name cannot be empty") - # Trigger validation (if provided) if self.triggers is not None: - for trigger in self.triggers: - if not isinstance(trigger, NotificationTriggerType): - errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] + errors.extend(_validate_triggers(self.triggers)) return errors