#130 Standardize Notification Configuration option models on Pydantic

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

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,

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" }

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",

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,99 +252,70 @@ 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
 
         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,44 +326,38 @@ 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
 
         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