feat: add execute_command example and utility functions for environment management

Author: akosyakov

examples/execute_command.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python
+
+from gitpod import Gitpod
+from utils import get_context_url, get_authenticated_user, get_environment_class_id, get_environment_class, create_environment
+import time
+import requests
+
+def check_environment_status(client, environment_id):
+    environment = client.environments.retrieve(environment_id=environment_id).environment
+    print(f"Environment status: {environment.status.phase}")
+    if environment.status.phase in ["ENVIRONMENT_PHASE_STOPPING", "ENVIRONMENT_PHASE_STOPPED", "ENVIRONMENT_PHASE_DELETING", "ENVIRONMENT_PHASE_DELETED"]:
+        raise RuntimeError(f"Environment {environment_id} is in an unexpected phase: {environment.status.phase}")
+    elif environment.status.failure_message and len(environment.status.failure_message) > 0:
+        raise RuntimeError(f"Environment {environment_id} failed: {'; '.join(environment.status.failure_message)}")
+
+def main() -> None:
+    client: Gitpod = Gitpod()
+    context_url: str = get_context_url()
+    user = get_authenticated_user(client)
+    environment_class_id: str = get_environment_class_id(client, user)
+
+    if not environment_class_id:
+        raise ValueError("no environment class found")
+
+    environment_class = get_environment_class(client, environment_class_id)
+
+    if not environment_class:
+        raise ValueError("no environment class found")
+
+    print(f"Using repository: {context_url}")
+    print(f"Using environment class: {environment_class.display_name} (ID: {environment_class.id})")
+
+    environment_id: str = create_environment(client, context_url, environment_class)
+    print(f"Created environment: {environment_id}")
+
+    try:
+        reference = "hello-world"
+        task = client.environments.automations.tasks.create(
+            spec={
+                "command": "echo 'Hello, World!'"
+            },
+            environment_id=environment_id,
+            metadata={
+                "name": "Hello, World!",
+                "description": "Prints 'Hello, World!' to the console.",
+                "reference": reference,
+            }
+        ).task
+        print(f"Created task: {task.id}")
+
+        execution = client.environments.automations.tasks.start(
+            id=task.id
+        ).task_execution
+        print(f"Started task execution: {execution.id}")
+
+        logs_access_token = client.environments.create_logs_token(environment_id=environment_id).access_token
+        print(f"Waiting for task logs to be available...")
+        
+        event_stream = client.events.watch(
+            environment_id=environment_id
+        )
+
+        log_url = None
+        for event in event_stream:
+            print(f"Received event: {event}")
+            if event.resource_type == "RESOURCE_TYPE_ENVIRONMENT" and event.resource_id == environment_id:
+                check_environment_status(client, environment_id)
+            elif event.resource_type == "RESOURCE_TYPE_TASK_EXECUTION" and event.resource_id == execution.id:
+                execution = client.environments.automations.tasks.executions.retrieve(id=execution.id).task_execution
+                print(f"Task execution status: {execution.status.phase}")
+                if execution.status.log_url:
+                    log_url = execution.status.log_url
+                    break
+    
+        if not log_url:
+            raise RuntimeError("Task logs are not available.")
+        
+        print(f"Task logs are available at: {log_url}")
+
+        # Stream the log URL and print the logs
+        response = requests.get(log_url, stream=True, headers={"Authorization": f"Bearer {logs_access_token}"})
+        for line in response.iter_lines():
+            if line:
+                print(line.decode('utf-8'))
+
+    finally:
+        client.environments.delete(environment_id=environment_id)
+        print(f"Deleted environment: {environment_id}")
+
+if __name__ == "__main__":
+    main()

examples/utils.py

@@ -0,0 +1,77 @@
+import os
+from typing import Optional
+from gitpod.types.environment_create_params import SpecContentInitializerSpecContextURL
+from gitpod import Gitpod
+from gitpod.types.environment_retrieve_response import Environment
+from gitpod.types.user_get_authenticated_user_response import User
+from gitpod.types.environments.class_list_response import ClassListResponse
+
+def get_context_url() -> str:
+    return os.environ.get("CONTEXT_URL", "https://github.com/gitpod-io/empty")
+
+def get_authenticated_user(client: Gitpod) -> User:
+    return client.users.get_authenticated_user(body={}).user
+
+def get_environment_class_id(client: Gitpod, user: User) -> Optional[str]:
+    environment_class_id = os.environ.get("ENVIRONMENT_CLASS_ID")
+    if environment_class_id:
+        return environment_class_id
+
+    class_usage = {}
+    page = client.environments.list(organization_id=user.organization_id)
+    while page:
+        for env in page.environments:
+            env_class = env.spec.machine.class_
+            if env_class not in class_usage:
+                class_usage[env_class] = 0
+            class_usage[env_class] += 1
+        if page.pagination and page.pagination.next_token:
+            page = client.environments.list(token=page.pagination.next_token, organization_id=user.organization_id)
+        else:
+            break
+
+    sorted_classes = sorted(class_usage.items(), key=lambda item: -item[1])
+    return sorted_classes[0][0] if sorted_classes else None
+
+def get_environment_class(client: Gitpod, environment_class_id: str) -> Optional[ClassListResponse]:
+    page = client.environments.classes.list(filter={"enabled": True})
+    while page:
+        for cls in page.environment_classes:
+            if cls.id == environment_class_id:
+                return cls
+        if page.pagination and page.pagination.next_token:
+            page = client.environments.classes.list(token=page.pagination.next_token)
+        else:
+            break
+    return None
+
+def create_environment(client: Gitpod, context_url: str, environment_class: ClassListResponse) -> str:
+    resp = client.environments.create(
+        spec={
+            "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
+            "content": {
+                "initializer": {
+                    "specs": [
+                        SpecContentInitializerSpecContextURL(
+                            context_url={
+                                "url": context_url
+                            }
+                        )
+                    ]
+                }
+            },
+            "machine": {
+                "class": environment_class.id
+            }
+        }
+    )
+    return resp.environment.id
+
+def check_environment_status(client: Gitpod, environment_id: str) -> None:
+    environment = client.environments.retrieve(environment_id=environment_id).environment
+    print(f"Environment status: {environment.status.phase}")
+    if environment.status.phase in ["ENVIRONMENT_PHASE_STOPPING", "ENVIRONMENT_PHASE_STOPPED", "ENVIRONMENT_PHASE_DELETING", "ENVIRONMENT_PHASE_DELETED"]:
+        raise RuntimeError(f"Environment {environment_id} is in an unexpected phase: {environment.status.phase}")
+    elif environment.status.failure_message and len(environment.status.failure_message) > 0:
+        raise RuntimeError(f"Environment {environment_id} failed: {'; '.join(environment.status.failure_message)}")
+