[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

src/swerex/deployment/config.py

@@ -186,6 +186,7 @@ def get_deployment(self) -> AbstractDeployment:
 
         return DummyDeployment.from_config(self)
 
+
 class VeFaasDeploymentConfig(BaseModel):
     image: str = "python:3.11"
     platform: str | None = None
@@ -234,6 +235,7 @@ def get_deployment(self) -> AbstractDeployment:
 
         return VeFaasDeployment.from_config(self)
 
+
 class DaytonaDeploymentConfig(BaseModel):
     """Configuration for Daytona deployment."""
 

src/swerex/deployment/vefaas.py

@@ -1,45 +1,36 @@
 import logging
 import time
 import uuid
-
 from typing import Any
-from typing_extensions import Self
 
-from swerex.exceptions import DeploymentNotStartedError, DeploymentStartupError
+from typing_extensions import Self
+from volcenginesdkapig import APIGApi
+from volcenginesdkapig.models import (
+    GetGatewayServiceRequest,
+    GetGatewayServiceResponse,
+)
+from volcenginesdkcore import ApiClient, Configuration
+from volcenginesdkvefaas import (
+    CreateSandboxRequest,
+    CreateSandboxResponse,
+    InstanceImageInfoForCreateSandboxInput,
+    KillSandboxRequest,
+    VEFAASApi,
+)
 
 from swerex.deployment.abstract import AbstractDeployment
 from swerex.deployment.config import VeFaasDeploymentConfig
-from swerex.deployment.hooks.abstract import DeploymentHook, CombinedDeploymentHook
+from swerex.deployment.hooks.abstract import CombinedDeploymentHook, DeploymentHook
+from swerex.exceptions import DeploymentNotStartedError, DeploymentStartupError
 from swerex.runtime.abstract import IsAliveResponse
-from swerex.runtime.remote import RemoteRuntime
 from swerex.runtime.config import RemoteRuntimeConfig
-
+from swerex.runtime.remote import RemoteRuntime
 from swerex.utils.log import get_logger
 from swerex.utils.wait import _wait_until_alive
 
 
-from volcenginesdkvefaas import (
-    VEFAASApi,
-    InstanceImageInfoForCreateSandboxInput,
-    CreateSandboxRequest, 
-    CreateSandboxResponse,
-    KillSandboxRequest,
-    )
-from volcenginesdkcore import ApiClient, Configuration
-
-from volcenginesdkapig import APIGApi
-from volcenginesdkapig.models import (
-    GetGatewayServiceRequest,
-    GetGatewayServiceResponse,
-)
-
 class VeFaasDeployment(AbstractDeployment):
-    def __init__(
-        self,
-        *,
-        logger: logging.Logger | None = None,
-        **kwargs: Any
-    ):
+    def __init__(self, *, logger: logging.Logger | None = None, **kwargs: Any):
         self._config = VeFaasDeploymentConfig(**kwargs)
         self._runtime: RemoteRuntime | None = None
         self._container_name = None
@@ -53,13 +44,13 @@ def __init__(
     @classmethod
     def from_config(cls, config: VeFaasDeploymentConfig) -> Self:
         return cls(**config.model_dump())
-    
+
     def add_hook(self, hook: DeploymentHook):
         self._hooks.add_hook(hook)
-    
+
     def _get_token(self) -> str:
         return str(uuid.uuid4())
-    
+
     async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:
         """Checks if the runtime is alive. The return value can be
         tested with bool().
@@ -71,7 +62,7 @@ async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:
             msg = "Runtime not started"
             raise RuntimeError(msg)
         return await self._runtime.is_alive(timeout=timeout)
-    
+
     async def _wait_until_alive(self, timeout: float = 10.0):
         try:
             return await _wait_until_alive(self.is_alive, timeout=timeout, function_timeout=self._runtime_timeout)
@@ -83,7 +74,7 @@ async def _wait_until_alive(self, timeout: float = 10.0):
     def _get_domain(self, apigs_id):
         api_instance = APIGApi(self._get_api_client())
         req = GetGatewayServiceRequest(
-            id = apigs_id,
+            id=apigs_id,
         )
         response = api_instance.get_gateway_service(req)
         if not isinstance(response, GetGatewayServiceResponse):
@@ -100,25 +91,25 @@ def _get_domain(self, apigs_id):
 
     def _get_container_name(self) -> str:
         """Returns a unique container name based on the image name."""
-        image_str = self._config.image.split('/')
-        image_name_sanitized = image_str[-1].replace('_', '-')
-        image_name_sanitized = image_name_sanitized.replace(':', '-')
-        image_name_sanitized = image_name_sanitized.replace('.', '-')
+        image_str = self._config.image.split("/")
+        image_name_sanitized = image_str[-1].replace("_", "-")
+        image_name_sanitized = image_name_sanitized.replace(":", "-")
+        image_name_sanitized = image_name_sanitized.replace(".", "-")
 
         return image_name_sanitized[:-14]
 
     def _get_api_client(self) -> ApiClient:
         if self._api_client:
             return self._api_client
-        
+
         access_key = self._config.ak
         secret_key = self._config.sk
         region = self._config.region
-        
+
         if not access_key or not secret_key:
             emsg = "VOLCENGINE_ACCESS_KEY and VOLCENGINE_SECRET_KEY must be set"
             raise DeploymentStartupError(emsg)
-        
+
         config = Configuration()
         config.ak = access_key
         config.sk = secret_key
@@ -131,11 +122,7 @@ def _get_api_client(self) -> ApiClient:
     async def create_sandbox(self, function_id, image, cmd, request_timeout) -> str:
         client = VEFAASApi(self._get_api_client())
 
-        instance_image_info = InstanceImageInfoForCreateSandboxInput(
-            image=image,
-            port=8000,
-            command=cmd
-        )
+        instance_image_info = InstanceImageInfoForCreateSandboxInput(image=image, port=8000, command=cmd)
 
         response = client.create_sandbox(
             CreateSandboxRequest(
@@ -152,54 +139,51 @@ async def create_sandbox(self, function_id, image, cmd, request_timeout) -> str:
             emsg = "Failed to create sandbox: no sandbox id"
             raise DeploymentStartupError(emsg)
         return response.sandbox_id
-    
+
     async def kill_sandbox(self) -> str:
         client = VEFAASApi(self._get_api_client())
 
         if self._sandbox_id:
             response = client.kill_sandbox(
-                KillSandboxRequest(
-                    function_id=self._config.function_id,
-                    sandbox_id=self._sandbox_id
-                )
-            )        
+                KillSandboxRequest(function_id=self._config.function_id, sandbox_id=self._sandbox_id)
+            )
             if not isinstance(response, CreateSandboxResponse):
-                self.logger.warning(
-                    f"Kill Sandbox {self._sandbox_id} Failed"
-                )
+                self.logger.warning(f"Kill Sandbox {self._sandbox_id} Failed")
         self._sandbox_id = ""
 
     async def start(self):
-        """ Start Faas runtime"""
+        """Start Faas runtime"""
 
         assert self._container_name is None
         self._container_name = self._get_container_name()
 
-        self.logger.info(
-            f"Starting container {self._container_name}"
-        )
+        self.logger.info(f"Starting container {self._container_name}")
 
         # Gen swe-rex command
         token = self._get_token()
         mount_path = self._config.mount_path
         cmd = f"source {mount_path}/SWE-ReX/bin/activate && python3 {mount_path}/SWE-ReX/src/swerex/server.py --auth-token {token}"
 
         # create sandbox
-        sandbox_id = await self.create_sandbox(self._config.function_id, self._config.image, cmd, self._config.request_timeout)
+        sandbox_id = await self.create_sandbox(
+            self._config.function_id, self._config.image, cmd, self._config.request_timeout
+        )
         self._sandbox_id = sandbox_id
 
         domain = self._get_domain(self._config.apigateway_service_id)
 
         self._runtime = RemoteRuntime.from_config(
-            RemoteRuntimeConfig(host=domain, timeout=self._runtime_timeout, auth_token=token, faas_instance_name=self._sandbox_id)
+            RemoteRuntimeConfig(
+                host=domain, timeout=self._runtime_timeout, auth_token=token, faas_instance_name=self._sandbox_id
+            )
         )
 
         t0 = time.time()
         await self._wait_until_alive(timeout=self._config.startup_timeout)
         self.logger.info(f"Runtime started in {time.time() - t0:.2f}s")
 
     async def stop(self):
-        """ Stop the runtime """
+        """Stop the runtime"""
         if self._runtime is not None:
             await self._runtime.close()
             self._runtime = None

tests/test_faas_deployment.py

@@ -2,6 +2,7 @@
 
 from swerex.deployment.vefaas import VeFaasDeployment
 
+
 async def test_faas_deployment():
     f = VeFaasDeployment(
         image="enterprise-public-cn-beijing.cr.volces.com/swe-bench/sweb.eval.x86_64.django_1776_django-15414:latest",
@@ -10,9 +11,9 @@ async def test_faas_deployment():
         region="cn-beijing",
         function_id="awokjltn",
         apigateway_service_id="sd2on64i5ni4n75n9unpg",
-        )
+    )
     with pytest.raises(RuntimeError):
         await f.is_alive()
     await f.start()
     assert await f.is_alive()
-    await f.stop()
+    await f.stop()