Features providing Plan and Apply API specs

examples/apply.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import argparse
+import os
+
+from tfe import TFEClient, TFEConfig
+
+
+def _print_header(title: str):
+    print("\n" + "=" * 80)
+    print(title)
+    print("=" * 80)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Applies demo for python-tfe SDK")
+    parser.add_argument(
+        "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io")
+    )
+    parser.add_argument("--token", default=os.getenv("TFE_TOKEN", ""))
+    parser.add_argument("--apply-id", required=True, help="Apply ID to work with")
+    args = parser.parse_args()
+
+    cfg = TFEConfig(address=args.address, token=args.token)
+    client = TFEClient(cfg)
+
+    # 1) Read the apply details
+    _print_header("Reading Apply Details")
+    try:
+        apply = client.applies.read(args.apply_id)
+        print(f"Apply ID: {apply.id}")
+        print(f"Status: {apply.status}")
+        print(f"Resource Additions: {apply.resource_additions}")
+        print(f"Resource Changes: {apply.resource_changes}")
+        print(f"Resource Destructions: {apply.resource_destructions}")
+        print(f"Resource Imports: {apply.resource_imports}")
+        print(f"Created At: {apply.created_at}")
+        print(f"Status Timestamps: {apply.status_timestamps}")
+        print(f"Log Read URL: {apply.log_read_url}")
+        print(
+            f"Execution Details ID: {apply.execution_details.id if apply.execution_details else 'None'}"
+        )
+    except Exception as e:
+        print(f"Error reading apply: {e}")
+        return 1
+
+    print("\n" + "=" * 80)
+    print("Apply demo completed successfully!")
+    print("=" * 80)
+    return 0
+
+
+if __name__ == "__main__":
+    exit(main())

examples/plan.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+
+from tfe import TFEClient, TFEConfig
+
+
+def _print_header(title: str):
+    print("\n" + "=" * 80)
+    print(title)
+    print("=" * 80)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Plans demo for python-tfe SDK")
+    parser.add_argument(
+        "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io")
+    )
+    parser.add_argument("--token", default=os.getenv("TFE_TOKEN", ""))
+    parser.add_argument("--plan-id", required=True, help="Plan ID to work with")
+    parser.add_argument("--save-json", help="Path to save JSON output")
+    args = parser.parse_args()
+
+    cfg = TFEConfig(address=args.address, token=args.token)
+    client = TFEClient(cfg)
+
+    # 1) Read the plan details
+    _print_header("Reading Plan Details")
+    try:
+        plan = client.plans.read(args.plan_id)
+        print(f"Plan ID: {plan.id}")
+        print(f"Status: {plan.status}")
+        print(f"Has Changes: {plan.has_changes}")
+        print(f"Resource Additions: {plan.resource_additions}")
+        print(f"Resource Changes: {plan.resource_changes}")
+        print(f"Resource Destructions: {plan.resource_destructions}")
+        print(f"Resource Imports: {plan.resource_imports}")
+        print(f"Status Timestamps: {plan.status_timestamps}")
+        print(f"Log Read URL: {plan.log_read_url}")
+    except Exception as e:
+        print(f"Error reading plan: {e}")
+        return 1
+
+    # 2) Get JSON output if the plan has it
+    _print_header("Reading JSON Output")
+    try:
+        json_output = client.plans.read_json_output(args.plan_id)
+        print(
+            f"JSON Output Keys: {list(json_output.keys()) if isinstance(json_output, dict) else 'Not a dict'}"
+        )
+
+        if isinstance(json_output, dict):
+            # Print some key information from the JSON output
+            if "format_version" in json_output:
+                print(f"Format Version: {json_output['format_version']}")
+            if "terraform_version" in json_output:
+                print(f"Terraform Version: {json_output['terraform_version']}")
+            if "resource_changes" in json_output:
+                changes = json_output["resource_changes"]
+                print(f"Number of Resource Changes: {len(changes) if changes else 0}")
+
+                # Show first few resource changes
+                if changes:
+                    print("\nFirst few resource changes:")
+                    for i, change in enumerate(changes[:3]):
+                        action = change.get("change", {}).get("actions", [])
+                        address = change.get("address", "unknown")
+                        print(f"  {i + 1}. {address}: {action}")
+
+        # Save JSON output if requested
+        if args.save_json:
+            with open(args.save_json, "w") as f:
+                json.dump(json_output, f, indent=2, default=str)
+            print(f"\nJSON output saved to: {args.save_json}")
+
+    except Exception as e:
+        print(f"Error reading JSON output: {e}")
+
+    print("\n" + "=" * 80)
+    print("Plan demo completed successfully!")
+    print("=" * 80)
+    return 0
+
+
+if __name__ == "__main__":
+    exit(main())

