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,63 @@
+#!/usr/bin/env python
+
+from utils.task import run_command, create_command, delete_command, update_command
+from utils.context import get_context_url, get_authenticated_user
+from utils.environment import create_environment, get_environment_class, get_environment_class_id
+
+from gitpod import Gitpod
+
+
+def main() -> None:
+    """
+    Main function to create an environment, start a task, and stream its logs.
+    """
+    # Initialize Gitpod client
+    client: Gitpod = Gitpod()
+
+    # Get context URL and authenticated user
+    context_url: str = get_context_url()
+    user = get_authenticated_user(client)
+
+    # Get environment class ID and details
+    environment_class_id: str = get_environment_class_id(client, user)
+    if not environment_class_id:
+        raise ValueError("no environment class found, please set the ENVIRONMENT_CLASS_ID environment variable")
+    environment_class = get_environment_class(client, environment_class_id)
+    if not environment_class:
+        raise ValueError("no environment class found, please set the ENVIRONMENT_CLASS_ID environment variable")
+
+    print(f"Using repository: {context_url}")
+    print(f"Using environment class: {environment_class.display_name} ({environment_class.description})")
+
+    # Create environment
+    environment_id: str = create_environment(client, context_url, environment_class)
+    print(f"\nCreated environment: {environment_id}")
+
+    try:
+        # Define command
+        reference = "greeting"
+        name = "Greeting"
+        description = "Prints a greeting to the console."
+        command = "echo 'Hello, World!'"
+        task_id = create_command(client, environment_id, reference, name, description, command)
+        try:
+            # Run command with first greeting
+            run_command(client, environment_id, task_id)
+
+            # Update command with second greeting
+            command = "echo 'Howdy, Partner!'"
+            update_command(client, task_id, command)
+
+            # Run command with second greeting
+            run_command(client, environment_id, task_id)
+        finally:
+            # Clean up command
+            delete_command(client, task_id)
+    finally:
+        # Clean up environment
+        client.environments.delete(environment_id=environment_id)
+        print(f"\nDeleted environment: {environment_id}")
+
+
+if __name__ == "__main__":
+    main()

examples/utils/context.py

@@ -0,0 +1,27 @@
+import os
+
+from gitpod import Gitpod
+from gitpod.types.user_get_authenticated_user_response import User
+
+
+def get_context_url() -> str:
+    """
+    Retrieve the context URL from the environment variable or use a default value.
+
+    Returns:
+        str: The context URL.
+    """
+    return os.environ.get("CONTEXT_URL", "https://github.com/gitpod-io/empty")
+
+
+def get_authenticated_user(client: Gitpod) -> User:
+    """
+    Get the authenticated user using the Gitpod client.
+
+    Args:
+        client (Gitpod): The Gitpod client instance.
+
+    Returns:
+        User: The authenticated user.
+    """
+    return client.users.get_authenticated_user(body={}).user

examples/utils/environment.py

@@ -0,0 +1,110 @@
+import os
+from typing import Optional
+
+from gitpod import Gitpod
+from gitpod.types.environment_create_params import SpecContentInitializerSpecContextURL
+from gitpod.types.environments.class_list_response import ClassListResponse
+from gitpod.types.user_get_authenticated_user_response import User
+
+
+def get_environment_class_id(client: Gitpod, user: User) -> Optional[str]:
+    """
+    Get the most used environment class ID for the given user.
+
+    Args:
+        client (Gitpod): The Gitpod client instance.
+        user (User): The authenticated user.
+
+    Returns:
+        Optional[str]: The environment class ID or None if not found.
+    """
+    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]:
+    """
+    Get the environment class details for the given environment class ID.
+
+    Args:
+        client (Gitpod): The Gitpod client instance.
+        environment_class_id (str): The environment class ID.
+
+    Returns:
+        Optional[ClassListResponse]: The environment class details or None if not found.
+    """
+    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:
+    """
+    Create a new environment with the given context URL and environment class.
+
+    Args:
+        client (Gitpod): The Gitpod client instance.
+        context_url (str): The context URL.
+        environment_class (ClassListResponse): The environment class details.
+
+    Returns:
+        str: The created environment ID.
+    """
+    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 ensure_environment_healthy(client: Gitpod, environment_id: str) -> None:
+    """
+    Ensure the environment is running or will be running and raise an error if it is in an unexpected phase.
+
+    Args:
+        client (Gitpod): The Gitpod client instance.
+        environment_id (str): The environment ID.
+
+    Raises:
+        RuntimeError: If the environment is in an unexpected phase or has a failure message.
+    """
+    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)}")

examples/utils/logs.py

@@ -0,0 +1,16 @@
+import httpx
+
+
+async def stream_logs(log_url: str, logs_access_token: str) -> None:
+    """
+    Stream logs from the given log URL using the provided access token.
+
+    Args:
+        log_url (str): The log URL.
+        logs_access_token (str): The access token for the logs.
+    """
+    async with httpx.AsyncClient() as client:
+        async with client.stream("GET", log_url, headers={"Authorization": f"Bearer {logs_access_token}"}) as response:
+            async for line in response.aiter_lines():
+                if line:
+                    print(line)