Fix: We can't access size_mib about Ephemeral and Persistant class because of some changes on aleph_message

Before any change, there were classes dedicated to the size of Ephemeral and Persistent class such as
PersistentVolumeSizeMb and EphemeralVolumeSizeMb.
Now, the size is included inside the classes EphemeralVolume and PersistentVolume so have to access the size
by another method.

- Creating a function to get the max and min size, that way it is more rigorous
- Calling this function to get max and min size of the different volumes

Author: Antonyjin

src/aleph_client/commands/instance/__init__.py

@@ -45,8 +45,9 @@
     NodeRequirements,
     TrustedExecutionEnvironment,
 )
-from aleph_message.models.execution.volume import PersistentVolumeSizeMib
+from aleph_message.models.execution.volume import PersistentVolume
 from aleph_message.models.item_hash import ItemHash
+from aleph_message.utils import Gigabytes
 from click import echo
 from rich import box
 from rich.console import Console
@@ -71,6 +72,7 @@
     filter_only_valid_messages,
     find_sevctl_or_exit,
     found_gpus_by_model,
+    get_field_constraints,
     get_or_prompt_volumes,
     setup_logging,
     str_to_datetime,
@@ -91,6 +93,7 @@
 metavar_valid_chains = f"[{'|'.join(hold_chains)}]"
 super_token_chains = get_chains_with_super_token()
 metavar_valid_payg_chains = f"[{'|'.join(super_token_chains)}]"
+_, max_rootfs_size = get_field_constraints(PersistentVolume, "size_mib", Gigabytes(2048), 0)
 
 
 @app.command()
