fix linting

Author: akosyakov

examples/anthropic_tool_use.py

@@ -2,11 +2,14 @@
 
 from __future__ import annotations
 
+from typing import cast
+
 from anthropic import Anthropic
 from anthropic.types import ToolParam, MessageParam
 
 import gitpod.lib as util
 from gitpod import AsyncGitpod
+from gitpod.types.environment_initializer_param import Spec
 
 gpclient = AsyncGitpod()
 llmclient = Anthropic()
@@ -38,29 +41,40 @@ async def create_environment(args: dict[str, str], cleanup: util.Disposables) ->
     env_class = await util.find_most_used_environment_class(gpclient)
     if not env_class:
         raise Exception("No environment class found. Please create one first.")
-    environment_id = (await gpclient.environments.create(
+    env_class_id = env_class.id
+    assert env_class_id is not None
+
+    environment = (await gpclient.environments.create(
         spec={
             "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
             "content": {
-                "initializer": {"specs": [{"contextUrl": {"url": args["context_url"]}}]},
+                "initializer": {"specs": [Spec(
+                    context_url={
+                        "url": args["context_url"]
+                    }
+                )]},
             },
-            "machine": {"class": env_class.id},
+            "machine": {"class": env_class_id},
         }
-    )).environment.id
-    cleanup.add(lambda: asyncio.run(gpclient.environments.delete(environment_id=environment_id)))
+    )).environment
+    assert environment is not None
+    environment_id = environment.id
+    assert environment_id is not None
+    cleanup.add(lambda: asyncio.run(gpclient.environments.delete(environment_id=environment_id)))    
+    
     print(f"\nCreated environment: {environment_id} - waiting for it to be ready...")
     await util.wait_for_environment_ready(gpclient, environment_id)
     print(f"\nEnvironment is ready: {environment_id}")
     return environment_id
 
 async def execute_command(args: dict[str, str]) -> str:
     lines_iter = await util.run_command(gpclient, args["environment_id"], args["command"])
-    lines = []
+    lines: list[str] = []
     async for line in lines_iter:
         lines.append(line)
     return "\n".join(lines)
 
-async def main(cleanup: util.Disposables):
+async def main(cleanup: util.Disposables) -> None:
     messages = [user_message]
     while True:
         message = llmclient.messages.create(
@@ -83,7 +97,8 @@ async def main(cleanup: util.Disposables):
         for tool in (c for c in message.content if c.type == "tool_use"):
             try:
                 if tool.name == "create_environment":
-                    environment_id = await create_environment(tool.input, cleanup)
+                    args = cast(dict[str, str], tool.input)
+                    environment_id = await create_environment(args, cleanup)
                     messages.append({
                         "role": "user",
                         "content": [{
@@ -93,7 +108,8 @@ async def main(cleanup: util.Disposables):
                         }],
                     })
                 elif tool.name == "execute_command":
-                    output = await execute_command(tool.input)
+                    args = cast(dict[str, str], tool.input)
+                    output = await execute_command(args)
                     messages.append({
                         "role": "user",
                         "content": [{

examples/fs_access.py

@@ -9,6 +9,7 @@
 import gitpod.lib as util
 from gitpod import AsyncGitpod
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
+from gitpod.types.environment_initializer_param import Spec
 
 
 # Examples:
@@ -24,6 +25,8 @@ async def main(cleanup: util.Disposables) -> None:
         print("Error: No environment class found. Please create one first.")
         sys.exit(1)
     print(f"Found environment class: {env_class.display_name} ({env_class.description})")
+    env_class_id = env_class.id
+    assert env_class_id is not None
 
     print("Generating SSH key pair")
     key = paramiko.RSAKey.generate(2048)
@@ -36,22 +39,26 @@ async def main(cleanup: util.Disposables) -> None:
     key_id = "fs-access-example"
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class.id},
+        "machine": {"class": env_class_id},
         "ssh_public_keys": [{
             "id": key_id,
             "value": public_key
         }]
     }
     if context_url:
         spec["content"] = {
-            "initializer": {"specs": [{
-                "contextUrl": {
+            "initializer": {"specs": [Spec(
+                context_url={
                     "url": context_url
                 }
-            }]}
+            )]}
         }
 
-    environment_id = (await client.environments.create(spec=spec)).environment.id
+    print("Creating environment")
+    environment = (await client.environments.create(spec=spec)).environment
+    assert environment is not None
+    environment_id = environment.id
+    assert environment_id is not None
     cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
 
     env = util.EnvironmentState(client, environment_id)
@@ -70,8 +77,8 @@ async def main(cleanup: util.Disposables) -> None:
     # Parse ssh://username@host:port format
     url_parts = ssh_url.split('://')[-1]
     username, rest = url_parts.split('@')
-    host, port = rest.split(':')
-    port = int(port)
+    host, port_str = rest.split(':')
+    port = int(port_str)
 
     ssh = paramiko.SSHClient()
     ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
@@ -94,7 +101,7 @@ async def main(cleanup: util.Disposables) -> None:
 
     with sftp.file('test.txt', 'r') as f:
         content = f.read()
-        print(f"File content: {content}")
+        print(f"File content: {content.decode()}")
 
 if __name__ == "__main__":
     disposables = util.Disposables()

examples/run_command.py

@@ -6,6 +6,7 @@
 import gitpod.lib as util
 from gitpod import AsyncGitpod
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
+from gitpod.types.environment_initializer_param import Spec
 
 
 # Examples:
@@ -26,22 +27,27 @@ async def main(cleanup: util.Disposables) -> None:
         print("Error: No environment class found. Please create one first.")
         sys.exit(1)
     print(f"Found environment class: {env_class.display_name} ({env_class.description})")
+    env_class_id = env_class.id
+    assert env_class_id is not None
 
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class.id},
+        "machine": {"class": env_class_id},
     }
     if context_url:
         spec["content"] = {
-            "initializer": {"specs": [{
-                "contextUrl": {
+            "initializer": {"specs": [Spec(
+                context_url={
                     "url": context_url
                 }
-            }]}
+            )]}
         }
 
     print("Creating environment")
-    environment_id = (await client.environments.create(spec=spec)).environment.id
+    environment = (await client.environments.create(spec=spec)).environment
+    assert environment is not None
+    environment_id = environment.id
+    assert environment_id is not None
     cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
 
     print("Waiting for environment to be ready")

examples/run_service.py

@@ -6,6 +6,7 @@
 import gitpod.lib as util
 from gitpod import AsyncGitpod
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
+from gitpod.types.environment_initializer_param import Spec
 
 
 # Examples:
@@ -21,11 +22,13 @@ async def main(cleanup: util.Disposables) -> None:
         print("Error: No environment class found. Please create one first.")
         sys.exit(1)
     print(f"Found environment class: {env_class.display_name} ({env_class.description})")
-
+    env_class_id = env_class.id
+    assert env_class_id is not None
+    
     port = 8888
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class.id},
+        "machine": {"class": env_class_id},
         "ports": [{
             "name": "Lama Service",
             "port": port,
@@ -34,15 +37,18 @@ async def main(cleanup: util.Disposables) -> None:
     }
     if context_url:
         spec["content"] = {
-            "initializer": {"specs": [{
-                "contextUrl": {
-                    "url": context_url
-                }
-            }]}
-        }
+            "initializer": {"specs": [Spec(
+             context_url={
+                 "url": context_url
+             }
+        )]}
+    }
 
     print("Creating environment")
-    environment_id = (await client.environments.create(spec=spec)).environment.id
+    environment = (await client.environments.create(spec=spec)).environment
+    assert environment is not None
+    environment_id = environment.id
+    assert environment_id is not None
     cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
 
     print("Waiting for environment to be ready")

pyproject.toml

@@ -58,6 +58,7 @@ dev-dependencies = [
     "nest_asyncio==1.6.0",
     "paramiko>=3.5.1",
     "anthropic>=0.45.2",
+    "types-paramiko>=3.5.0.20240928",
 ]
 
 [tool.rye.scripts]

requirements-dev.lock

@@ -30,6 +30,7 @@ colorlog==6.7.0
     # via nox
 cryptography==44.0.1
     # via paramiko
+    # via types-paramiko
 dirty-equals==0.6.0
 distlib==0.3.7
     # via virtualenv
@@ -111,6 +112,7 @@ time-machine==2.9.0
 tomli==2.0.2
     # via mypy
     # via pytest
+types-paramiko==3.5.0.20240928
 typing-extensions==4.12.2
     # via anthropic
     # via anyio

src/gitpod/lib/automation.py

@@ -14,7 +14,7 @@ async def run_service(
     metadata: service_create_params.ServiceMetadataParam,
     spec: service_create_params.ServiceSpecParam
 ) -> AsyncIterator[str]:
-    reference = metadata["reference"]
+    reference = metadata.get("reference")
     if not reference:
         raise ValueError("metadata.reference is required")
 
@@ -25,15 +25,18 @@ async def run_service(
         }
     )).services
 
