diff --git a/CHANGELOG.md b/CHANGELOG.md index 4db63b0d..55e5164c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Unreleased +# Released +# v0.1.5 + +* `pytfe.__version__` added in src/pytfe/init.py via importlib.metadata.version("pytfe"). This will resolve to the version from pyproject.toml. +* Updated comments, sshkey, stateversion and cost-estimate models to have id as mandatory attribute by @isivaselvan [#137](https://github.com/hashicorp/python-tfe/pull/137) +* Updated workspace resource to include additional relationship models include AgentPool, Configuration-version, Run, Variables and State-version by @isivaselvan [#138](https://github.com/hashicorp/python-tfe/pull/138) + +## Bug Fixes +* Run.read / Run.create fail with pydantic ValidationError when response has a `cost-estimate` and `comments` relationship. + +# 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 07e1b857..3e333220 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 c8986fbd..4f311fc4 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.5" 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/__init__.py b/src/pytfe/__init__.py index 68cea566..9d518c48 100644 --- a/src/pytfe/__init__.py +++ b/src/pytfe/__init__.py @@ -1,8 +1,16 @@ # Copyright IBM Corp. 2025, 2026 # SPDX-License-Identifier: MPL-2.0 +from importlib.metadata import PackageNotFoundError +from importlib.metadata import version as _pkg_version + from . import errors, models from .client import TFEClient from .config import TFEConfig -__all__ = ["TFEConfig", "TFEClient", "errors", "models"] +try: + __version__ = _pkg_version("pytfe") +except PackageNotFoundError: # running from a source checkout without install + __version__ = "0.0.0+unknown" + +__all__ = ["TFEConfig", "TFEClient", "errors", "models", "__version__"] diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 0f1435d8..7aa184f0 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/comment.py b/src/pytfe/models/comment.py index da2cd213..19cc25ca 100644 --- a/src/pytfe/models/comment.py +++ b/src/pytfe/models/comment.py @@ -10,4 +10,4 @@ class Comment(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str - body: str = Field(..., alias="body") + body: str = Field(default="", alias="body") diff --git a/src/pytfe/models/cost_estimate.py b/src/pytfe/models/cost_estimate.py index d1b6ff6b..4ae1c614 100644 --- a/src/pytfe/models/cost_estimate.py +++ b/src/pytfe/models/cost_estimate.py @@ -13,17 +13,17 @@ class CostEstimate(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str - delta_monthly_cost: str = Field(..., alias="delta-monthly-cost") - error_message: str = Field(..., alias="error-message") - matched_resources_count: int = Field(..., alias="matched-resources-count") - prior_monthly_cost: str = Field(..., alias="prior-monthly-cost") - proposed_monthly_cost: str = Field(..., alias="proposed-monthly-cost") - resources_count: int = Field(..., alias="resources-count") - status: CostEstimateStatus = Field(..., alias="status") - status_timestamps: CostEstimateStatusTimestamps = Field( - ..., alias="status-timestamps" + delta_monthly_cost: str = Field(default="", alias="delta-monthly-cost") + error_message: str = Field(default="", alias="error-message") + matched_resources_count: int = Field(default=0, alias="matched-resources-count") + prior_monthly_cost: str = Field(default="", alias="prior-monthly-cost") + proposed_monthly_cost: str = Field(default="", alias="proposed-monthly-cost") + resources_count: int = Field(default=0, alias="resources-count") + status: CostEstimateStatus | None = Field(default=None, alias="status") + status_timestamps: CostEstimateStatusTimestamps | None = Field( + default=None, alias="status-timestamps" ) - unmatched_resources_count: int = Field(..., alias="unmatched-resources-count") + unmatched_resources_count: int = Field(default=0, alias="unmatched-resources-count") class CostEstimateStatus(str, Enum): diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index e1af877a..c9b2934a 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -9,10 +9,13 @@ from __future__ import annotations +from collections.abc import Iterator from datetime import datetime from enum import Enum from typing import Any +from pydantic import BaseModel, ConfigDict, Field, field_validator + class NotificationTriggerType(Enum): """Represents the different TFE notifications that can be sent as a run's progress transitions between different states.""" @@ -47,209 +50,143 @@ class NotificationDestinationType(Enum): MICROSOFT_TEAMS = "microsoft-teams" -class DeliveryResponse: +class DeliveryResponse(BaseModel): """Represents a notification configuration delivery response.""" - # Type annotations for instance attributes - body: str - code: str - headers: dict[str, Any] - sent_at: datetime | None - successful: str - url: str - - def __init__(self, data: dict[str, Any]): - self.body = data.get("body", "") - self.code = data.get("code", "") - self.headers = data.get("headers", {}) - self.sent_at = self._parse_datetime(data.get("sent-at")) - self.successful = data.get("successful", "") - self.url = data.get("url", "") - - def _parse_datetime(self, date_str: str | None) -> datetime | None: - """Parse ISO 8601 datetime string.""" - if not date_str: - return None - try: - return datetime.fromisoformat(date_str.replace("Z", "+00:00")) - except (ValueError, AttributeError): - return None + model_config = ConfigDict(populate_by_name=True) - def __repr__(self) -> str: - return f"DeliveryResponse(url='{self.url}', code='{self.code}', successful='{self.successful}')" + body: str | None = None + code: str | None = None + headers: dict[str, Any] | None = Field(default_factory=dict) + sent_at: datetime | None = Field(default=None, alias="sent-at") + successful: str | None = None + url: str | None = None + def __init__(self, data: dict[str, Any] | None = None, /, **kwargs: Any) -> None: + if data is not None: + super().__init__(**{**data, **kwargs}) + else: + super().__init__(**kwargs) -class NotificationConfigurationSubscribableChoice: - """Choice type struct that represents the possible values within a polymorphic relation.""" - # Type annotations for instance attributes - team: Any | None - workspace: Any | None +class NotificationConfigurationSubscribableChoice(BaseModel): + """Choice type struct that represents the possible values within a polymorphic relation.""" - def __init__(self, team: Any | None = None, workspace: Any | None = None): - self.team = team - self.workspace = workspace + model_config = ConfigDict(arbitrary_types_allowed=True) - def __repr__(self) -> str: - if self.team: - return f"NotificationConfigurationSubscribableChoice(team={self.team})" - elif self.workspace: - return f"NotificationConfigurationSubscribableChoice(workspace={self.workspace})" - return "NotificationConfigurationSubscribableChoice()" + team: Any | None = None + workspace: Any | None = None -class NotificationConfiguration: +class NotificationConfiguration(BaseModel): """Represents a Notification Configuration.""" - # Type annotations for instance attributes - id: str | None - created_at: datetime | None - updated_at: datetime | None - destination_type: str | None - enabled: bool - name: str - token: str - url: str - triggers: list[NotificationTriggerType] - delivery_responses: list[Any] - email_addresses: list[str] - email_users: list[Any] - subscribable: Any - subscribable_choice: Any | None - - def __init__(self, data: dict[str, Any]): - self.id = data.get("id") - self.created_at = self._parse_datetime(data.get("created-at")) - self.updated_at = self._parse_datetime(data.get("updated-at")) - - # Core attributes - self.destination_type = data.get("destination-type") - self.enabled = data.get("enabled", False) - self.name = data.get("name", "") - self.token = data.get("token", "") - self.url = data.get("url", "") - - # Triggers - convert from strings to enum values - self.triggers = self._parse_triggers(data.get("triggers", [])) - - # Delivery responses - delivery_responses_data = data.get("delivery-responses", []) - self.delivery_responses = [ - DeliveryResponse(dr) for dr in delivery_responses_data - ] - - # Email configuration - self.email_addresses = data.get("email-addresses", []) - self.email_users = data.get("email-users", []) - - # Relationships - using polymorphic relation pattern - self.subscribable = data.get( - "subscribable" - ) # Deprecated but maintained for compatibility - self.subscribable_choice = self._parse_subscribable_choice( - data.get("subscribable-choice") - ) - - def _parse_datetime(self, date_str: str | None) -> datetime | None: - """Parse ISO 8601 datetime string.""" - if not date_str: - return None - try: - return datetime.fromisoformat(date_str.replace("Z", "+00:00")) - except (ValueError, AttributeError): - return None - - def _parse_triggers(self, triggers: list[str]) -> list[NotificationTriggerType]: - """Parse trigger strings to enum values.""" - parsed_triggers = [] - for trigger in triggers: + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) + + id: str | None = None + created_at: datetime | None = Field(default=None, alias="created-at") + updated_at: datetime | None = Field(default=None, alias="updated-at") + destination_type: str | None = Field(default=None, alias="destination-type") + enabled: bool = False + name: str | None = None + token: str | None = None + url: str | None = None + triggers: list[NotificationTriggerType] = Field(default_factory=list) + delivery_responses: list[DeliveryResponse] = Field( + default_factory=list, alias="delivery-responses" + ) + email_addresses: list[str] = Field(default_factory=list, alias="email-addresses") + email_users: list[Any] = Field(default_factory=list, alias="email-users") + subscribable: Any = None + subscribable_choice: NotificationConfigurationSubscribableChoice | None = Field( + default=None, alias="subscribable-choice" + ) + + @field_validator( + "delivery_responses", + "email_addresses", + "email_users", + mode="before", + ) + @classmethod + def _none_to_empty_list(cls, value: Any) -> Any: + return [] if value is None else value + + @field_validator("triggers", mode="before") + @classmethod + def _coerce_triggers(cls, value: Any) -> list[NotificationTriggerType]: + if not value: + return [] + parsed: list[NotificationTriggerType] = [] + for trigger in value: + if isinstance(trigger, NotificationTriggerType): + parsed.append(trigger) + continue try: - parsed_triggers.append(NotificationTriggerType(trigger)) - except ValueError: - # If trigger is not in enum, keep as string for backwards compatibility + parsed.append(NotificationTriggerType(trigger)) + except (ValueError, TypeError): + # Silently drop unknown triggers for backwards compatibility pass - return parsed_triggers - - def _parse_subscribable_choice( - self, choice_data: dict[str, Any] | None - ) -> NotificationConfigurationSubscribableChoice | None: - """Parse subscribable choice data.""" - if not choice_data: - return None - - team = choice_data.get("team") - workspace = choice_data.get("workspace") - return NotificationConfigurationSubscribableChoice( - team=team, workspace=workspace - ) - - def __repr__(self) -> str: - return f"NotificationConfiguration(id='{self.id}', name='{self.name}', enabled={self.enabled})" + return parsed + + def __init__(self, data: dict[str, Any] | None = None, /, **kwargs: Any) -> None: + if data is not None: + super().__init__(**{**data, **kwargs}) + else: + super().__init__(**kwargs) + + +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: +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 self.model_dump(by_alias=True, exclude_none=True) - return params - -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 +199,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 +211,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 +273,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,75 +281,71 @@ 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 -class NotificationConfigurationList: +class NotificationConfigurationList(BaseModel): """Represents a list of notification configurations with pagination.""" - # Type annotations for instance attributes - items: list[NotificationConfiguration] - current_page: int - page_size: int - prev_page: int | None - next_page: int | None - total_pages: int - total_count: int - - def __init__(self, data: dict[str, Any]): - self.items = [ - NotificationConfiguration(item.get("attributes", {})) - for item in data.get("data", []) - ] - - # Pagination metadata - meta = data.get("meta", {}) - pagination = meta.get("pagination", {}) - - self.current_page = pagination.get("current-page", 0) - self.page_size = pagination.get("page-size", 20) - self.prev_page = pagination.get("prev-page") - self.next_page = pagination.get("next-page") - self.total_pages = pagination.get("total-pages", 0) - self.total_count = pagination.get("total-count", 0) + model_config = ConfigDict(populate_by_name=True) + + items: list[NotificationConfiguration] = Field(default_factory=list) + current_page: int = 0 + page_size: int = 20 + prev_page: int | None = None + next_page: int | None = None + total_pages: int = 0 + total_count: int = 0 + + def __init__(self, data: dict[str, Any] | None = None, /, **kwargs: Any) -> None: + if data is None: + super().__init__(**kwargs) + return + + items_data = [item.get("attributes", {}) for item in data.get("data") or []] + pagination = (data.get("meta") or {}).get("pagination") or {} + parsed: dict[str, Any] = { + "items": items_data, + "current_page": pagination.get("current-page", 0), + "page_size": pagination.get("page-size", 20), + "prev_page": pagination.get("prev-page"), + "next_page": pagination.get("next-page"), + "total_pages": pagination.get("total-pages", 0), + "total_count": pagination.get("total-count", 0), + } + parsed.update(kwargs) + super().__init__(**parsed) def __len__(self) -> int: return len(self.items) - def __iter__(self) -> Any: + def __iter__(self) -> Iterator[NotificationConfiguration]: # type: ignore[override] return iter(self.items) def __getitem__(self, index: int) -> NotificationConfiguration: return self.items[index] - - def __repr__(self) -> str: - return f"NotificationConfigurationList(count={len(self.items)}, page={self.current_page}, total={self.total_count})" diff --git a/src/pytfe/models/ssh_key.py b/src/pytfe/models/ssh_key.py index 4df3d0a0..52423fb8 100644 --- a/src/pytfe/models/ssh_key.py +++ b/src/pytfe/models/ssh_key.py @@ -13,7 +13,7 @@ class SSHKey(BaseModel): id: str = Field(..., description="The unique identifier for this SSH key") type: str = Field(default="ssh-keys", description="The type of this resource") - name: str = Field(..., description="A name to identify the SSH key") + name: str = Field(default="", description="A name to identify the SSH key") class SSHKeyCreateOptions(BaseModel): diff --git a/src/pytfe/models/state_version.py b/src/pytfe/models/state_version.py index 4d8607dd..dab42619 100644 --- a/src/pytfe/models/state_version.py +++ b/src/pytfe/models/state_version.py @@ -32,7 +32,7 @@ class StateVersion(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str = Field(..., alias="id") - created_at: datetime = Field(..., alias="created-at") + created_at: datetime | None = Field(None, alias="created-at") hosted_state_download_url: str | None = Field( None, alias="hosted-state-download-url" ) diff --git a/src/pytfe/models/workspace.py b/src/pytfe/models/workspace.py index d24ab35d..e0be77a7 100644 --- a/src/pytfe/models/workspace.py +++ b/src/pytfe/models/workspace.py @@ -23,9 +23,13 @@ from ..utils import has_tags_regex_defined, is_valid_workspace_name, valid_string from .agent import AgentPool from .common import EffectiveTagBinding, Tag, TagBinding +from .configuration_version import ConfigurationVersion from .data_retention_policy import DataRetentionPolicyChoice from .organization import ExecutionMode, Organization from .project import Project +from .ssh_key import SSHKey +from .state_version import StateVersion +from .variable import Variable if TYPE_CHECKING: from .run import Run @@ -168,15 +172,17 @@ class Workspace(BaseModel): # Relations agent_pool: AgentPool | None = None # AgentPool object current_run: Run | None = None # Run object - current_state_version: Any | None = None # StateVersion object + current_state_version: StateVersion | None = None # StateVersion object organization: Organization | None = None project: Project | None = None - ssh_key: Any | None = None # SSHKey object + ssh_key: SSHKey | None = None # SSHKey object outputs: list[WorkspaceOutputs] = Field(default_factory=list) tags: list[Tag] = Field(default_factory=list) - current_configuration_version: Any | None = None # ConfigurationVersion object + current_configuration_version: ConfigurationVersion | None = ( + None # ConfigurationVersion object + ) locked_by: LockedByChoice | None = None - variables: list[Any] = Field(default_factory=list) # Variable objects + variables: list[Variable] = Field(default_factory=list) # Variable objects tag_bindings: list[TagBinding] = Field(default_factory=list) effective_tag_bindings: list[EffectiveTagBinding] = Field(default_factory=list) diff --git a/src/pytfe/resources/workspaces.py b/src/pytfe/resources/workspaces.py index 2bd2d9b3..1a6ac6cb 100644 --- a/src/pytfe/resources/workspaces.py +++ b/src/pytfe/resources/workspaces.py @@ -7,6 +7,8 @@ from collections.abc import Iterator from typing import Any +from pytfe.models.ssh_key import SSHKey + from ..errors import ( InvalidOrgError, InvalidSSHKeyIDError, @@ -19,11 +21,13 @@ WorkspaceMinimumLimitError, WorkspaceRequiredError, ) +from ..models.agent import AgentPool from ..models.common import ( EffectiveTagBinding, Tag, TagBinding, ) +from ..models.configuration_version import ConfigurationVersion from ..models.data_retention_policy import ( DataRetentionPolicy, DataRetentionPolicyChoice, @@ -34,6 +38,9 @@ ) from ..models.organization import Organization from ..models.project import Project +from ..models.run import Run +from ..models.state_version import StateVersion +from ..models.variable import Variable from ..models.workspace import ( ExecutionMode, LockedByChoice, @@ -181,7 +188,32 @@ def _ws_from(d: dict[str, Any]) -> Workspace: {"id": relationships["project"]["data"].get("id")} ) if relationships.get("ssh-key", {}).get("data"): - attr["ssh_key"] = relationships["ssh-key"]["data"].get("id") + attr["ssh_key"] = SSHKey.model_validate( + {"id": relationships["ssh-key"]["data"].get("id")} + ) + if relationships.get("agent-pool", {}).get("data"): + attr["agent_pools"] = AgentPool.model_validate( + {"id": relationships["agent-pool"]["data"].get("id")} + ) + if relationships.get("current-run", {}).get("data"): + attr["current_run"] = Run.model_validate( + {"id": relationships["current-run"]["data"].get("id")} + ) + if relationships.get("current-configuration-version", {}).get("data"): + attr["current_configuration_version"] = ConfigurationVersion.model_validate( + {"id": relationships["current-configuration-version"]["data"].get("id")} + ) + if relationships.get("vars", {}).get("data"): + attr["variables"] = [ + Variable.model_validate({"id": item.get("id")}) + for item in relationships["vars"]["data"] + if item.get("id") + ] + if relationships.get("current-state-version", {}).get("data"): + attr["current_state_version"] = StateVersion.model_validate( + {"id": relationships["current-state-version"]["data"].get("id")} + ) + attr["outputs"] = outputs attr["locked_by"] = locked_by attr["data_retention_policy_choice"] = data_retention_policy_choice