@@ -118,9 +121,7 @@ async def create(
     compute_units: Annotated[Optional[int], typer.Option(help=help_strings.COMPUTE_UNITS)] = None,
     vcpus: Annotated[Optional[int], typer.Option(help=help_strings.VCPUS)] = None,
     memory: Annotated[Optional[int], typer.Option(help=help_strings.MEMORY)] = None,
-    rootfs_size: Annotated[
-        Optional[int], typer.Option(help=help_strings.ROOTFS_SIZE, max=PersistentVolumeSizeMib.le)
-    ] = None,
+    rootfs_size: Annotated[Optional[int], typer.Option(help=help_strings.ROOTFS_SIZE, max=max_rootfs_size)] = None,
     timeout_seconds: Annotated[float, typer.Option(help=help_strings.TIMEOUT_SECONDS)] = settings.DEFAULT_VM_TIMEOUT,
     ssh_pubkey_file: Annotated[Path, typer.Option(help=help_strings.SSH_PUBKEY_FILE)] = Path(
         "~/.ssh/id_rsa.pub"
@@ -341,7 +342,7 @@ async def create(
     disk_size_info = f"Rootfs Size: {round(disk_size/1024, 2)} GiB (defaulted to included storage in tier)"
     if not isinstance(rootfs_size, int):
         rootfs_size = validated_int_prompt(
-            "Custom Rootfs Size (MiB)", min_value=disk_size, max_value=PersistentVolumeSizeMib.le, default=disk_size
+            "Custom Rootfs Size (MiB)", min_value=disk_size, max_value=max_rootfs_size, default=disk_size
         )
     if rootfs_size > disk_size:
         disk_size = rootfs_size
@@ -1197,7 +1198,6 @@ async def confidential_init_session(
 
     # Generate sessions certificate files
     if not ((session_dir / "vm_godh.b64").exists() and keep_session):
-
         code, platform_file = await client.get_certificates()
         if code != 200:
             echo(
@@ -1367,9 +1367,7 @@ async def confidential_create(
     compute_units: Annotated[Optional[int], typer.Option(help=help_strings.COMPUTE_UNITS)] = None,
     vcpus: Annotated[Optional[int], typer.Option(help=help_strings.VCPUS)] = None,
     memory: Annotated[Optional[int], typer.Option(help=help_strings.MEMORY)] = None,
-    rootfs_size: Annotated[
-        Optional[int], typer.Option(help=help_strings.ROOTFS_SIZE, max=PersistentVolumeSizeMib.le)
-    ] = None,
+    rootfs_size: Annotated[Optional[int], typer.Option(help=help_strings.ROOTFS_SIZE, max=max_rootfs_size)] = None,
     timeout_seconds: Annotated[float, typer.Option(help=help_strings.TIMEOUT_SECONDS)] = settings.DEFAULT_VM_TIMEOUT,
     ssh_pubkey_file: Annotated[Path, typer.Option(help=help_strings.SSH_PUBKEY_FILE)] = Path(
         "~/.ssh/id_rsa.pub"
@@ -1515,9 +1513,7 @@ async def gpu_create(
     compute_units: Annotated[Optional[int], typer.Option(help=help_strings.COMPUTE_UNITS)] = None,
     vcpus: Annotated[Optional[int], typer.Option(help=help_strings.VCPUS)] = None,
     memory: Annotated[Optional[int], typer.Option(help=help_strings.MEMORY)] = None,
-    rootfs_size: Annotated[
-        Optional[int], typer.Option(help=help_strings.ROOTFS_SIZE, max=PersistentVolumeSizeMib.le)
-    ] = None,
+    rootfs_size: Annotated[Optional[int], typer.Option(help=help_strings.ROOTFS_SIZE, max=max_rootfs_size)] = None,
     premium: Annotated[Optional[bool], typer.Option(help=help_strings.GPU_PREMIUM_OPTION)] = None,
     timeout_seconds: Annotated[float, typer.Option(help=help_strings.TIMEOUT_SECONDS)] = settings.DEFAULT_VM_TIMEOUT,
     ssh_pubkey_file: Annotated[Path, typer.Option(help=help_strings.SSH_PUBKEY_FILE)] = Path(

src/aleph_client/commands/utils.py

@@ -18,11 +18,10 @@
 from aleph.sdk.types import GenericMessage
 from aleph.sdk.utils import safe_getattr
 from aleph_message.models import AlephMessage, InstanceMessage, ItemHash, ProgramMessage
-from aleph_message.models.execution.volume import (
-    EphemeralVolumeSize,
-    PersistentVolumeSizeMib,
-)
+from aleph_message.models.execution.volume import EphemeralVolume, PersistentVolume
 from aleph_message.status import MessageStatus
+from aleph_message.utils import Gigabytes
+from pydantic import BaseModel
 from pygments import highlight
 from pygments.formatters.terminal256 import Terminal256Formatter
 from pygments.lexers import JsonLexer
@@ -89,6 +88,14 @@ def is_valid_mount_path(mount: str) -> bool:
     return mount.startswith("/") and len(mount) > 1
 
 
+def get_field_constraints(model: type[BaseModel], field_name: str, default_min: int, default_max: int):
+    """Get the constraints for a field in a model"""
+    field_metadata = model.model_fields[field_name].metadata
+    max_value = next((meta.le for meta in field_metadata if hasattr(meta, "le")), default_max)
+    min_value = next((meta.gt for meta in field_metadata if hasattr(meta, "gt")), default_min)
+    return min_value, max_value
+
+
 def prompt_for_volumes():
     while yes_no_input("Add volume?", default=False):
         mount = validated_prompt("Mount path (must be absolute, ex: /opt/data)", is_valid_mount_path)
@@ -108,10 +115,11 @@ def prompt_for_volumes():
             if yes_no_input("Copy from a parent volume?", default=False):
                 parent = {"ref": validated_prompt("Item hash", lambda text: len(text) == 64), "use_latest": True}
             name = validated_prompt("Name", lambda text: len(text) > 0)
+            min_size, max_size = get_field_constraints(PersistentVolume, "size_mib", Gigabytes(2048), 0)
             size_mib = validated_int_prompt(
                 "Size (MiB)",
-                min_value=PersistentVolumeSizeMib.gt + 1,
-                max_value=PersistentVolumeSizeMib.le,
+                min_value=max_size,
+                max_value=min_size,
             )
             yield {
                 **base_volume,
@@ -121,9 +129,8 @@ def prompt_for_volumes():
                 "size_mib": size_mib,
             }
         else:  # Ephemeral
-            size_mib = validated_int_prompt(
-                "Size (MiB)", min_value=EphemeralVolumeSize.gt + 1, max_value=EphemeralVolumeSize.le
-            )
+            min_size, max_size = get_field_constraints(EphemeralVolume, "size_mib", Gigabytes(1), 0)
+            size_mib = validated_int_prompt("Size (MiB)", min_value=min_size + 1, max_value=max_size)
             yield {
                 **base_volume,
                 "ephemeral": True,