refactor: update environment and automation library methods

Author: akosyakov

examples/run_command.py

@@ -4,7 +4,6 @@
 import asyncio
 from gitpod import Gitpod
 import gitpod.lib as util
-from gitpod.types.environment_create_params import SpecContentInitializerSpecContextURL
 
 # Example: ./examples/run_command.py https://github.com/gitpod-io/empty 'echo "Hello World!"'
 async def main() -> None:
@@ -25,7 +24,11 @@ async def main() -> None:
         spec={
             "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
             "content": {
-                "initializer": {"specs": [SpecContentInitializerSpecContextURL(context_url={"url": context_url})]}
+                "initializer": {"specs": [{
+                    "contextUrl": {
+                        "url": context_url
+                    }
+                }]}
             },
             "machine": {"class": env_class.id},
         }

examples/run_service.py

@@ -4,7 +4,6 @@
 import asyncio
 from gitpod import Gitpod
 import gitpod.lib as util
-from gitpod.types.environment_create_params import SpecContentInitializerSpecContextURL
 
 # Example: ./examples/run_service.py https://github.com/gitpod-io/empty
 async def main() -> None:
@@ -26,7 +25,11 @@ async def main() -> None:
         spec={
             "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
             "content": {
-                "initializer": {"specs": [SpecContentInitializerSpecContextURL(context_url={"url": context_url})]}
+                "initializer": {"specs": [{
+                    "contextUrl": {
+                        "url": context_url
+                    }
+                }]}
             },
             "machine": {"class": env_class.id},
             "ports": [

src/gitpod/lib/automation.py

@@ -1,3 +1,4 @@
+import asyncio
 import httpx
 from typing import AsyncIterator
 
@@ -8,8 +9,8 @@
 def run_service(
     client: Gitpod,
     environment_id: str,
-    metadata: service_create_params.Metadata,
-    spec: service_create_params.Spec
+    metadata: service_create_params.ServiceMetadataParam,
+    spec: service_create_params.ServiceSpecParam
 ) -> AsyncIterator[str]:
     reference = metadata["reference"]
     if not reference:
@@ -36,7 +37,7 @@ def run_service(
     log_url = wait_for_service_log_url(client, environment_id, service_id)
     return stream_logs(client, environment_id, log_url)
 
-async def run_command(client: Gitpod, environment_id: str, command: str) -> AsyncIterator[str]:
+def run_command(client: Gitpod, environment_id: str, command: str) -> AsyncIterator[str]:
     tasks = client.environments.automations.tasks.list(
         filter={
             "references": [TASK_REFERENCE],
@@ -70,20 +71,6 @@ async def run_command(client: Gitpod, environment_id: str, command: str) -> Asyn
     log_url = wait_for_task_log_url(client, environment_id, task_execution_id)
     return stream_logs(client, environment_id, log_url)
 
-async def stream_logs(client: Gitpod, environment_id: str, log_url: str) -> AsyncIterator[str]:
-    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}"}, timeout=None) as response:
-            buffer = ""
-            async for chunk in response.aiter_text():
-                buffer += chunk
-                while "\n" in buffer:
-                    line, buffer = buffer.split("\n", 1)
-                    if line:
-                        yield line
-            if buffer:
-                yield buffer
-
 def wait_for_task_log_url(client: Gitpod, environment_id: str, task_execution_id: str) -> str:
     def get_log_url():
         execution = client.environments.automations.tasks.executions.retrieve(id=task_execution_id).task_execution
@@ -118,3 +105,35 @@ def wait_for_log_url(client: Gitpod, environment_id: str, resource_id: str, get_
                     return log_url
     finally:
         event_stream.http_response.close()
+
+async def stream_logs(client: Gitpod, environment_id: str, log_url: str) -> AsyncIterator[str]:
+    logs_access_token = client.environments.create_logs_token(environment_id=environment_id).access_token
+    async with httpx.AsyncClient() as http_client:
+        retries = 3
+        while retries > 0:
+            try:
+                async with http_client.stream("GET", log_url, headers={"Authorization": f"Bearer {logs_access_token}"}, timeout=None) as response:
+                    if response.status_code == 502:  # Bad Gateway
+                        retries -= 1
+                        if retries == 0:
+                            raise Exception("Failed to stream logs after 3 retries")
+                        await asyncio.sleep(1)  # Wait before retrying
+                        continue
+                        
+                    buffer = ""
+                    async for chunk in response.aiter_text():
+                        buffer += chunk
+                        while "\n" in buffer:
+                            line, buffer = buffer.split("\n", 1)
+                            if line:
+                                yield line
+                    if buffer:
+                        yield buffer
+                    break  # Success - exit retry loop
+                    
+            except httpx.HTTPError as e:
+                if retries > 0 and (isinstance(e, httpx.HTTPStatusError) and e.response.status_code == 502):
+                    retries -= 1
+                    await asyncio.sleep(1)  # Wait before retrying
+                    continue
+                raise  # Re-raise if not a 502 or out of retries

src/gitpod/lib/environment.py

@@ -1,24 +1,22 @@
 from typing import Optional
 
 from gitpod import Gitpod
-from gitpod.types.environments.class_list_response import ClassListResponse
+from gitpod.types.shared import EnvironmentClass
 
-def find_most_used_environment_class(client: Gitpod) -> Optional[ClassListResponse]:
+def find_most_used_environment_class(client: Gitpod) -> Optional[EnvironmentClass]:
     """
     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)
+    page = client.environments.list()
     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)
+            page = client.environments.list(token=page.pagination.next_token)
         else:
             break
 
@@ -27,7 +25,7 @@ def find_most_used_environment_class(client: Gitpod) -> Optional[ClassListRespon
     if not environment_class_id:
         return None
 
-    page = client.environments.classes.list(filter={"enabled": True})
+    page = client.environments.classes.list(filter={"can_create_environments": True})
     while page:
         for cls in page.environment_classes:
             if cls.id == environment_class_id: