add execute command example

Author: akosyakov

.devcontainer/Dockerfile

@@ -1,9 +1,8 @@
-ARG VARIANT="3.9"
-FROM mcr.microsoft.com/vscode/devcontainers/python:0-${VARIANT}
+FROM mcr.microsoft.com/devcontainers/python:3.12
 
 USER vscode
 
 RUN curl -sSf https://rye.astral.sh/get | RYE_VERSION="0.35.0" RYE_INSTALL_OPTION="--yes" bash
 ENV PATH=/home/vscode/.rye/shims:$PATH
 
-RUN echo "[[ -d .venv ]] && source .venv/bin/activate" >> /home/vscode/.bashrc
+RUN echo "[[ -d .venv ]] && source .venv/bin/activate || export PATH=\$PATH" >> /home/vscode/.bashrc

examples/execute_command.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python
+
+import sys
+import asyncio
+from gitpod import Gitpod
+import gitpod.lib as util
+from gitpod.types.environment_create_params import SpecContentInitializerSpecContextURL
+
+# Example: ./examples/execute_command.py https://github.com/gitpod-io/empty 'echo "Hello World!"'
+async def main() -> None:
+    client = Gitpod()
+
+    if len(sys.argv) < 3:
+        print("Usage: ./examples/execute_command.py <CONTEXT_URL> <COMMAND>")
+        sys.exit(1)
+    context_url = sys.argv[1]
+    command = " ".join(sys.argv[2:])
+
+    env_class = util.find_most_used_environment_class(client)
+    if not env_class:
+        print("Error: No environment class found. Please create one first.")
+        sys.exit(1)
+
+    environment_id = client.environments.create(
+        spec={
+            "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
+            "content": {
+                "initializer": {"specs": [SpecContentInitializerSpecContextURL(context_url={"url": context_url})]}
+            },
+            "machine": {"class": env_class.id},
+        }
+    ).environment.id
+
+    try:
+        util.wait_for_environment_ready(client, environment_id)
+        
+        output = await util.run_command(client, environment_id, command)
+        print(output)
+    finally:
+        client.environments.delete(environment_id=environment_id)
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/gitpod/lib/__init__.py

@@ -0,0 +1,8 @@
+from .task import run_command
+from .environment import find_most_used_environment_class, wait_for_environment_ready
+
+__all__ = [
+    'find_most_used_environment_class',
+    'wait_for_environment_ready',
+    'run_command'
+] 

src/gitpod/lib/environment.py

@@ -0,0 +1,67 @@
+from typing import Optional
+
+from gitpod import Gitpod
+from gitpod.types.environments.class_list_response import ClassListResponse
+
+def find_most_used_environment_class(client: Gitpod) -> Optional[ClassListResponse]:
+    """
+    Find the most used environment class.
+    """
+    user = client.users.get_authenticated_user(body={}).user
+
+    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])
+    environment_class_id = sorted_classes[0][0] if sorted_classes else None
+    if not environment_class_id:
+        return None
+    
+    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 wait_for_environment_ready(client: Gitpod, environment_id: str) -> None:
+    def is_ready() -> bool:
+        environment = client.environments.retrieve(environment_id=environment_id).environment
+        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)}")
+        return environment.status.phase == "ENVIRONMENT_PHASE_RUNNING"
+    
+    event_stream = client.events.watch(environment_id=environment_id, timeout=None)
+    try:
+        if is_ready():
+            return
+
+        for event in event_stream:
+            if event.resource_type == "RESOURCE_TYPE_ENVIRONMENT" and event.resource_id == environment_id:
+                if is_ready():
+                    return
+    finally:
+        event_stream.http_response.close()
+    

src/gitpod/lib/task.py

@@ -0,0 +1,67 @@
+import httpx
+
+from gitpod import Gitpod
+TASK_REFERENCE = "gitpod-python-sdk"
+
+async def run_command(client: Gitpod, environment_id: str, command: str) -> str:
+    tasks = client.environments.automations.tasks.list(
+        filter={
+            "references": [TASK_REFERENCE],
+            "environment_ids": [environment_id]
+        }
+    ).tasks
+
+    task_id = ""
+    if not tasks:
+        task_id = client.environments.automations.tasks.create(
+            spec={
+                "command": command,
+            },
+            environment_id=environment_id,
+            metadata={
+                "name": "Gitpod Python SDK Task",
+                "description": "Gitpod Python SDK Task",
+                "reference": TASK_REFERENCE,
+            },
+        ).task.id
+    else:
+        task_id = tasks[0].id
+        client.environments.automations.tasks.update(
+            id=task_id,
+            spec={
+                "command": command,
+            },
+        )
+
+    task_execution_id = client.environments.automations.tasks.start(id=task_id).task_execution.id
+    log_url = wait_for_log_url(client, environment_id, task_execution_id)
+    logs_access_token = client.environments.create_logs_token(environment_id=environment_id).access_token
+    async with httpx.AsyncClient() as http_client:
+        async with http_client.stream("GET", log_url, headers={"Authorization": f"Bearer {logs_access_token}"}) as response:
+            chunks: list[str] = []
+            async for chunk in response.aiter_text():
+                chunks.append(chunk)
+            return "".join(chunks)
+
+def wait_for_log_url(client: Gitpod, environment_id: str, task_execution_id: str) -> str:
+    def get_log_url() -> str:
+        execution = client.environments.automations.tasks.executions.retrieve(id=task_execution_id).task_execution
+        return execution.status.log_url
+    
+    log_url = get_log_url()
+    if log_url:
+        return log_url 
+    
+    event_stream = client.events.watch(environment_id=environment_id, timeout=None)
+    try:
+        log_url = get_log_url()
+        if log_url:
+            return log_url 
+    
+        for event in event_stream:
+            if event.resource_type == "RESOURCE_TYPE_TASK_EXECUTION" and event.resource_id == task_execution_id:
+                log_url = get_log_url()
+                if log_url:
+                    break
+    finally:
+        event_stream.http_response.close()