feat: add workspace run task resource and models

Author: isivaselvan

examples/workspace_run_task.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""
+Workspace Run Task Operations Example
+
+Demonstrates workspace run task operations:
+1. create() - Attach a run task to a workspace
+2. list() - List all workspace run tasks for a workspace
+3. read() - Read a workspace run task by ID
+4. update() - Update enforcement/stage settings
+5. delete() - Delete a workspace run task
+
+Prerequisites:
+- Set TFE_TOKEN and TFE_ADDRESS environment variables
+- Use valid workspace and run task IDs
+"""
+
+import argparse
+
+from pytfe import TFEClient, TFEConfig
+from pytfe.models import (
+    RunTask,
+    WorkspaceRunTaskCreateOptions,
+    WorkspaceRunTaskUpdateOptions,
+)
+
+
+def _find_matching_workspace_run_task(items, run_task_id: str):
+    for item in items:
+        if item.run_task and item.run_task.id == run_task_id:
+            return item
+
+    if len(items) == 1:
+        return items[0]
+
+    return None
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Workspace run task operations demo for python-tfe SDK"
+    )
+    parser.add_argument(
+        "--workspace-id",
+        required=True,
+        help="Workspace ID (example: ws-abc123)",
+    )
+    parser.add_argument(
+        "--run-task-id",
+        required=True,
+        help="Run task ID (example: task-abc123)",
+    )
+    parser.add_argument(
+        "--delete-existing",
+        action="store_true",
+        help="Allow delete() to remove a workspace run task that existed before this example ran",
+    )
+    args = parser.parse_args()
+
+    client = TFEClient(TFEConfig.from_env())
+
+    workspace_id = args.workspace_id
+    run_task_id = args.run_task_id
+
+    if workspace_id == "ws-xxxxxxxx" or run_task_id == "task-xxxxxxxx":
+        print("Please provide real IDs for --workspace-id and --run-task-id")
+        return
+
+    print("=" * 80)
+    print("WORKSPACE RUN TASK OPERATIONS")
+    print("=" * 80)
+
+    workspace_task = None
+    created_in_run = False
+
+    print("\n1. create()")
+    try:
+        create_options = WorkspaceRunTaskCreateOptions(
+            enforcement_level="advisory",
+            run_task=RunTask(id=run_task_id),
+            stages=["post_plan"],
+        )
+        workspace_task = client.workspace_run_tasks.create(workspace_id, create_options)
+        created_in_run = True
+        print(f"Created workspace run task: {workspace_task.id}")
+    except Exception as exc:
+        print(f"Create result: {exc}")
+
+    print("\n2. list()")
+    items = []
+    try:
+        items = list(client.workspace_run_tasks.list(workspace_id))
+        print(f"Found {len(items)} workspace run task(s)")
+        for item in items:
+            run_task = item.run_task.id if item.run_task else None
+            print(
+                f"- {item.id} run_task={run_task} enforcement={item.enforcement_level} stages={item.stages}"
+            )
+
+        if workspace_task is None:
+            workspace_task = _find_matching_workspace_run_task(items, run_task_id)
+            if workspace_task is not None:
+                print(f"Using existing workspace run task: {workspace_task.id}")
+    except Exception as exc:
+        print(f"List failed: {exc}")
+
+    print("\n3. read()")
+    if workspace_task is None:
+        print("Read skipped: no workspace run task ID available")
+    else:
+        try:
+            workspace_task = client.workspace_run_tasks.read(
+                workspace_id, workspace_task.id
+            )
+            print(
+                f"Read workspace run task: {workspace_task.id} enforcement={workspace_task.enforcement_level} stages={workspace_task.stages}"
+            )
+        except Exception as exc:
+            print(f"Read failed: {exc}")
+
+    print("\n4. update()")
+    if workspace_task is None:
+        print("Update skipped: no workspace run task ID available")
+    else:
+        try:
+            update_options = WorkspaceRunTaskUpdateOptions(
+                # enforcement_level="mandatory",
+                stages=["post_plan", "pre_plan"],
+            )
+            workspace_task = client.workspace_run_tasks.update(
+                workspace_id, workspace_task.id, update_options
+            )
+            print(
+                f"Updated workspace run task: {workspace_task.id} enforcement={workspace_task.enforcement_level} stages={workspace_task.stages}"
+            )
+        except Exception as exc:
+            print(f"Update failed: {exc}")
+
+    print("\n5. delete()")
+    if workspace_task is None:
+        print("Delete skipped: no workspace run task ID available")
+    elif not created_in_run and not args.delete_existing:
+        print(
+            "Delete skipped: workspace run task existed before this example. "
+            "Re-run with --delete-existing to delete it."
+        )
+    else:
+        try:
+            client.workspace_run_tasks.delete(workspace_id, workspace_task.id)
+            print(f"Deleted workspace run task: {workspace_task.id}")
+        except Exception as exc:
+            print(f"Delete failed: {exc}")
+
+    print("=" * 80)
+    print("WORKSPACE RUN TASK OPERATIONS COMPLETED")
+    print("=" * 80)
+
+
+if __name__ == "__main__":
+    main()

src/pytfe/client.py

@@ -52,6 +52,7 @@
 from .resources.variable import Variables
 from .resources.variable_sets import VariableSets, VariableSetVariables
 from .resources.workspace_resources import WorkspaceResourcesService
+from .resources.workspace_run_task import WorkspaceRunTasks
 from .resources.workspaces import Workspaces
 
 
@@ -103,6 +104,7 @@ def __init__(self, config: TFEConfig | None = None):
         self.variable_set_variables = VariableSetVariables(self._transport)
         self.workspaces = Workspaces(self._transport)
         self.workspace_resources = WorkspaceResourcesService(self._transport)
+        self.workspace_run_tasks = WorkspaceRunTasks(self._transport)
         self.registry_modules = RegistryModules(self._transport)
         self.registry_providers = RegistryProviders(self._transport)
         self.registry_provider_versions = RegistryProviderVersions(self._transport)

src/pytfe/errors.py

@@ -338,6 +338,13 @@ def __init__(self, message: str = 'category must be "task"'):
         super().__init__(message)
 
 
+class InvalidWorkspaceRunTaskIDError(InvalidValues):
+    """Raised when an invalid workspace run task ID is provided."""
+
+    def __init__(self, message: str = "invalid value for workspace run task ID"):
+        super().__init__(message)
+
+
 # Run Trigger errors
 class RequiredRunTriggerListOpsError(RequiredFieldMissing):
     """Raised when required run trigger list options are missing."""

src/pytfe/models/__init__.py

@@ -328,6 +328,10 @@
 from .run_task_integration import (
     TaskResultStatus as TaskResultCallbackStatus,
 )
+from .run_task_request import (
+    RunTaskRequest,
+    RunTaskRequestCapabilitites,
+)
 from .run_trigger import (
     RunTrigger,
     RunTriggerCreateOptions,
@@ -456,6 +460,12 @@
     WorkspaceResource,
     WorkspaceResourceListOptions,
 )
+from .workspace_run_task import (
+    WorkspaceRunTask,
+    WorkspaceRunTaskCreateOptions,
+    WorkspaceRunTaskListOptions,
+    WorkspaceRunTaskUpdateOptions,
+)
 
 # ── Public surface ────────────────────────────────────────────────────────────
 __all__ = [
@@ -684,6 +694,11 @@
     # Workspace Resources
     "WorkspaceResource",
     "WorkspaceResourceListOptions",
+    # Workspace Run Tasks
+    "WorkspaceRunTask",
+    "WorkspaceRunTaskListOptions",
+    "WorkspaceRunTaskCreateOptions",
+    "WorkspaceRunTaskUpdateOptions",
     "RunQueue",
     "ReadRunQueueOptions",
     # Runs
@@ -728,6 +743,9 @@
     "RunTaskCreateOptions",
     "RunTaskUpdateOptions",
     "RunTaskReadOptions",
+    # Run Task Request
+    "RunTaskRequest",
+    "RunTaskRequestCapabilitites",
     # Task Result
     "TaskResult",
     "TaskResultEnforcementLevel",

src/pytfe/models/run_task.py

@@ -4,23 +4,26 @@
 from __future__ import annotations
 
 from enum import Enum
+from typing import TYPE_CHECKING
 
 from pydantic import BaseModel, Field
 
 from ..models.common import Pagination
 from .agent import AgentPool
 from .organization import Organization
-from .workspace_run_task import WorkspaceRunTask
+
+if TYPE_CHECKING:
+    from .workspace_run_task import WorkspaceRunTask
 
 
 class RunTask(BaseModel):
     id: str
-    name: str
+    name: str | None = None
     description: str | None = None
-    url: str
-    category: str
+    url: str | None = None
+    category: str | None = None
     hmac_key: str | None = None
-    enabled: bool
+    enabled: bool | None = None
     global_configuration: GlobalRunTask | None = None
 
     agent_pool: AgentPool | None = None
@@ -41,10 +44,10 @@ class GlobalRunTaskOptions(BaseModel):
 
 
 class Stage(str, Enum):
-    PRE_PLAN = "pre-plan"
-    POST_PLAN = "post-plan"
-    PRE_APPLY = "pre-apply"
-    POST_APPLY = "post-apply"
+    PRE_PLAN = "pre_plan"
+    POST_PLAN = "post_plan"
+    PRE_APPLY = "pre_apply"
+    POST_APPLY = "post_apply"
 
 
 class TaskEnforcementLevel(str, Enum):