src/tfe/client.py

@@ -2,7 +2,9 @@
 
 from ._http import HTTPTransport
 from .config import TFEConfig
+from .resources.apply import Applies
 from .resources.organizations import Organizations
+from .resources.plan import Plans
 from .resources.projects import Projects
 from .resources.registry_module import RegistryModules
 from .resources.registry_provider import RegistryProviders
@@ -33,6 +35,8 @@ def __init__(self, config: TFEConfig | None = None):
             proxies=cfg.proxies,
             ca_bundle=cfg.ca_bundle,
         )
+        self.applies = Applies(self._transport)
+        self.plans = Plans(self._transport)
         self.organizations = Organizations(self._transport)
         self.projects = Projects(self._transport)
         self.variables = Variables(self._transport)

src/tfe/errors.py

@@ -339,3 +339,19 @@ def __init__(
         message: str = "setting terraform-version is only valid when plan-only is set to true",
     ):
         super().__init__(message)
+
+
+# Plan errors
+class InvalidPlanIDError(InvalidValues):
+    """Raised when an invalid plan ID is provided."""
+
+    def __init__(self, message: str = "invalid value for plan ID"):
+        super().__init__(message)
+
+
+# Apply errors
+class InvalidApplyIDError(InvalidValues):
+    """Raised when an invalid apply ID is provided."""
+
+    def __init__(self, message: str = "invalid value for apply ID"):
+        super().__init__(message)

src/tfe/models/apply.py

@@ -6,35 +6,39 @@
 from pydantic import BaseModel, ConfigDict, Field
 
 
+class ApplyStatus(str, Enum):
+    APPLY_CANCELED = "canceled"
+    APPLY_CREATED = "created"
+    APPLY_ERRORED = "errored"
+    APPLY_FINISHED = "finished"
+    APPLY_MFA_WAITING = "mfa_waiting"
+    APPLY_PENDING = "pending"
+    APPLY_QUEUED = "queued"
+    APPLY_RUNNING = "running"
+    APPLY_UNREACHABLE = "unreachable"
+
+
 class Apply(BaseModel):
     model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
 
     id: str
     log_read_url: str | None = Field(None, alias="log-read-url")
-    raiseesource_additions: int = Field(..., alias="resource-additions")
-    resource_changes: int = Field(..., alias="resource-changes")
-    resource_destructions: int = Field(..., alias="resource-destructions")
-    status: ApplyStatus = Field(..., alias="status")
-    status_timestamps: ApplyStatusTimestamps = Field(..., alias="status-timestamps")
-
-
-class ApplyStatus(str, Enum):
-    Apply_Canceled = "canceled"
-    Apply_Created = "created"
-    Apply_Errored = "errored"
-    Apply_Finished = "finished"
-    Apply_MFA_Waiting = "mfa_waiting"
-    Apply_Pending = "pending"
-    Apply_Queued = "queued"
-    Apply_Running = "running"
-    Apply_Unreachable = "unreachable"
+    resource_additions: int | None = Field(None, alias="resource-additions")
+    resource_changes: int | None = Field(None, alias="resource-changes")
+    resource_destructions: int | None = Field(None, alias="resource-destructions")
+    resource_imports: int | None = Field(None, alias="resource-imports")
+    status: ApplyStatus | None = Field(None, alias="status")
+    status_timestamps: ApplyStatusTimestamps | None = Field(
+        None, alias="status-timestamps"
+    )
 
 
 class ApplyStatusTimestamps(BaseModel):
     model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
-    canceled_at: datetime = Field(..., alias="canceled-at")
-    errored_at: datetime = Field(..., alias="errored-at")
-    finished_at: datetime = Field(..., alias="finished-at")
-    force_canceled_at: datetime = Field(..., alias="force-canceled-at")
-    queued_at: datetime = Field(..., alias="queued-at")
-    started_at: datetime = Field(..., alias="started-at")
+
+    canceled_at: datetime | None = Field(None, alias="canceled-at")
+    errored_at: datetime | None = Field(None, alias="errored-at")
+    finished_at: datetime | None = Field(None, alias="finished-at")
+    force_canceled_at: datetime | None = Field(None, alias="force-canceled-at")
+    queued_at: datetime | None = Field(None, alias="queued-at")
+    started_at: datetime | None = Field(None, alias="started-at")

src/tfe/models/plan.py

@@ -5,43 +5,49 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from ..models.plan_export import PlanExport
+
 
 class PlanStatus(str, Enum):
