Add create_instance method (#78)

* Add allow_amend, internet and aleph_api parameters to create_program

* Add method to create VM instances

* Fix formatting and interface issues

* Fix create_program

Author: MHHukiewitz

src/aleph/sdk/client/abstract.py

@@ -348,6 +348,54 @@ async def create_program(
         """
         pass
 
+    @abstractmethod
+    async def create_instance(
+        self,
+        rootfs: str,
+        rootfs_size: int,
+        rootfs_name: str,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        storage_engine: StorageEnum = StorageEnum.storage,
+        channel: Optional[str] = None,
+        address: Optional[str] = None,
+        sync: bool = False,
+        memory: Optional[int] = None,
+        vcpus: Optional[int] = None,
+        timeout_seconds: Optional[float] = None,
+        allow_amend: bool = False,
+        internet: bool = True,
+        aleph_api: bool = True,
+        encoding: Encoding = Encoding.zip,
+        volumes: Optional[List[Mapping]] = None,
+        volume_persistence: str = "host",
+        ssh_keys: Optional[List[str]] = None,
+        metadata: Optional[Mapping[str, Any]] = None,
+    ) -> Tuple[AlephMessage, MessageStatus]:
+        """
+        Post a (create) PROGRAM message.
+
+        :param rootfs: Root filesystem to use
+        :param rootfs_size: Size of root filesystem
+        :param rootfs_name: Name of root filesystem
+        :param environment_variables: Environment variables to pass to the program
+        :param storage_engine: Storage engine to use (Default: "storage")
+        :param channel: Channel to use (Default: "TEST")
+        :param address: Address to use (Default: account.get_address())
+        :param sync: If true, waits for the message to be processed by the API server
+        :param memory: Memory in MB for the VM to be allocated (Default: 128)
+        :param vcpus: Number of vCPUs to allocate (Default: 1)
+        :param timeout_seconds: Timeout in seconds (Default: 30.0)
+        :param allow_amend: Whether the deployed VM image may be changed (Default: False)
+        :param internet: Whether the VM should have internet connectivity. (Default: True)
+        :param aleph_api: Whether the VM needs access to Aleph messages API (Default: True)
+        :param encoding: Encoding to use (Default: Encoding.zip)
+        :param volumes: Volumes to mount
+        :param volume_persistence: Where volumes are persisted, can be "host" or "store", meaning distributed across Aleph.im (Default: "host")
+        :param ssh_keys: SSH keys to authorize access to the VM
+        :param metadata: Metadata to attach to the message
+        """
+        pass
+
     @abstractmethod
     async def forget(
         self,

src/aleph/sdk/client/authenticated_http.py

@@ -13,6 +13,8 @@
     AlephMessage,
     ForgetContent,
     ForgetMessage,
+    InstanceContent,
+    InstanceMessage,
     ItemType,
     MessageType,
     PostContent,
@@ -27,8 +29,9 @@
     FunctionEnvironment,
     MachineResources,
 )
+from aleph_message.models.execution.instance import RootfsVolume
 from aleph_message.models.execution.program import CodeContent, FunctionRuntime
-from aleph_message.models.execution.volume import MachineVolume
+from aleph_message.models.execution.volume import MachineVolume, ParentVolume
 from aleph_message.status import MessageStatus
 from pydantic.json import pydantic_encoder
 
@@ -463,7 +466,7 @@ async def create_program(
                 if runtime == settings.DEFAULT_RUNTIME_ID
                 else "",
             ),
-            volumes=volumes,
+            volumes=[MachineVolume.parse_obj(volume) for volume in volumes],
             time=time.time(),
             metadata=metadata,
         )
@@ -479,6 +482,76 @@ async def create_program(
             sync=sync,
         )
 
+    async def create_instance(
+        self,
+        rootfs: str,
+        rootfs_size: int,
+        rootfs_name: str,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        storage_engine: StorageEnum = StorageEnum.storage,
+        channel: Optional[str] = None,
+        address: Optional[str] = None,
+        sync: bool = False,
+        memory: Optional[int] = None,
+        vcpus: Optional[int] = None,
+        timeout_seconds: Optional[float] = None,
+        allow_amend: bool = False,
+        internet: bool = True,
+        aleph_api: bool = True,
+        encoding: Encoding = Encoding.zip,
+        volumes: Optional[List[Mapping]] = None,
+        volume_persistence: str = "host",
+        ssh_keys: Optional[List[str]] = None,
+        metadata: Optional[Mapping[str, Any]] = None,
+    ) -> Tuple[InstanceMessage, MessageStatus]:
+        address = address or settings.ADDRESS_TO_USE or self.account.get_address()
+
+        volumes = volumes if volumes is not None else []
+        memory = memory or settings.DEFAULT_VM_MEMORY
+        vcpus = vcpus or settings.DEFAULT_VM_VCPUS
+        timeout_seconds = timeout_seconds or settings.DEFAULT_VM_TIMEOUT
+
+        content = InstanceContent(
+            address=address,
+            allow_amend=allow_amend,
+            environment=FunctionEnvironment(
+                reproducible=False,
+                internet=internet,
+                aleph_api=aleph_api,
+            ),
+            variables=environment_variables,
+            resources=MachineResources(
+                vcpus=vcpus,
+                memory=memory,
+                seconds=timeout_seconds,
+            ),
+            rootfs=RootfsVolume(
+                parent=ParentVolume(
+                    ref=rootfs,
+                    use_latest=True,
+                ),
+                name=rootfs_name,
+                size_mib=rootfs_size,
+                persistence="host",
+                use_latest=True,
+                comment="Official Aleph Debian root filesystem"
+                if rootfs == settings.DEFAULT_RUNTIME_ID
+                else "",
+            ),
+            volumes=[MachineVolume.parse_obj(volume) for volume in volumes],
+            time=time.time(),
+            authorized_keys=ssh_keys,
+            metadata=metadata,
+        )
+
+        return await self.submit(
+            content=content.dict(exclude_none=True),
+            message_type=MessageType.instance,
+            channel=channel,
+            storage_engine=storage_engine,
+            sync=sync,
+        )
+
     async def forget(
         self,
         hashes: List[str],

tests/unit/test_asynchronous.py

@@ -5,6 +5,7 @@
 from aleph_message.models import (
     AggregateMessage,
     ForgetMessage,
+    InstanceMessage,
     PostMessage,
     ProgramMessage,
     StoreMessage,
@@ -141,6 +142,21 @@ async def test_create_program(mock_session_with_post_success):
     assert isinstance(program_message, ProgramMessage)
 
 
+@pytest.mark.asyncio
+async def test_create_instance(mock_session_with_post_success):
+    async with mock_session_with_post_success as session:
+        instance_message, message_status = await session.create_instance(
+            rootfs="cafecafecafecafecafecafecafecafecafecafecafecafecafecafecafecafe",
+            rootfs_size=1,
+            rootfs_name="rootfs",
+            channel="TEST",
+            metadata={"tags": ["test"]},
+        )
+
+    assert mock_session_with_post_success.http_session.post.called_once
+    assert isinstance(instance_message, InstanceMessage)
+
+
 @pytest.mark.asyncio
 async def test_forget(mock_session_with_post_success):
     async with mock_session_with_post_success as session: