Abstract

swerex.deployment.abstract.AbstractDeployment

AbstractDeployment(*args, **kwargs)

Bases: ABC

Source code in swerex/deployment/abstract.py

def __init__(self, *args, **kwargs):
    self.logger: logging.Logger

logger instance-attribute

logger: Logger

runtime abstractmethod property

runtime: AbstractRuntime

Returns the runtime if running.

Raises:

Type	Description
DeploymentNotStartedError	If the deployment was not started.

__del__

__del__()

Stops the runtime when the object is deleted.

Source code in swerex/deployment/abstract.py

def __del__(self):
    """Stops the runtime when the object is deleted."""
    # Need to be check whether we are in an async event loop or not
    # https://stackoverflow.com/questions/54770360/
    self.logger.debug("Ensuring deployment is stopped because object is deleted")
    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            loop.create_task(self.stop())
        else:
            loop.run_until_complete(self.stop())
    except Exception:
        pass

add_hook abstractmethod

add_hook(hook: DeploymentHook)

Source code in swerex/deployment/abstract.py

@abstractmethod
def add_hook(self, hook: DeploymentHook): ...

is_alive abstractmethod 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/abstract.py

@abstractmethod
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.
    """

start abstractmethod async

start(*args, **kwargs)

Starts the runtime.

Source code in swerex/deployment/abstract.py

@abstractmethod
async def start(self, *args, **kwargs):
    """Starts the runtime."""

stop abstractmethod async

stop(*args, **kwargs)

Stops the runtime.

Source code in swerex/deployment/abstract.py

@abstractmethod
async def stop(self, *args, **kwargs):
    """Stops the runtime."""