Modal

swerex.deployment.modal.ModalDeployment

ModalDeployment(*, logger: Logger | None = None, image: str | Image | PurePath, startup_timeout: float = 0.4, runtime_timeout: float = 1800.0, modal_sandbox_kwargs: dict[str, Any] | None = None, install_pipx: bool = True, deployment_timeout: float = 1800.0)

Bases: AbstractDeployment

Deployment for modal.com. The deployment will only start when the start method is being called.

Parameters:

Name	Type	Description	Default
image	str | Image | PurePath	Image to use for the deployment. One of the following: 1. modal.Image object 2. Path to a Dockerfile 3. Dockerhub image name (e.g. python:3.11-slim) 4. ECR image name (e.g. 123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:tag)	required
startup_timeout	float	The time to wait for the runtime to start.	0.4
runtime_timeout	float	The runtime timeout.	1800.0
deployment_timeout	float	The deployment timeout.	1800.0
modal_sandbox_kwargs	dict[str, Any] | None	Additional arguments to pass to modal.Sandbox.create	None

Source code in swerex/deployment/modal.py

def __init__(
    self,
    *,
    logger: logging.Logger | None = None,
    image: str | modal.Image | PurePath,
    startup_timeout: float = 0.4,
    runtime_timeout: float = 1800.0,
    modal_sandbox_kwargs: dict[str, Any] | None = None,
    install_pipx: bool = True,
    deployment_timeout: float = 1800.0,
):
    """Deployment for modal.com. The deployment will only start when the
    `start` method is being called.

    Args:
        image: Image to use for the deployment. One of the following:
            1. `modal.Image` object
            2. Path to a Dockerfile
            3. Dockerhub image name (e.g. `python:3.11-slim`)
            4. ECR image name (e.g. `123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:tag`)
        startup_timeout: The time to wait for the runtime to start.
        runtime_timeout: The runtime timeout.
        deployment_timeout: The deployment timeout.
        modal_sandbox_kwargs: Additional arguments to pass to `modal.Sandbox.create`
    """
    self._image = _ImageBuilder(install_pipx=install_pipx, logger=logger).auto(image)
    self._runtime: RemoteRuntime | None = None
    self._startup_timeout = startup_timeout
    self._sandbox: modal.Sandbox | None = None
    self._port = 8880
    self.logger = logger or get_logger("rex-deploy")
    self._app = modal.App.lookup("swe-rex", create_if_missing=True)
    self._user = _get_modal_user()
    self._runtime_timeout = runtime_timeout
    self._deployment_timeout = deployment_timeout
    if modal_sandbox_kwargs is None:
        modal_sandbox_kwargs = {}
    self._modal_kwargs = modal_sandbox_kwargs
    self._hooks = CombinedDeploymentHook()

_app instance-attribute

_app = lookup('swe-rex', create_if_missing=True)

_deployment_timeout instance-attribute

_deployment_timeout = deployment_timeout

_hooks instance-attribute

_hooks = CombinedDeploymentHook()

_image instance-attribute

_image = auto(image)

_modal_kwargs instance-attribute

_modal_kwargs = modal_sandbox_kwargs

_port instance-attribute

_port = 8880

_runtime instance-attribute

_runtime: RemoteRuntime | None = None

_runtime_timeout instance-attribute

_runtime_timeout = runtime_timeout

_sandbox instance-attribute

_sandbox: Sandbox | None = None

_startup_timeout instance-attribute

_startup_timeout = startup_timeout

_user instance-attribute

_user = _get_modal_user()

app property

app: App

Returns the modal app

Raises:

Type	Description
DeploymentNotStartedError	If the deployment is not started.

logger instance-attribute

logger = logger or get_logger('rex-deploy')

runtime property

runtime: RemoteRuntime

Returns the runtime if running.

Raises:

Type	Description
DeploymentNotStartedError	If the deployment was not started.

sandbox property

sandbox: Sandbox

Returns the modal sandbox

Raises:

Type	Description
DeploymentNotStartedError	If the deployment is not started.

_get_token

_get_token() -> str

Source code in swerex/deployment/modal.py

def _get_token(self) -> str:
    return str(uuid.uuid4())

_start_swerex_cmd

_start_swerex_cmd(token: str) -> str

Start swerex-server on the remote. If swerex is not installed arelady, install pipx and then run swerex-server with pipx run

Source code in swerex/deployment/modal.py