-    Plan_Canceled = "canceled"
-    Plan_Created = "created"
-    Plan_Errored = "errored"
-    Plan_Finished = "finished"
-    Plan_MFA_Waiting = "mfa_waiting"
-    Plan_Pending = "pending"
-    Plan_Queued = "queued"
-    Plan_Running = "running"
-    Plan_Unreachable = "unreachable"
+    """The status of a plan."""
+
+    PLAN_CANCELED = "canceled"
+    PLAN_CREATED = "created"
+    PLAN_ERRORED = "errored"
+    PLAN_FINISHED = "finished"
+    PLAN_MFA_WAITING = "mfa_waiting"
+    PLAN_PENDING = "pending"
+    PLAN_QUEUED = "queued"
+    PLAN_RUNNING = "running"
+    PLAN_UNREACHABLE = "unreachable"
 
 
 class Plan(BaseModel):
     model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
 
     id: str
-    has_changes: bool = Field(..., alias="has-changes")
-    generated_configuration: bool = Field(..., alias="generated-configuration")
-    log_read_url: str = Field(..., alias="log-read-url")
-    resource_additions: int = Field(..., alias="resource-additions")
-    resource_changes: int = Field(..., alias="resource-changes")
-    resource_destructions: int = Field(..., alias="resource-destructions")
-    resource_imports: int = Field(..., alias="resource-imports")
-    status: PlanStatus = Field(..., alias="status")
-    status_timestamps: PlanStatusTimestamps = Field(..., alias="status-timestamps")
+    has_changes: bool | None = Field(None, alias="has-changes")
+    generated_configuration: bool | None = Field(None, alias="generated-configuration")
+    log_read_url: str | None = Field(None, alias="log-read-url")
+    resource_additions: int | None = Field(None, alias="resource-additions")
+    resource_changes: int | None = Field(None, alias="resource-changes")
+    resource_destructions: int | None = Field(None, alias="resource-destructions")
+    resource_imports: int | None = Field(None, alias="resource-imports")
+    status: PlanStatus | None = Field(None, alias="status")
+    status_timestamps: PlanStatusTimestamps | None = Field(
+        None, alias="status-timestamps"
+    )
 
     # Relations
-    # exports: list[PlanExport] = Field(..., alias="exports")
+    exports: list[PlanExport] | None = Field(None, alias="exports")
 
 
 class PlanStatusTimestamps(BaseModel):
     model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
 
-    canceled_at: datetime = Field(..., alias="canceled-at")
-    errored_at: datetime = Field(..., alias="errored-at")
-    finished_at: datetime = Field(..., alias="finished-at")
-    force_canceled_at: datetime = Field(..., alias="force-canceled-at")
-    queued_at: datetime = Field(..., alias="queued-at")
-    started_at: datetime = Field(..., alias="started-at")
+    canceled_at: datetime | None = Field(None, alias="canceled-at")
+    errored_at: datetime | None = Field(None, alias="errored-at")
+    finished_at: datetime | None = Field(None, alias="finished-at")
+    force_canceled_at: datetime | None = Field(None, alias="force-canceled-at")
+    queued_at: datetime | None = Field(None, alias="queued-at")
+    started_at: datetime | None = Field(None, alias="started-at")

src/tfe/models/plan_export.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict
+
+
+class PlanExport(BaseModel):
+    model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
+
+    id: str

src/tfe/resources/apply.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from ..errors import InvalidApplyIDError
+from ..models.apply import (
+    Apply,
+)
+from ..utils import valid_string_id, validate_log_url
+from ._base import _Service
+
+
+class Applies(_Service):
+    def read(self, apply_id: str) -> Apply:
+        """Read a specific apply by its ID."""
+        if not valid_string_id(apply_id):
+            raise InvalidApplyIDError()
+
+        r = self.t.request(
+            "GET",
+            f"/api/v2/applies/{apply_id}",
+        )
+        d = r.json()["data"]
+        attr = d.get("attributes", {}) or {}
+        return Apply(
+            id=d.get("id"),
+            **{k.replace("-", "_"): v for k, v in attr.items()},
+        )
+
+    def logs(self, apply_id: str) -> str:
+        """Get logs for a specific apply"""
+        # Validate apply ID
+        if not valid_string_id(apply_id):
+            raise InvalidApplyIDError()
+
+        # Get the apply and validate log URL
+        apply = self.read(apply_id)
+        if not apply.log_read_url:
+            raise ValueError(f"Apply {apply_id} does not have a log URL")
+
+        validate_log_url(apply.log_read_url)
+
+        # Placeholder implementation - in future this would stream logs
+        return ""
+
+    def _done(self, apply_id: str) -> tuple[bool, Exception | None]:
+        """Check if an apply is in a terminal state."""
+        try:
+            apply_obj = self.read(apply_id)
+            terminal_states = {"canceled", "errored", "finished", "unreachable"}
+            is_complete = apply_obj.status in terminal_states
+            return is_complete, None
+        except Exception as e:
+            return False, e