fix: resolve task stage relationship parsing

Author: TanyaSingh369-svg

examples/task_stage_example.py

@@ -32,7 +32,7 @@ def main():
     print("\nReading task stage...")
     try:
         stage = client.task_stages.read(task_stage_id)
-        print(f"ID: {stage.id}, Status: {stage.status}")
+        print(stage)
     except Exception as e:
         print(f"Read failed: {e}")
 
@@ -41,7 +41,8 @@ def main():
     try:
         stages = list(client.task_stages.list(run_id))
         for s in stages:
-            print(f"{s.id} - {s.status}")
+            print(s)
+            print("-" * 80)
     except Exception as e:
         print(f"List failed: {e}")
 

src/pytfe/models/task_result.py

@@ -36,26 +36,29 @@ class TaskResultStatusTimestamps(BaseModel):
 class TaskResult(BaseModel):
     model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
 
+    # All non-id fields are optional so JSON:API relationship references
+    # ({id, type} only) can be hydrated as TaskResult stubs without raising
+    # "field required" validation errors.
     id: str
-    status: TaskResultStatus = Field(..., alias="status")
-    message: str = Field(..., alias="message")
+    status: TaskResultStatus | None = Field(None, alias="status")
+    message: str | None = Field(None, alias="message")
 
-    status_timestamps: TaskResultStatusTimestamps = Field(
-        ..., alias="status-timestamps"
+    status_timestamps: TaskResultStatusTimestamps | None = Field(
+        None, alias="status-timestamps"
     )
 
-    url: str = Field(..., alias="url")
+    url: str | None = Field(None, alias="url")
 
-    created_at: datetime = Field(..., alias="created-at")
-    updated_at: datetime = Field(..., alias="updated-at")
+    created_at: datetime | None = Field(None, alias="created-at")
+    updated_at: datetime | None = Field(None, alias="updated-at")
 
-    task_id: str = Field(..., alias="task-id")
-    task_name: str = Field(..., alias="task-name")
-    task_url: str = Field(..., alias="task-url")
+    task_id: str | None = Field(None, alias="task-id")
+    task_name: str | None = Field(None, alias="task-name")
+    task_url: str | None = Field(None, alias="task-url")
 
-    workspace_task_id: str = Field(..., alias="workspace-task-id")
-    workspace_task_enforcement_level: TaskEnforcementLevel = Field(
-        ..., alias="workspace-task-enforcement-level"
+    workspace_task_id: str | None = Field(None, alias="workspace-task-id")
+    workspace_task_enforcement_level: TaskEnforcementLevel | None = Field(
+        None, alias="workspace-task-enforcement-level"
     )
 
     agent_pool_id: str | None = Field(None, alias="agent-pool-id")

src/pytfe/models/task_stage.py

@@ -9,9 +9,11 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from pytfe.models.policy_evaluation import PolicyEvaluation
+from pytfe.models.task_result import TaskResult
+
 if TYPE_CHECKING:
-    from pytfe.models.policy_evaluation import PolicyEvaluation
-    from pytfe.models.task_result import TaskResult
+    from pytfe.models.run import Run
 
 
 class Stage(str, Enum):
@@ -56,36 +58,65 @@ class Actions(BaseModel):
     is_overridable: bool | None = Field(None, alias="is-overridable")
 
 
-# TaskStage represents a HCP Terraform or Terraform Enterprise run's stage
 class TaskStage(BaseModel):
     model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
 
     id: str
 
     stage: Stage = Field(..., alias="stage")
-    status: TaskStageStatus = Field(..., alias="status")
-    status_timestamps: TaskStageStatusTimestamps = Field(..., alias="status-timestamps")
+
+    status: TaskStageStatus = Field(
+        ...,
+        alias="status",
+    )
+
+    status_timestamps: TaskStageStatusTimestamps = Field(
+        ...,
+        alias="status-timestamps",
+    )
 
     created_at: datetime = Field(..., alias="created-at")
     updated_at: datetime = Field(..., alias="updated-at")
 
-    permissions: Permissions | None = Field(None, alias="permissions")
-    actions: Actions | None = Field(None, alias="actions")
+    permissions: Permissions | None = Field(
+        None,
+        alias="permissions",
+    )
 
-    run: dict | None = Field(None, alias="run")
-    task_results: list[TaskResult] | None = Field(None, alias="task-results")
-    policy_evaluations: list[PolicyEvaluation] | None = Field(
-        None, alias="policy-evaluations"
+    actions: Actions | None = Field(
+        None,
+        alias="actions",
     )
 
+    # Relationships
+    run: Run | None = Field(
+        None,
+        alias="run",
+    )
 
-def _rebuild_task_stage_model() -> None:
-    global TaskResult, PolicyEvaluation
+    task_results: list[TaskResult] | None = Field(
+        None,
+        alias="task-results",
+    )
 
-    from pytfe.models.policy_evaluation import PolicyEvaluation
-    from pytfe.models.task_result import TaskResult
+    policy_evaluations: list[PolicyEvaluation] | None = Field(
+        None,
+        alias="policy-evaluations",
+    )
 
-    TaskStage.model_rebuild()
+
+def _rebuild_task_stage_model() -> None:
+    # Do not import Run here: run.py imports TaskStage, so importing Run during
+    # TaskStage module initialization creates a circular import on Python 3.14.
+    TaskStage.model_rebuild(
+        # Leave unresolved cyclic refs (Run) for later resolution while still
+        # resolving non-cyclic refs needed during import-time schema generation.
+        raise_errors=False,
+        _types_namespace={
+            "TaskResult": TaskResult,
+            "PolicyEvaluation": PolicyEvaluation,
+        },
+    )
 
 
 _rebuild_task_stage_model()

src/pytfe/resources/task_stage.py

@@ -7,6 +7,9 @@
 from typing import Any
 
 from ..errors import InvalidRunIDError
+from ..models.policy_evaluation import PolicyEvaluation
+from ..models.run import Run
+from ..models.task_result import TaskResult
 from ..models.task_stage import TaskStage
 from ..utils import _safe_str, valid_string_id
 from ._base import _Service
@@ -16,22 +19,44 @@ class TaskStages(_Service):
     """TaskStages provides access to task stage endpoints."""
 
     def _parse_task_stage(self, data: dict[str, Any]) -> TaskStage:
+        # TaskStage defers Run resolution to avoid model import-time cycles.
+        # Rebuild here where Run is already imported and fully available.
+        TaskStage.model_rebuild(
+            raise_errors=False,
+            _types_namespace={"Run": Run},
+        )
+
         attributes = data.get("attributes", {})
 
         attributes["id"] = _safe_str(data.get("id"))
 
         relationships = data.get("relationships", {})
 
         run_data = relationships.get("run", {}).get("data")
-        attributes["run"] = run_data
+        if run_data:
+            attributes["run"] = Run.model_validate(run_data)
 
-        task_results_data = relationships.get("task-results", {}).get("data", [])
-        attributes["task-results"] = task_results_data
+        task_results_data = relationships.get("task-results", {}).get(
+            "data",
+            [],
+        )
 
-        policy_evaluations_data = relationships.get("policy-evaluations", {}).get(
-            "data", []
+        attributes["task-results"] = [
+            TaskResult.model_validate(task_result) for task_result in task_results_data
+        ]
+
+        policy_evaluations_data = relationships.get(
+            "policy-evaluations",
+            {},
+        ).get(
+            "data",
+            [],
         )
-        attributes["policy-evaluations"] = policy_evaluations_data
+
+        attributes["policy-evaluations"] = [
+            PolicyEvaluation.model_validate(policy_evaluation)
+            for policy_evaluation in policy_evaluations_data
+        ]
 
         return TaskStage.model_validate(attributes)