def _start_swerex_cmd(self, token: str) -> str:
    """Start swerex-server on the remote. If swerex is not installed arelady,
    install pipx and then run swerex-server with pipx run
    """
    rex_args = f"--port {self._port} --auth-token {token}"
    return f"{REMOTE_EXECUTABLE_NAME} {rex_args} || pipx run {PACKAGE_NAME} {rex_args}"

_wait_until_alive async

_wait_until_alive(timeout: float = 10.0)

Source code in swerex/deployment/modal.py

async def _wait_until_alive(self, timeout: float = 10.0):
    assert self._runtime is not None
    return await _wait_until_alive(self.is_alive, timeout=timeout, function_timeout=self._runtime._config.timeout)

add_hook

add_hook(hook: DeploymentHook)

Source code in swerex/deployment/modal.py

def add_hook(self, hook: DeploymentHook):
    self._hooks.add_hook(hook)

from_config classmethod

from_config(config: ModalDeploymentConfig) -> Self

Source code in swerex/deployment/modal.py

@classmethod
def from_config(cls, config: ModalDeploymentConfig) -> Self:
    return cls(
        image=config.image,
        install_pipx=config.install_pipx,
        startup_timeout=config.startup_timeout,
        runtime_timeout=config.runtime_timeout,
        deployment_timeout=config.deployment_timeout,
        modal_sandbox_kwargs=config.modal_sandbox_kwargs,
    )

get_modal_log_url

get_modal_log_url() -> str

Returns URL to modal logs

Raises:

Type	Description
DeploymentNotStartedError	If the deployment was not started.

Source code in swerex/deployment/modal.py

def get_modal_log_url(self) -> str:
    """Returns URL to modal logs

    Raises:
        DeploymentNotStartedError: If the deployment was not started.
    """
    return f"https://modal.com/apps/{self._user}/main/deployed/{self.app.name}?activeTab=logs&taskId={self.sandbox._get_task_id()}"

is_alive async

is_alive(*, timeout: float | None = None) -> IsAliveResponse

Checks if the runtime is alive. The return value can be tested with bool().

Raises:

Type	Description
DeploymentNotStartedError	If the deployment was not started.

Source code in swerex/deployment/modal.py

async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:
    """Checks if the runtime is alive. The return value can be
    tested with bool().

    Raises:
        DeploymentNotStartedError: If the deployment was not started.
    """
    if self._runtime is None or self._sandbox is None:
        raise DeploymentNotStartedError()
    if self._sandbox.poll() is not None:
        msg = "Container process terminated."
        output = "stdout:\n" + self._sandbox.stdout.read()  # type: ignore
        output += "\nstderr:\n" + self._sandbox.stderr.read()  # type: ignore
        msg += "\n" + output
        raise RuntimeError(msg)
    return await self._runtime.is_alive(timeout=timeout)

start async

start()

Starts the runtime.

Source code in swerex/deployment/modal.py

async def start(
    self,
):
    """Starts the runtime."""
    self.logger.info("Starting modal sandbox")
    self._hooks.on_custom_step("Starting modal sandbox")
    t0 = time.time()
    token = self._get_token()
    self._sandbox = modal.Sandbox.create(
        "/bin/bash",
        "-c",
        self._start_swerex_cmd(token),
        image=self._image,
        timeout=int(self._deployment_timeout),
        unencrypted_ports=[self._port],
        app=self._app,
        **self._modal_kwargs,
    )
    tunnel = self._sandbox.tunnels()[self._port]
    elapsed_sandbox_creation = time.time() - t0
    self.logger.info(f"Sandbox ({self._sandbox.object_id}) created in {elapsed_sandbox_creation:.2f}s")
    self.logger.info(f"Check sandbox logs at {self.get_modal_log_url()}")
    self.logger.info(f"Sandbox created with id {self._sandbox.object_id}")
    await asyncio.sleep(1)
    self.logger.info(f"Starting runtime at {tunnel.url}")
    self._hooks.on_custom_step("Starting runtime")
    self._runtime = RemoteRuntime(
        host=tunnel.url, timeout=self._runtime_timeout, auth_token=token, logger=self.logger
    )
    remaining_startup_timeout = max(0, self._startup_timeout - elapsed_sandbox_creation)
    t1 = time.time()
    await self._wait_until_alive(timeout=remaining_startup_timeout)
    self.logger.info(f"Runtime started in {time.time() - t1:.2f}s")

stop async

stop()

Stops the runtime.

Source code in swerex/deployment/modal.py

async def stop(self):
    """Stops the runtime."""
    if self._runtime is not None:
        await self._runtime.close()
        self._runtime = None
    if self._sandbox is not None and not self._sandbox.poll():
        self._sandbox.terminate()
    self._sandbox = None
    self._app = None