-    service_id = ""
+    service_id: Optional[str] = None
     if not services:
-        service_id = (await client.environments.automations.services.create(
+        service = (await client.environments.automations.services.create(   
             environment_id=environment_id,
             spec=spec,
             metadata=metadata
-        )).service.id
+        )).service
+        assert service is not None
+        service_id = service.id
     else:
         service_id = services[0].id
+    assert service_id is not None
 
     await client.environments.automations.services.start(id=service_id)
     log_url = await wait_for_service_log_url(client, environment_id, service_id)
@@ -47,9 +50,9 @@ async def run_command(client: AsyncGitpod, environment_id: str, command: str) ->
         }
     )).tasks
 
-    task_id = ""
+    task_id: Optional[str] = None
     if not tasks:
-        task_id = (await client.environments.automations.tasks.create(
+        task = (await client.environments.automations.tasks.create(
             spec={
                 "command": command,
             },
@@ -59,30 +62,40 @@ async def run_command(client: AsyncGitpod, environment_id: str, command: str) ->
                 "description": "Gitpod Python SDK Task",
                 "reference": TASK_REFERENCE,
             },
-        )).task.id
+        )).task
+        assert task is not None
+        task_id = task.id
     else:
         task_id = tasks[0].id
+        assert task_id is not None
         await client.environments.automations.tasks.update(
             id=task_id,
             spec={
                 "command": command,
             },
         )
-
-    task_execution_id = (await client.environments.automations.tasks.start(id=task_id)).task_execution.id
+    assert task_id is not None
+    task_execution = (await client.environments.automations.tasks.start(id=task_id)).task_execution
+    assert task_execution is not None
+    task_execution_id = task_execution.id
+    assert task_execution_id is not None
     log_url = await wait_for_task_log_url(client, environment_id, task_execution_id)
     return stream_logs(client, environment_id, log_url)
 
 async def wait_for_task_log_url(client: AsyncGitpod, environment_id: str, task_execution_id: str) -> str:
-    async def get_log_url():
+    async def get_log_url() -> Optional[str]:
         execution = (await client.environments.automations.tasks.executions.retrieve(id=task_execution_id)).task_execution
+        if not execution or not execution.status:
+            return None
         return execution.status.log_url
 
     return await wait_for_log_url(client, environment_id, task_execution_id, get_log_url, "RESOURCE_TYPE_TASK_EXECUTION")
 
 async def wait_for_service_log_url(client: AsyncGitpod, environment_id: str, service_id: str) -> str:
-    async def get_log_url():
+    async def get_log_url() -> Optional[str]:
         service = (await client.environments.automations.services.retrieve(id=service_id)).service
+        if not service or not service.status:
+            return None
         if service.status.phase != "SERVICE_PHASE_RUNNING":
             return None
         return service.status.log_url
@@ -108,6 +121,8 @@ async def wait_for_log_url(client: AsyncGitpod, environment_id: str, resource_id
     finally:
         await event_stream.http_response.aclose()
 
+    raise Exception("Failed to get log URL")
+
 async def stream_logs(client: AsyncGitpod, environment_id: str, log_url: str) -> AsyncIterator[str]:
     logs_access_token = (await client.environments.create_logs_token(environment_id=environment_id)).access_token
     async with httpx.AsyncClient() as http_client: