Run Event API Specs

examples/run_events.py

@@ -0,0 +1,149 @@
+"""Run Events Example for python-tfe SDK
+
+This example demonstrates how to work with run events using the python-tfe SDK.
+Run events represent activities that happen during a run's lifecycle, such as:
+- Run state changes (queued, planning, planned, etc.)
+- User actions (approve, discard, cancel)
+- System events (plan finished, apply started, etc.)
+
+Features demonstrated:
+1. Listing all run events for a specific run
+2. Reading individual run event details
+3. Including related data (actor, comment) in responses
+4. Error handling and proper client configuration
+
+Usage examples:
+  # List all events for a run
+  python examples/run_events.py --run-id run-abc123
+
+  # List events with actor and comment information included
+  python examples/run_events.py --run-id run-abc123 --include-actor --include-comment
+
+  # Read a specific run event
+  python examples/run_events.py --run-id run-abc123 --event-id re-xyz789
+
+Environment variables:
+  TFE_ADDRESS: Terraform Enterprise/Cloud address (default: https://app.terraform.io)
+  TFE_TOKEN: API token for authentication (required)
+
+Requirements:
+  - Valid TFE API token with appropriate permissions
+  - Valid run ID (from an existing run in your organization)
+  - Optional: Valid run event ID for detailed reading
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+
+from tfe import TFEClient, TFEConfig
+from tfe.models.run_event import (
+    RunEventIncludeOpt,
+    RunEventListOptions,
+    RunEventReadOptions,
+)
+
+
+def _print_header(title: str):
+    print("\n" + "=" * 80)
+    print(title)
+    print("=" * 80)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Run Events 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("--run-id", required=True, help="Run ID to list events for")
+    parser.add_argument("--event-id", help="Specific Run Event ID to read")
+    parser.add_argument(
+        "--include-actor",
+        action="store_true",
+        help="Include actor information in the response",
+    )
+    parser.add_argument(
+        "--include-comment",
+        action="store_true",
+        help="Include comment information in the response",
+    )
+    parser.add_argument("--page", type=int, default=1)
+    parser.add_argument("--page-size", type=int, default=20)
+    args = parser.parse_args()
+
+    if not args.token:
+        print("Error: TFE_TOKEN environment variable or --token argument is required")
+        return 1
+
+    # Configure the client
+    cfg = TFEConfig(address=args.address, token=args.token)
+    client = TFEClient(cfg)
+
+    # Build include options if requested
+    include_opts = []
+    if args.include_actor:
+        include_opts.append(RunEventIncludeOpt.RUN_EVENT_ACTOR)
+    if args.include_comment:
+        include_opts.append(RunEventIncludeOpt.RUN_EVENT_COMMENT)
+
+    # 1) List run events for the specified run
+    _print_header(f"Listing Run Events for Run: {args.run_id}")
+
+    options = RunEventListOptions(include=include_opts if include_opts else None)
+
+    try:
+        event_list = client.run_events.list(args.run_id, options)
+
+        print(f"Total run events: {event_list.total_count or 'N/A'}")
+        if event_list.current_page and event_list.total_pages:
+            print(f"Page {event_list.current_page} of {event_list.total_pages}")
+        print()
+
+        if not event_list.items:
+            print("No run events found for this run.")
+        else:
+            for event in event_list.items:
+                print(f"Event ID: {event.id}")
+                print(f"  Action: {event.action or 'N/A'}")
+                print(f"  Description: {event.description or 'N/A'}")
+                print(f"  Created At: {event.created_at or 'N/A'}")
+
+                print()
+
+    except Exception as e:
+        print(f"Error listing run events: {e}")
+        return 1
+
+    # 2) Read a specific run event if provided
+    if args.event_id:
+        _print_header(f"Reading Run Event: {args.event_id}")
+
+        read_options = RunEventReadOptions(
+            include=include_opts if include_opts else None
+        )
+
+        try:
+            event = client.run_events.read_with_options(args.event_id, read_options)
+
+            print(f"Event ID: {event.id}")
+            print(f"Action: {event.action or 'N/A'}")
+            print(f"Description: {event.description or 'N/A'}")
+            print(f"Created At: {event.created_at or 'N/A'}")
+
+        except Exception as e:
+            print(f"Error reading run event: {e}")
+            return 1
+
+    # 3) Summary
+    _print_header("Summary")
+    print(f"Successfully demonstrated run events for run: {args.run_id}")
+    print(f"Total events found: {event_list.total_count or 'N/A'}")
+    if args.event_id:
+        print(f"Successfully read specific event: {args.event_id}")
+    return 0
+
+
+if __name__ == "__main__":
+    exit(main())

src/tfe/client.py

@@ -13,6 +13,7 @@
 from .resources.registry_module import RegistryModules
 from .resources.registry_provider import RegistryProviders
 from .resources.run import Runs
+from .resources.run_event import RunEvents
 from .resources.run_task import RunTasks
 from .resources.run_trigger import RunTriggers
 from .resources.state_version_outputs import StateVersionOutputs
@@ -64,6 +65,7 @@ def __init__(self, config: TFEConfig | None = None):
         self.run_triggers = RunTriggers(self._transport)
         self.runs = Runs(self._transport)
         self.query_runs = QueryRuns(self._transport)
+        self.run_events = RunEvents(self._transport)
 
     def close(self) -> None:
         pass

src/tfe/errors.py

@@ -364,3 +364,11 @@ class InvalidApplyIDError(InvalidValues):
 
     def __init__(self, message: str = "invalid value for apply ID"):
         super().__init__(message)
+
+
+# Run Event errors
+class InvalidRunEventIDError(InvalidValues):
+    """Raised when an invalid run event ID is provided."""
+
+    def __init__(self, message: str = "invalid value for run event ID"):
+        super().__init__(message)

src/tfe/models/run_event.py

@@ -1,22 +1,50 @@
 from __future__ import annotations
 
 from datetime import datetime
+from enum import Enum
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from .comment import Comment
 from .user import User
 
-# from .comment import Comment
+
+class RunEventIncludeOpt(str, Enum):
+    RUN_EVENT_ACTOR = "actor"
+    RUN_EVENT_COMMENT = "comment"
 
 
 class RunEvent(BaseModel):
     model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
 
     id: str
-    # action: RunEventAction = Field(..., alias="action")
-    created_at: datetime = Field(..., alias="created-at")
-    description: str = Field(..., alias="description")
+    action: str | None = Field(None, alias="action")
+    created_at: datetime | None = Field(None, alias="created-at")
+    description: str | None = Field(None, alias="description")
 
     # Relations - Note that `target` is not supported yet
-    actor: User = Field(..., alias="actor")
-    # comment: Comment | None = Field(None, alias="comment")
+    actor: User | None = Field(None, alias="actor")
+    comment: Comment | None = Field(None, alias="comment")
+
+
+class RunEventList(BaseModel):
+    model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
+
+    items: list[RunEvent] = Field(default_factory=list)
+    current_page: int | None = None
+    total_pages: int | None = None
+    prev_page: int | None = None
+    next_page: int | None = None
+    total_count: int | None = None
+
+
+class RunEventListOptions(BaseModel):
+    model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
+
+    include: list[RunEventIncludeOpt] | None = Field(None, alias="include")
+
+
+class RunEventReadOptions(BaseModel):
+    model_config = ConfigDict(populate_by_name=True, validate_by_name=True)
+
+    include: list[RunEventIncludeOpt] | None = Field(None, alias="include")

src/tfe/resources/configuration_version.py

@@ -174,7 +174,7 @@ def upload_tar_gzip(self, upload_url: str, archive: io.IOBase) -> None:
                         f"Upload failed with status {response.status_code}: {response.text}"
                     )
         except Exception as e:
-            if isinstance(e, (NotFound, AuthError, ServerError, TFEError)):
+            if isinstance(e, NotFound | AuthError | ServerError | TFEError):
                 raise
             raise TFEError(f"Upload failed: {str(e)}") from e
 

src/tfe/resources/run_event.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from ..errors import InvalidRunEventIDError, InvalidRunIDError
+from ..models.run_event import (
+    RunEvent,
+    RunEventList,
+    RunEventListOptions,
+    RunEventReadOptions,
+)
+from ..utils import valid_string_id
+from ._base import _Service
+
+
+class RunEvents(_Service):
+    def list(
+        self, run_id: str, options: RunEventListOptions | None = None
+    ) -> RunEventList:
+        """List all the run events of the given run."""
+        if not valid_string_id(run_id):
+            raise InvalidRunIDError()
+        params = (
+            options.model_dump(by_alias=True, exclude_none=True) if options else None
+        )
+        r = self.t.request(
+            "GET",
+            f"/api/v2/runs/{run_id}/run-events",
+            params=params,
+        )
+        jd = r.json()
+        items = []
+        meta = jd.get("meta", {})
+        pagination = meta.get("pagination", {})
+        for d in jd.get("data", []):
+            attrs = d.get("attributes", {})
+            attrs["id"] = d.get("id")
+            items.append(RunEvent.model_validate(attrs))
+        return RunEventList(
+            items=items,
+            current_page=pagination.get("current-page"),
+            total_pages=pagination.get("total-pages"),
+            prev_page=pagination.get("prev-page"),
+            next_page=pagination.get("next-page"),
+            total_count=pagination.get("total-count"),
+        )
+
+    def read(self, run_event_id: str) -> RunEvent:
+        """Read a specific run event by its ID."""
+        return self.read_with_options(run_event_id, None)
+
+    def read_with_options(
+        self, run_event_id: str, options: RunEventReadOptions | None = None
+    ) -> RunEvent:
+        """Read a specific run event by its ID with the given options."""
+        if not valid_string_id(run_event_id):
+            raise InvalidRunEventIDError()
+        params = (
+            options.model_dump(by_alias=True, exclude_none=True) if options else None
+        )
+        r = self.t.request(
+            "GET",
+            f"/api/v2/run-events/{run_event_id}",
+            params=params,
+        )
+        d = r.json().get("data", {})
+        attr = d.get("attributes", {}) or {}
+        return RunEvent(
+            id=d.get("id"),
+            **{k.replace("-", "_"): v for k, v in attr.items()},
+        )