From 41575ac2c9d98576e358d9225201a9ae705f2b00 Mon Sep 17 00:00:00 2001 From: philogicae Date: Tue, 18 Feb 2025 19:35:04 +0200 Subject: [PATCH] Wrap all cmd args in Annotated + minor fix on file list (#339) - Wrap all command args in Annotated to allow to easily import CLI as a lib --- src/aleph_client/commands/account.py | 56 ++- src/aleph_client/commands/domain.py | 50 +- src/aleph_client/commands/files.py | 96 ++-- .../commands/instance/__init__.py | 431 +++++++++--------- src/aleph_client/commands/message.py | 96 ++-- src/aleph_client/commands/node.py | 32 +- src/aleph_client/commands/program.py | 192 ++++---- tests/unit/test_commands.py | 13 + 8 files changed, 518 insertions(+), 448 deletions(-) diff --git a/src/aleph_client/commands/account.py b/src/aleph_client/commands/account.py index 3ac4d9b0..85c20657 100644 --- a/src/aleph_client/commands/account.py +++ b/src/aleph_client/commands/account.py @@ -3,7 +3,7 @@ import asyncio import logging from pathlib import Path -from typing import Optional +from typing import Annotated, Optional import aiohttp import typer @@ -46,12 +46,12 @@ @app.command() async def create( - private_key: Optional[str] = typer.Option(None, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(None, help=help_strings.PRIVATE_KEY_FILE), - chain: Optional[Chain] = typer.Option(default=None, help=help_strings.ORIGIN_CHAIN), - replace: bool = typer.Option(default=False, help=help_strings.CREATE_REPLACE), - active: bool = typer.Option(default=True, help=help_strings.CREATE_ACTIVE), - debug: bool = False, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = None, + private_key_file: Annotated[Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE)] = None, + chain: Annotated[Optional[Chain], typer.Option(help=help_strings.ORIGIN_CHAIN)] = None, + replace: Annotated[bool, typer.Option(help=help_strings.CREATE_REPLACE)] = False, + active: Annotated[bool, typer.Option(help=help_strings.CREATE_ACTIVE)] = True, + debug: Annotated[bool, typer.Option()] = False, ): """Create or import a private key.""" @@ -120,8 +120,10 @@ async def create( @app.command(name="address") def display_active_address( - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, ): """ Display your public address(es). @@ -183,8 +185,10 @@ def path_directory(): @app.command() def show( - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, ): """Display current configuration.""" @@ -194,8 +198,8 @@ def show( @app.command() def export_private_key( - private_key: Optional[str] = typer.Option(None, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(None, help=help_strings.PRIVATE_KEY_FILE), + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = None, + private_key_file: Annotated[Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE)] = None, ): """ Display your private key. @@ -220,11 +224,13 @@ def export_private_key( @app.command("sign-bytes") def sign_bytes( - message: Optional[str] = typer.Option(None, help="Message to sign"), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - chain: Optional[Chain] = typer.Option(None, help=help_strings.ADDRESS_CHAIN), - debug: bool = False, + message: Annotated[Optional[str], typer.Option(help="Message to sign")] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + chain: Annotated[Optional[Chain], typer.Option(help=help_strings.ADDRESS_CHAIN)] = None, + debug: Annotated[bool, typer.Option()] = False, ): """Sign a message using your private key.""" @@ -257,10 +263,12 @@ async def get_balance(address: str) -> dict: @app.command() async def balance( - address: Optional[str] = typer.Option(None, help="Address"), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - chain: Optional[Chain] = typer.Option(None, help=help_strings.ADDRESS_CHAIN), + address: Annotated[Optional[str], typer.Option(help="Address")] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + chain: Annotated[Optional[Chain], typer.Option(help=help_strings.ADDRESS_CHAIN)] = None, ): """Display your ALEPH balance.""" account = _load_account(private_key, private_key_file, chain=chain) @@ -363,8 +371,8 @@ async def list_accounts(): @app.command(name="config") async def configure( - private_key_file: Optional[Path] = typer.Option(None, help="New path to the private key file"), - chain: Optional[Chain] = typer.Option(None, help="New active chain"), + private_key_file: Annotated[Optional[Path], typer.Option(help="New path to the private key file")] = None, + chain: Annotated[Optional[Chain], typer.Option(help="New active chain")] = None, ): """Configure current private key file and active chain (default selection)""" diff --git a/src/aleph_client/commands/domain.py b/src/aleph_client/commands/domain.py index 83c74b86..538bdcbd 100644 --- a/src/aleph_client/commands/domain.py +++ b/src/aleph_client/commands/domain.py @@ -3,7 +3,7 @@ import logging from pathlib import Path from time import sleep -from typing import Optional, cast +from typing import Annotated, Optional, cast import typer from aleph.sdk.account import _load_account @@ -176,13 +176,15 @@ async def detach_resource(account: AccountFromPrivateKey, fqdn: Hostname, intera @app.command() async def add( - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - fqdn: str = typer.Argument(..., help=help_strings.CUSTOM_DOMAIN_NAME), - target: Optional[TargetType] = typer.Option(None, help=help_strings.CUSTOM_DOMAIN_TARGET_TYPES), - item_hash: Optional[str] = typer.Option(None, help=help_strings.CUSTOM_DOMAIN_ITEM_HASH), - owner: Optional[str] = typer.Option(None, help=help_strings.CUSTOM_DOMAIN_OWNER_ADDRESS), - ask: bool = typer.Option(default=True, help=help_strings.ASK_FOR_CONFIRMATION), + fqdn: Annotated[str, typer.Argument(help=help_strings.CUSTOM_DOMAIN_NAME)], + target: Annotated[Optional[TargetType], typer.Option(help=help_strings.CUSTOM_DOMAIN_TARGET_TYPES)] = None, + item_hash: Annotated[Optional[str], typer.Option(help=help_strings.CUSTOM_DOMAIN_ITEM_HASH)] = None, + owner: Annotated[Optional[str], typer.Option(help=help_strings.CUSTOM_DOMAIN_OWNER_ADDRESS)] = None, + ask: Annotated[bool, typer.Option(help=help_strings.ASK_FOR_CONFIRMATION)] = True, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, ): """Add and link a Custom Domain.""" account: AccountFromPrivateKey = _load_account(private_key, private_key_file) @@ -260,12 +262,14 @@ async def add( @app.command() async def attach( - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - fqdn: str = typer.Argument(..., help=help_strings.CUSTOM_DOMAIN_NAME), - item_hash: Optional[str] = typer.Option(None, help=help_strings.CUSTOM_DOMAIN_ITEM_HASH), - catch_all_path: str = typer.Option(default=None, help=help_strings.IPFS_CATCH_ALL_PATH), - ask: bool = typer.Option(default=True, help=help_strings.ASK_FOR_CONFIRMATION), + fqdn: Annotated[str, typer.Argument(help=help_strings.CUSTOM_DOMAIN_NAME)], + item_hash: Annotated[Optional[str], typer.Option(help=help_strings.CUSTOM_DOMAIN_ITEM_HASH)] = None, + catch_all_path: Annotated[Optional[str], typer.Option(help=help_strings.IPFS_CATCH_ALL_PATH)] = None, + ask: Annotated[bool, typer.Option(help=help_strings.ASK_FOR_CONFIRMATION)] = True, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, ): """Attach resource to a Custom Domain.""" account: AccountFromPrivateKey = _load_account(private_key, private_key_file) @@ -282,10 +286,12 @@ async def attach( @app.command() async def detach( - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - fqdn: str = typer.Argument(..., help=help_strings.CUSTOM_DOMAIN_NAME), - ask: bool = typer.Option(default=True, help=help_strings.ASK_FOR_CONFIRMATION), + fqdn: Annotated[str, typer.Argument(help=help_strings.CUSTOM_DOMAIN_NAME)], + ask: Annotated[bool, typer.Option(help=help_strings.ASK_FOR_CONFIRMATION)] = True, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, ): """Unlink Custom Domain.""" account: AccountFromPrivateKey = _load_account(private_key, private_key_file) @@ -296,9 +302,11 @@ async def detach( @app.command() async def info( - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - fqdn: str = typer.Argument(..., help=help_strings.CUSTOM_DOMAIN_NAME), + fqdn: Annotated[str, typer.Argument(help=help_strings.CUSTOM_DOMAIN_NAME)], + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, ): """Show Custom Domain Details.""" account: AccountFromPrivateKey = _load_account(private_key, private_key_file) diff --git a/src/aleph_client/commands/files.py b/src/aleph_client/commands/files.py index d0f725cb..f991394c 100644 --- a/src/aleph_client/commands/files.py +++ b/src/aleph_client/commands/files.py @@ -4,7 +4,7 @@ import logging from datetime import datetime from pathlib import Path -from typing import Optional +from typing import Annotated, Optional import aiohttp import typer @@ -30,12 +30,14 @@ @app.command() async def pin( - item_hash: str = typer.Argument(..., help="IPFS hash to pin on aleph.im"), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - ref: Optional[str] = typer.Option(None, help=help_strings.REF), - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="IPFS hash to pin on aleph.im")], + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + ref: Annotated[Optional[str], typer.Option(help=help_strings.REF)] = None, + debug: Annotated[bool, typer.Option()] = False, ): """Persist a file from IPFS on aleph.im.""" @@ -58,12 +60,14 @@ async def pin( @app.command() async def upload( - path: Path = typer.Argument(..., help="Path of the file to upload"), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - ref: Optional[str] = typer.Option(None, help=help_strings.REF), - debug: bool = False, + path: Annotated[Path, typer.Argument(help="Path of the file to upload")], + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + ref: Annotated[Optional[str], typer.Option(help=help_strings.REF)] = None, + debug: Annotated[bool, typer.Option()] = False, ): """Upload and store a file on aleph.im.""" @@ -97,14 +101,14 @@ async def upload( @app.command() async def download( - hash: str = typer.Argument(..., help="hash to download from aleph."), - use_ipfs: bool = typer.Option(default=False, help="Download using IPFS instead of storage"), - output_path: Path = typer.Option(Path("."), help="Output directory path"), - file_name: str = typer.Option(None, help="Output file name (without extension)"), - file_extension: str = typer.Option(None, help="Output file extension"), - only_info: bool = False, - verbose: bool = True, - debug: bool = False, + hash: Annotated[str, typer.Argument(help="hash to download from aleph.")], + use_ipfs: Annotated[bool, typer.Option(help="Download using IPFS instead of storage")] = False, + output_path: Annotated[Path, typer.Option(help="Output directory path")] = Path("."), + file_name: Annotated[Optional[str], typer.Option(help="Output file name (without extension)")] = None, + file_extension: Annotated[Optional[str], typer.Option(help="Output file extension")] = None, + only_info: Annotated[bool, typer.Option()] = False, + verbose: Annotated[bool, typer.Option()] = True, + debug: Annotated[bool, typer.Option()] = False, ) -> Optional[StoredContent]: """Download a file from aleph.im or display related infos.""" @@ -142,14 +146,19 @@ async def download( @app.command() async def forget( - item_hash: str = typer.Argument( - ..., help="Hash(es) to forget. Must be a comma separated list. Example: `123...abc` or `123...abc,456...xyz`" - ), - reason: str = typer.Argument("User deletion", help="reason to forget"), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + item_hash: Annotated[ + str, + typer.Argument( + help="Hash(es) to forget. Must be a comma separated list. Example: `123...abc` or `123...abc,456...xyz`" + ), + ], + reason: Annotated[str, typer.Argument(help="reason to forget")] = "User deletion", + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option()] = False, ): """forget a file and his message on aleph.im.""" @@ -225,20 +234,23 @@ def _show_files(files_data: dict) -> None: console.print(table) -@app.command() -async def list( - address: Optional[str] = typer.Option(None, help="Address"), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - pagination: int = typer.Option(100, help="Maximum number of files to return."), - page: int = typer.Option(1, help="Offset in pages."), - sort_order: int = typer.Option( - -1, - help=( - "Order in which files should be listed: -1 means most recent messages first, 1 means older messages first." +@app.command(name="list") +async def list_files( + address: Annotated[Optional[str], typer.Option(help="Address")] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + pagination: Annotated[int, typer.Option(help="Maximum number of files to return.")] = 100, + page: Annotated[int, typer.Option(help="Offset in pages.")] = 1, + sort_order: Annotated[ + int, + typer.Option( + help="Order in which files should be listed: -1 means most recent messages first," + " 1 means older messages first." ), - ), - json: bool = typer.Option(default=False, help="Print as json instead of rich table"), + ] = -1, + json: Annotated[bool, typer.Option(help="Print as json instead of rich table")] = False, ): """List all files for a given address""" account: AccountFromPrivateKey = _load_account(private_key, private_key_file) diff --git a/src/aleph_client/commands/instance/__init__.py b/src/aleph_client/commands/instance/__init__.py index 3e5385e8..fd55e395 100644 --- a/src/aleph_client/commands/instance/__init__.py +++ b/src/aleph_client/commands/instance/__init__.py @@ -7,7 +7,7 @@ import shutil from decimal import Decimal from pathlib import Path -from typing import Any, Optional, Union, cast +from typing import Annotated, Any, Optional, Union, cast import aiohttp import typer @@ -93,57 +93,56 @@ @app.command() async def create( - payment_type: Optional[str] = typer.Option( - None, - help=help_strings.PAYMENT_TYPE, - callback=lambda pt: None if pt is None else pt.lower(), - metavar=metavar_valid_payment_types, - case_sensitive=False, - ), - payment_chain: Optional[Chain] = typer.Option( - None, - help=help_strings.PAYMENT_CHAIN, - metavar=metavar_valid_chains, - case_sensitive=False, - ), - hypervisor: HypervisorType = typer.Option(HypervisorType.qemu, help=help_strings.HYPERVISOR), - name: Optional[str] = typer.Option(None, help=help_strings.INSTANCE_NAME), - rootfs: Optional[str] = typer.Option(None, help=help_strings.ROOTFS), - compute_units: Optional[int] = typer.Option(None, help=help_strings.COMPUTE_UNITS), - vcpus: Optional[int] = typer.Option(None, help=help_strings.VCPUS), - memory: Optional[int] = typer.Option(None, help=help_strings.MEMORY), - rootfs_size: Optional[int] = typer.Option(None, help=help_strings.ROOTFS_SIZE), - timeout_seconds: float = typer.Option( - settings.DEFAULT_VM_TIMEOUT, - help=help_strings.TIMEOUT_SECONDS, - ), - ssh_pubkey_file: Path = typer.Option( - Path("~/.ssh/id_rsa.pub").expanduser(), - help=help_strings.SSH_PUBKEY_FILE, - ), - address: Optional[str] = typer.Option(None, help=help_strings.ADDRESS_PAYER), - crn_hash: Optional[str] = typer.Option(None, help=help_strings.CRN_HASH), - crn_url: Optional[str] = typer.Option(None, help=help_strings.CRN_URL), - confidential: bool = typer.Option(False, help=help_strings.CONFIDENTIAL_OPTION), - confidential_firmware: str = typer.Option( - default=settings.DEFAULT_CONFIDENTIAL_FIRMWARE, help=help_strings.CONFIDENTIAL_FIRMWARE - ), - gpu: bool = typer.Option(False, help=help_strings.GPU_OPTION), - premium: Optional[bool] = typer.Option(None, help=help_strings.GPU_PREMIUM_OPTION), - skip_volume: bool = typer.Option(False, help=help_strings.SKIP_VOLUME), - persistent_volume: Optional[list[str]] = typer.Option(None, help=help_strings.PERSISTENT_VOLUME), - ephemeral_volume: Optional[list[str]] = typer.Option(None, help=help_strings.EPHEMERAL_VOLUME), - immutable_volume: Optional[list[str]] = typer.Option( - None, - help=help_strings.IMMUTABLE_VOLUME, - ), - crn_auto_tac: bool = typer.Option(False, help=help_strings.CRN_AUTO_TAC), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - print_message: bool = typer.Option(False), - verbose: bool = typer.Option(True), - debug: bool = False, + payment_type: Annotated[ + Optional[str], + typer.Option( + help=help_strings.PAYMENT_TYPE, + callback=lambda pt: None if pt is None else pt.lower(), + metavar=metavar_valid_payment_types, + case_sensitive=False, + ), + ] = None, + payment_chain: Annotated[ + Optional[Chain], + typer.Option( + help=help_strings.PAYMENT_CHAIN, + metavar=metavar_valid_chains, + case_sensitive=False, + ), + ] = None, + hypervisor: Annotated[HypervisorType, typer.Option(help=help_strings.HYPERVISOR)] = HypervisorType.qemu, + name: Annotated[Optional[str], typer.Option(help=help_strings.INSTANCE_NAME)] = None, + rootfs: Annotated[Optional[str], typer.Option(help=help_strings.ROOTFS)] = None, + 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)] = 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" + ).expanduser(), + address: Annotated[Optional[str], typer.Option(help=help_strings.ADDRESS_PAYER)] = None, + crn_hash: Annotated[Optional[str], typer.Option(help=help_strings.CRN_HASH)] = None, + crn_url: Annotated[Optional[str], typer.Option(help=help_strings.CRN_URL)] = None, + confidential: Annotated[bool, typer.Option(help=help_strings.CONFIDENTIAL_OPTION)] = False, + confidential_firmware: Annotated[ + str, typer.Option(help=help_strings.CONFIDENTIAL_FIRMWARE) + ] = settings.DEFAULT_CONFIDENTIAL_FIRMWARE, + gpu: Annotated[bool, typer.Option(help=help_strings.GPU_OPTION)] = False, + premium: Annotated[Optional[bool], typer.Option(help=help_strings.GPU_PREMIUM_OPTION)] = None, + skip_volume: Annotated[bool, typer.Option(help=help_strings.SKIP_VOLUME)] = False, + persistent_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.PERSISTENT_VOLUME)] = None, + ephemeral_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.EPHEMERAL_VOLUME)] = None, + immutable_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.IMMUTABLE_VOLUME)] = None, + crn_auto_tac: Annotated[bool, typer.Option(help=help_strings.CRN_AUTO_TAC)] = False, + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_message: Annotated[bool, typer.Option(help="Print the message after creation")] = False, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ) -> tuple[ItemHash, Optional[str], Chain]: """Create and register a new instance on aleph.im""" setup_logging(debug) @@ -688,14 +687,18 @@ async def create( @app.command() async def delete( - item_hash: str = typer.Argument(..., help="Instance item hash to forget"), - reason: str = typer.Option("User deletion", help="Reason for deleting the instance"), - chain: Optional[Chain] = typer.Option(None, help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains), - domain: Optional[str] = typer.Option(None, help=help_strings.CRN_URL_VM_DELETION), - private_key: Optional[str] = settings.PRIVATE_KEY_STRING, - private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE, - print_message: bool = typer.Option(False), - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Instance item hash to forget")], + reason: Annotated[str, typer.Option(help="Reason for deleting the instance")] = "User deletion", + chain: Annotated[ + Optional[Chain], typer.Option(help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains) + ] = None, + domain: Annotated[Optional[str], typer.Option(help=help_strings.CRN_URL_VM_DELETION)] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_message: Annotated[bool, typer.Option(help="Print the message after deletion")] = False, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """ Delete an instance, unallocating all resources associated with it. Associated VM will be stopped and erased. @@ -950,12 +953,16 @@ async def _show_instances(messages: builtins.list[InstanceMessage]): @app.command(name="list") async def list_instances( - address: Optional[str] = typer.Option(None, help="Owner address of the instances"), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - chain: Optional[Chain] = typer.Option(None, help=help_strings.ADDRESS_CHAIN, metavar=metavar_valid_chains), - json: bool = typer.Option(default=False, help="Print as json instead of rich table"), - debug: bool = False, + address: Annotated[Optional[str], typer.Option(help="Owner address of the instances")] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + chain: Annotated[ + Optional[Chain], typer.Option(help=help_strings.ADDRESS_CHAIN, metavar=metavar_valid_chains) + ] = None, + json: Annotated[bool, typer.Option(help="Print as json instead of rich table")] = False, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """List all instances associated to an account""" @@ -987,12 +994,16 @@ async def list_instances( @app.command() async def reboot( - vm_id: str = typer.Argument(..., help="VM item hash to reboot"), - domain: Optional[str] = typer.Option(None, help="CRN domain on which the VM is running"), - chain: Optional[Chain] = typer.Option(None, help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + vm_id: Annotated[str, typer.Argument(help="VM item hash to reboot")], + domain: Annotated[Optional[str], typer.Option(help="CRN domain on which the VM is running")] = None, + chain: Annotated[ + Optional[Chain], typer.Option(help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains) + ] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Reboot an instance""" @@ -1016,12 +1027,16 @@ async def reboot( @app.command() async def allocate( - vm_id: str = typer.Argument(..., help="VM item hash to allocate"), - domain: Optional[str] = typer.Option(None, help="CRN domain on which the VM will be allocated"), - chain: Optional[Chain] = typer.Option(None, help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + vm_id: Annotated[str, typer.Argument(help="VM item hash to allocate")], + domain: Annotated[Optional[str], typer.Option(help="CRN domain on which the VM will be allocated")] = None, + chain: Annotated[ + Optional[Chain], typer.Option(help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains) + ] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Notify a CRN to start an instance (for Pay-As-You-Go and confidential instances only)""" @@ -1045,12 +1060,16 @@ async def allocate( @app.command() async def logs( - vm_id: str = typer.Argument(..., help="VM item hash to retrieve the logs from"), - domain: Optional[str] = typer.Option(None, help="CRN domain on which the VM is running"), - chain: Optional[Chain] = typer.Option(None, help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + vm_id: Annotated[str, typer.Argument(help="VM item hash to retrieve the logs from")], + domain: Annotated[Optional[str], typer.Option(help="CRN domain on which the VM is running")] = None, + chain: Annotated[ + Optional[Chain], typer.Option(help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains) + ] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Retrieve the logs of an instance""" setup_logging(debug) @@ -1073,12 +1092,14 @@ async def logs( @app.command() async def stop( - vm_id: str = typer.Argument(..., help="VM item hash to stop"), - domain: Optional[str] = typer.Option(None, help="CRN domain on which the VM is running"), - chain: Optional[Chain] = typer.Option(None, help=help_strings.PAYMENT_CHAIN_USED), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + vm_id: Annotated[str, typer.Argument(help="VM item hash to stop")], + domain: Annotated[Optional[str], typer.Option(help="CRN domain on which the VM is running")] = None, + chain: Annotated[Optional[Chain], typer.Option(help=help_strings.PAYMENT_CHAIN_USED)] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Stop an instance""" @@ -1098,14 +1119,18 @@ async def stop( @app.command() async def confidential_init_session( - vm_id: str = typer.Argument(..., help="VM item hash to initialize the session for"), - domain: Optional[str] = typer.Option(None, help="CRN domain on which the session will be initialized"), - chain: Optional[Chain] = typer.Option(None, help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains), - policy: int = typer.Option(default=0x1), - keep_session: bool = typer.Option(None, help=help_strings.KEEP_SESSION), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + vm_id: Annotated[str, typer.Argument(help="VM item hash to initialize the session for")], + domain: Annotated[Optional[str], typer.Option(help="CRN domain on which the session will be initialized")] = None, + chain: Annotated[ + Optional[Chain], typer.Option(help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains) + ] = None, + policy: Annotated[int, typer.Option(help="Policy for the confidential session")] = 0x1, + keep_session: Annotated[Optional[bool], typer.Option(help=help_strings.KEEP_SESSION)] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Initialize a confidential communication session with the VM""" @@ -1173,18 +1198,22 @@ async def confidential_init_session( @app.command() async def confidential_start( - vm_id: str = typer.Argument(..., help="VM item hash to start"), - domain: Optional[str] = typer.Option(None, help="CRN domain on which the VM will be started"), - chain: Optional[Chain] = typer.Option(None, help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains), - firmware_hash: str = typer.Option( - settings.DEFAULT_CONFIDENTIAL_FIRMWARE_HASH, help=help_strings.CONFIDENTIAL_FIRMWARE_HASH - ), - firmware_file: str = typer.Option(None, help=help_strings.CONFIDENTIAL_FIRMWARE_PATH), - vm_secret: str = typer.Option(None, help=help_strings.VM_SECRET), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - verbose: bool = typer.Option(True), - debug: bool = False, + vm_id: Annotated[str, typer.Argument(help="VM item hash to start")], + domain: Annotated[Optional[str], typer.Option(help="CRN domain on which the VM will be started")] = None, + chain: Annotated[ + Optional[Chain], typer.Option(help=help_strings.PAYMENT_CHAIN_USED, metavar=metavar_valid_chains) + ] = None, + firmware_hash: Annotated[ + str, typer.Option(help=help_strings.CONFIDENTIAL_FIRMWARE_HASH) + ] = settings.DEFAULT_CONFIDENTIAL_FIRMWARE_HASH, + firmware_file: Annotated[Optional[str], typer.Option(help=help_strings.CONFIDENTIAL_FIRMWARE_PATH)] = None, + vm_secret: Annotated[Optional[str], typer.Option(help=help_strings.VM_SECRET)] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Validate the authenticity of the VM and start it""" @@ -1270,61 +1299,60 @@ async def confidential_start( @app.command(name="confidential") async def confidential_create( - vm_id: Optional[str] = typer.Argument(default=None, help=help_strings.VM_ID), - crn_url: Optional[str] = typer.Option(default=None, help=help_strings.CRN_URL), - crn_hash: Optional[str] = typer.Option(default=None, help=help_strings.CRN_HASH), - policy: int = typer.Option(default=0x1), - confidential_firmware: str = typer.Option( - default=settings.DEFAULT_CONFIDENTIAL_FIRMWARE, help=help_strings.CONFIDENTIAL_FIRMWARE - ), - firmware_hash: str = typer.Option( - settings.DEFAULT_CONFIDENTIAL_FIRMWARE_HASH, help=help_strings.CONFIDENTIAL_FIRMWARE_HASH - ), - firmware_file: Optional[str] = typer.Option(None, help=help_strings.CONFIDENTIAL_FIRMWARE_PATH), - keep_session: Optional[bool] = typer.Option(None, help=help_strings.KEEP_SESSION), - vm_secret: Optional[str] = typer.Option(None, help=help_strings.VM_SECRET), - payment_type: Optional[str] = typer.Option( - None, - help=help_strings.PAYMENT_TYPE, - callback=lambda pt: None if pt is None else pt.lower(), - metavar=metavar_valid_payment_types, - case_sensitive=False, - ), - payment_chain: Optional[Chain] = typer.Option( - None, - help=help_strings.PAYMENT_CHAIN, - metavar=metavar_valid_chains, - case_sensitive=False, - ), - name: Optional[str] = typer.Option(None, help=help_strings.INSTANCE_NAME), - rootfs: Optional[str] = typer.Option(None, help=help_strings.ROOTFS), - compute_units: Optional[int] = typer.Option(None, help=help_strings.COMPUTE_UNITS), - vcpus: Optional[int] = typer.Option(None, help=help_strings.VCPUS), - memory: Optional[int] = typer.Option(None, help=help_strings.MEMORY), - rootfs_size: Optional[int] = typer.Option(None, help=help_strings.ROOTFS_SIZE), - timeout_seconds: float = typer.Option( - settings.DEFAULT_VM_TIMEOUT, - help=help_strings.TIMEOUT_SECONDS, - ), - ssh_pubkey_file: Path = typer.Option( - Path("~/.ssh/id_rsa.pub").expanduser(), - help=help_strings.SSH_PUBKEY_FILE, - ), - address: Optional[str] = typer.Option(None, help=help_strings.ADDRESS_PAYER), - gpu: bool = typer.Option(False, help=help_strings.GPU_OPTION), - premium: Optional[bool] = typer.Option(None, help=help_strings.GPU_PREMIUM_OPTION), - skip_volume: bool = typer.Option(False, help=help_strings.SKIP_VOLUME), - persistent_volume: Optional[list[str]] = typer.Option(None, help=help_strings.PERSISTENT_VOLUME), - ephemeral_volume: Optional[list[str]] = typer.Option(None, help=help_strings.EPHEMERAL_VOLUME), - immutable_volume: Optional[list[str]] = typer.Option( - None, - help=help_strings.IMMUTABLE_VOLUME, - ), - crn_auto_tac: bool = typer.Option(False, help=help_strings.CRN_AUTO_TAC), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + vm_id: Annotated[Optional[str], typer.Argument(help=help_strings.VM_ID)] = None, + crn_url: Annotated[Optional[str], typer.Option(help=help_strings.CRN_URL)] = None, + crn_hash: Annotated[Optional[str], typer.Option(help=help_strings.CRN_HASH)] = None, + policy: Annotated[int, typer.Option(help="Policy for the confidential session")] = 0x1, + confidential_firmware: Annotated[ + str, typer.Option(help=help_strings.CONFIDENTIAL_FIRMWARE) + ] = settings.DEFAULT_CONFIDENTIAL_FIRMWARE, + firmware_hash: Annotated[ + str, typer.Option(help=help_strings.CONFIDENTIAL_FIRMWARE_HASH) + ] = settings.DEFAULT_CONFIDENTIAL_FIRMWARE_HASH, + firmware_file: Annotated[Optional[str], typer.Option(help=help_strings.CONFIDENTIAL_FIRMWARE_PATH)] = None, + keep_session: Annotated[Optional[bool], typer.Option(help=help_strings.KEEP_SESSION)] = None, + vm_secret: Annotated[Optional[str], typer.Option(help=help_strings.VM_SECRET)] = None, + payment_type: Annotated[ + Optional[str], + typer.Option( + help=help_strings.PAYMENT_TYPE, + callback=lambda pt: None if pt is None else pt.lower(), + metavar=metavar_valid_payment_types, + case_sensitive=False, + ), + ] = None, + payment_chain: Annotated[ + Optional[Chain], + typer.Option( + help=help_strings.PAYMENT_CHAIN, + metavar=metavar_valid_chains, + case_sensitive=False, + ), + ] = None, + name: Annotated[Optional[str], typer.Option(help=help_strings.INSTANCE_NAME)] = None, + rootfs: Annotated[Optional[str], typer.Option(help=help_strings.ROOTFS)] = None, + 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)] = 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" + ).expanduser(), + address: Annotated[Optional[str], typer.Option(help=help_strings.ADDRESS_PAYER)] = None, + gpu: Annotated[bool, typer.Option(help=help_strings.GPU_OPTION)] = False, + premium: Annotated[Optional[bool], typer.Option(help=help_strings.GPU_PREMIUM_OPTION)] = None, + skip_volume: Annotated[bool, typer.Option(help=help_strings.SKIP_VOLUME)] = False, + persistent_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.PERSISTENT_VOLUME)] = None, + ephemeral_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.EPHEMERAL_VOLUME)] = None, + immutable_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.IMMUTABLE_VOLUME)] = None, + crn_auto_tac: Annotated[bool, typer.Option(help=help_strings.CRN_AUTO_TAC)] = False, + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Create (optional), start and unlock a confidential VM (all-in-one command) @@ -1439,44 +1467,41 @@ async def confidential_create( @app.command(name="gpu") async def gpu_create( - payment_chain: Optional[Chain] = typer.Option( - None, - help=help_strings.PAYMENT_CHAIN, - metavar=metavar_valid_payg_chains, - case_sensitive=False, - ), - name: Optional[str] = typer.Option(None, help=help_strings.INSTANCE_NAME), - rootfs: Optional[str] = typer.Option(None, help=help_strings.ROOTFS), - compute_units: Optional[int] = typer.Option(None, help=help_strings.COMPUTE_UNITS), - vcpus: Optional[int] = typer.Option(None, help=help_strings.VCPUS), - memory: Optional[int] = typer.Option(None, help=help_strings.MEMORY), - rootfs_size: Optional[int] = typer.Option(None, help=help_strings.ROOTFS_SIZE), - premium: Optional[bool] = typer.Option(None, help=help_strings.GPU_PREMIUM_OPTION), - timeout_seconds: float = typer.Option( - settings.DEFAULT_VM_TIMEOUT, - help=help_strings.TIMEOUT_SECONDS, - ), - ssh_pubkey_file: Path = typer.Option( - Path("~/.ssh/id_rsa.pub").expanduser(), - help=help_strings.SSH_PUBKEY_FILE, - ), - address: Optional[str] = typer.Option(None, help=help_strings.ADDRESS_PAYER), - crn_hash: Optional[str] = typer.Option(None, help=help_strings.CRN_HASH), - crn_url: Optional[str] = typer.Option(None, help=help_strings.CRN_URL), - skip_volume: bool = typer.Option(False, help=help_strings.SKIP_VOLUME), - persistent_volume: Optional[list[str]] = typer.Option(None, help=help_strings.PERSISTENT_VOLUME), - ephemeral_volume: Optional[list[str]] = typer.Option(None, help=help_strings.EPHEMERAL_VOLUME), - immutable_volume: Optional[list[str]] = typer.Option( - None, - help=help_strings.IMMUTABLE_VOLUME, - ), - crn_auto_tac: bool = typer.Option(False, help=help_strings.CRN_AUTO_TAC), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - print_message: bool = typer.Option(False), - verbose: bool = typer.Option(True), - debug: bool = False, + payment_chain: Annotated[ + Optional[Chain], + typer.Option( + help=help_strings.PAYMENT_CHAIN, + metavar=metavar_valid_payg_chains, + case_sensitive=False, + ), + ] = None, + name: Annotated[Optional[str], typer.Option(help=help_strings.INSTANCE_NAME)] = None, + rootfs: Annotated[Optional[str], typer.Option(help=help_strings.ROOTFS)] = None, + 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)] = 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( + "~/.ssh/id_rsa.pub" + ).expanduser(), + address: Annotated[Optional[str], typer.Option(help=help_strings.ADDRESS_PAYER)] = None, + crn_hash: Annotated[Optional[str], typer.Option(help=help_strings.CRN_HASH)] = None, + crn_url: Annotated[Optional[str], typer.Option(help=help_strings.CRN_URL)] = None, + skip_volume: Annotated[bool, typer.Option(help=help_strings.SKIP_VOLUME)] = False, + persistent_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.PERSISTENT_VOLUME)] = None, + ephemeral_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.EPHEMERAL_VOLUME)] = None, + immutable_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.IMMUTABLE_VOLUME)] = None, + crn_auto_tac: Annotated[bool, typer.Option(help=help_strings.CRN_AUTO_TAC)] = False, + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_message: Annotated[bool, typer.Option(help="Print the message after creation")] = False, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Create and register a new GPU instance on aleph.im""" diff --git a/src/aleph_client/commands/message.py b/src/aleph_client/commands/message.py index f68e7b66..7b62a20e 100644 --- a/src/aleph_client/commands/message.py +++ b/src/aleph_client/commands/message.py @@ -7,7 +7,7 @@ import tempfile import time from pathlib import Path -from typing import Optional +from typing import Annotated, Optional import typer from aleph.sdk import AlephHttpClient, AuthenticatedAlephHttpClient @@ -39,7 +39,7 @@ @app.command() async def get( - item_hash: str = typer.Argument(..., help="Item hash of the message"), + item_hash: Annotated[str, typer.Argument(help="Item hash of the message")], ): async with AlephHttpClient(api_server=settings.API_HOST) as client: message: Optional[AlephMessage] = None @@ -60,20 +60,20 @@ async def get( @app.command() async def find( - pagination: int = 200, - page: int = 1, - message_types: Optional[str] = None, - content_types: Optional[str] = None, - content_keys: Optional[str] = None, - refs: Optional[str] = None, - addresses: Optional[str] = None, - tags: Optional[str] = None, - hashes: Optional[str] = None, - channels: Optional[str] = None, - chains: Optional[str] = None, - start_date: Optional[str] = None, - end_date: Optional[str] = None, - ignore_invalid_messages: bool = True, + pagination: Annotated[int, typer.Option()] = 200, + page: Annotated[int, typer.Option()] = 1, + message_types: Annotated[Optional[str], typer.Option()] = None, + content_types: Annotated[Optional[str], typer.Option()] = None, + content_keys: Annotated[Optional[str], typer.Option()] = None, + refs: Annotated[Optional[str], typer.Option()] = None, + addresses: Annotated[Optional[str], typer.Option()] = None, + tags: Annotated[Optional[str], typer.Option()] = None, + hashes: Annotated[Optional[str], typer.Option()] = None, + channels: Annotated[Optional[str], typer.Option()] = None, + chains: Annotated[Optional[str], typer.Option()] = None, + start_date: Annotated[Optional[str], typer.Option()] = None, + end_date: Annotated[Optional[str], typer.Option()] = None, + ignore_invalid_messages: Annotated[bool, typer.Option()] = True, ): parsed_message_types = ( [MessageType(message_type) for message_type in message_types.split(",")] if message_types else None @@ -114,16 +114,18 @@ async def find( @app.command() async def post( - path: Optional[Path] = typer.Option( - None, - help="Path to the content you want to post. If omitted, you can input your content directly", - ), - type: str = typer.Option("test", help="Text representing the message object type"), - ref: Optional[str] = typer.Option(None, help=help_strings.REF), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + path: Annotated[ + Optional[Path], + typer.Option(help="Path to the content you want to post. If omitted, you can input your content directly"), + ] = None, + type: Annotated[str, typer.Option(help="Text representing the message object type")] = "test", + ref: Annotated[Optional[str], typer.Option(help=help_strings.REF)] = None, + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option()] = False, ): """Post a message on aleph.im.""" @@ -168,10 +170,12 @@ async def post( @app.command() async def amend( - item_hash: str = typer.Argument(..., help="Hash reference of the message to amend"), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Hash reference of the message to amend")], + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option()] = False, ): """Amend an existing aleph.im message.""" @@ -225,12 +229,16 @@ async def amend( @app.command() async def forget( - hashes: str = typer.Argument(..., help="Comma separated list of hash references of messages to forget"), - reason: Optional[str] = typer.Option(None, help="A description of why the messages are being forgotten."), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + hashes: Annotated[str, typer.Argument(help="Comma separated list of hash references of messages to forget")], + reason: Annotated[ + Optional[str], typer.Option(help="A description of why the messages are being forgotten.") + ] = None, + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option()] = False, ): """Forget an existing aleph.im message.""" @@ -245,9 +253,9 @@ async def forget( @app.command() async def watch( - ref: str = typer.Argument(..., help="Hash reference of the message to watch"), - indent: Optional[int] = typer.Option(None, help="Number of indents to use"), - debug: bool = False, + ref: Annotated[str, typer.Argument(help="Hash reference of the message to watch")], + indent: Annotated[Optional[int], typer.Option(help="Number of indents to use")] = None, + debug: Annotated[bool, typer.Option()] = False, ): """Watch a hash for amends and print amend hashes""" @@ -270,10 +278,12 @@ async def watch( @app.command() def sign( - message: Optional[str] = typer.Option(None, help=help_strings.SIGNABLE_MESSAGE), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - debug: bool = False, + message: Annotated[Optional[str], typer.Option(help=help_strings.SIGNABLE_MESSAGE)] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + debug: Annotated[bool, typer.Option()] = False, ): """Sign an aleph message with a private key. If no --message is provided, the message will be read from stdin.""" diff --git a/src/aleph_client/commands/node.py b/src/aleph_client/commands/node.py index d577d2fd..f71b966e 100644 --- a/src/aleph_client/commands/node.py +++ b/src/aleph_client/commands/node.py @@ -5,7 +5,7 @@ import logging import re import unicodedata -from typing import Optional +from typing import Annotated, Optional import aiohttp import typer @@ -194,16 +194,16 @@ def _show_core(node_info): @app.command() async def compute( - json: bool = typer.Option(default=False, help="Print as json instead of rich table"), - active: bool = typer.Option(default=False, help="Only show active nodes"), - address: Optional[str] = typer.Option(default=None, help="Owner address to filter by"), - payg_receiver: Optional[str] = typer.Option( - default=None, help="PAYG (Pay-As-You-Go) receiver address to filter by" - ), - crn_url: Optional[str] = typer.Option(default=None, help="CRN Url to filter by"), - crn_hash: Optional[str] = typer.Option(default=None, help="CRN hash to filter by"), - ccn_hash: Optional[str] = typer.Option(default=None, help="Linked CCN hash to filter by"), - debug: bool = False, + json: Annotated[bool, typer.Option(help="Print as json instead of rich table")] = False, + active: Annotated[bool, typer.Option(help="Only show active nodes")] = False, + address: Annotated[Optional[str], typer.Option(help="Owner address to filter by")] = None, + payg_receiver: Annotated[ + Optional[str], typer.Option(help="PAYG (Pay-As-You-Go) receiver address to filter by") + ] = None, + crn_url: Annotated[Optional[str], typer.Option(help="CRN URL to filter by")] = None, + crn_hash: Annotated[Optional[str], typer.Option(help="CRN hash to filter by")] = None, + ccn_hash: Annotated[Optional[str], typer.Option(help="CCN hash to filter by")] = None, + debug: Annotated[bool, typer.Option()] = False, ): """Get all compute node (CRN) on aleph network""" @@ -228,11 +228,11 @@ async def compute( @app.command() async def core( - json: bool = typer.Option(default=False, help="Print as json instead of rich table"), - active: bool = typer.Option(default=False, help="Only show active nodes"), - address: Optional[str] = typer.Option(default=None, help="Owner address to filter by"), - ccn_hash: Optional[str] = typer.Option(default=None, help="CCN hash to filter by"), - debug: bool = False, + json: Annotated[bool, typer.Option(help="Print as json instead of rich table")] = False, + active: Annotated[bool, typer.Option(help="Only show active nodes")] = False, + address: Annotated[Optional[str], typer.Option(help="Owner address to filter by")] = None, + ccn_hash: Annotated[Optional[str], typer.Option(help="CCN hash to filter by")] = None, + debug: Annotated[bool, typer.Option()] = False, ): """Get all core node (CCN) on aleph""" setup_logging(debug) diff --git a/src/aleph_client/commands/program.py b/src/aleph_client/commands/program.py index 3942931e..3974048a 100644 --- a/src/aleph_client/commands/program.py +++ b/src/aleph_client/commands/program.py @@ -7,7 +7,7 @@ from collections.abc import Mapping from decimal import Decimal from pathlib import Path -from typing import Any, Optional, cast +from typing import Annotated, Any, Optional, cast from zipfile import BadZipFile import aiohttp @@ -59,51 +59,37 @@ @app.command(name="upload") @app.command(name="create") async def upload( - path: Path = typer.Argument(..., help=help_strings.PROGRAM_PATH), - entrypoint: str = typer.Argument( - ..., - help=help_strings.PROGRAM_ENTRYPOINT, - ), - name: Optional[str] = typer.Option(None, help="Name for your program"), - runtime: str = typer.Option( - None, - help=help_strings.PROGRAM_RUNTIME.format(runtime_id=settings.DEFAULT_RUNTIME_ID), - ), - compute_units: Optional[int] = typer.Option(None, help=help_strings.COMPUTE_UNITS), - vcpus: Optional[int] = typer.Option(None, help=help_strings.VCPUS), - memory: Optional[int] = typer.Option(None, help=help_strings.MEMORY), - timeout_seconds: float = typer.Option( - settings.DEFAULT_VM_TIMEOUT, - help=help_strings.TIMEOUT_SECONDS, - ), - internet: bool = typer.Option( - False, - help=help_strings.PROGRAM_INTERNET, - ), - updatable: bool = typer.Option(False, help=help_strings.PROGRAM_UPDATABLE), - beta: bool = typer.Option( - False, - help=help_strings.PROGRAM_BETA, - ), - persistent: bool = typer.Option(False, help=help_strings.PROGRAM_PERSISTENT), - skip_volume: bool = typer.Option(False, help=help_strings.SKIP_VOLUME), - persistent_volume: Optional[list[str]] = typer.Option(None, help=help_strings.PERSISTENT_VOLUME), - ephemeral_volume: Optional[list[str]] = typer.Option(None, help=help_strings.EPHEMERAL_VOLUME), - immutable_volume: Optional[list[str]] = typer.Option( - None, - help=help_strings.IMMUTABLE_VOLUME, - ), - skip_env_var: bool = typer.Option(False, help=help_strings.SKIP_ENV_VAR), - env_vars: Optional[str] = typer.Option(None, help=help_strings.ENVIRONMENT_VARIABLES), - address: Optional[str] = typer.Option(None, help=help_strings.ADDRESS_PAYER), - channel: Optional[str] = typer.Option(default=settings.DEFAULT_CHANNEL, help=help_strings.CHANNEL), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - print_messages: bool = typer.Option(False), - print_code_message: bool = typer.Option(False), - print_program_message: bool = typer.Option(False), - verbose: bool = True, - debug: bool = False, + path: Annotated[Path, typer.Argument(help=help_strings.PROGRAM_PATH)], + entrypoint: Annotated[str, typer.Argument(help=help_strings.PROGRAM_ENTRYPOINT)], + name: Annotated[Optional[str], typer.Option(help="Name for your program")] = None, + runtime: Annotated[ + Optional[str], typer.Option(help=help_strings.PROGRAM_RUNTIME.format(runtime_id=settings.DEFAULT_RUNTIME_ID)) + ] = None, + 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, + timeout_seconds: Annotated[float, typer.Option(help=help_strings.TIMEOUT_SECONDS)] = settings.DEFAULT_VM_TIMEOUT, + internet: Annotated[bool, typer.Option(help=help_strings.PROGRAM_INTERNET)] = False, + updatable: Annotated[bool, typer.Option(help=help_strings.PROGRAM_UPDATABLE)] = False, + beta: Annotated[bool, typer.Option(help=help_strings.PROGRAM_BETA)] = False, + persistent: Annotated[bool, typer.Option(help=help_strings.PROGRAM_PERSISTENT)] = False, + skip_volume: Annotated[bool, typer.Option(help=help_strings.SKIP_VOLUME)] = False, + persistent_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.PERSISTENT_VOLUME)] = None, + ephemeral_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.EPHEMERAL_VOLUME)] = None, + immutable_volume: Annotated[Optional[list[str]], typer.Option(help=help_strings.IMMUTABLE_VOLUME)] = None, + skip_env_var: Annotated[bool, typer.Option(help=help_strings.SKIP_ENV_VAR)] = False, + env_vars: Annotated[Optional[str], typer.Option(help=help_strings.ENVIRONMENT_VARIABLES)] = None, + address: Annotated[Optional[str], typer.Option(help=help_strings.ADDRESS_PAYER)] = None, + channel: Annotated[Optional[str], typer.Option(help=help_strings.CHANNEL)] = settings.DEFAULT_CHANNEL, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_messages: Annotated[bool, typer.Option(help="Print the messages after creation")] = False, + print_code_message: Annotated[bool, typer.Option(help="Print the code message after creation")] = False, + print_program_message: Annotated[bool, typer.Option(help="Print the program message after creation")] = False, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ) -> Optional[str]: """Register a program to run on aleph.im (create/upload are aliases) @@ -283,13 +269,15 @@ async def upload( @app.command() async def update( - item_hash: str = typer.Argument(..., help="Item hash to update"), - path: Path = typer.Argument(..., help=help_strings.PROGRAM_PATH), - private_key: Optional[str] = settings.PRIVATE_KEY_STRING, - private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE, - print_message: bool = typer.Option(False), - verbose: bool = True, - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Item hash to update")], + path: Annotated[Path, typer.Argument(help=help_strings.PROGRAM_PATH)], + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_message: Annotated[bool, typer.Option(help="Print the message after creation")] = False, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Update the code of an existing program (item hash will not change)""" @@ -390,14 +378,16 @@ async def update( @app.command() async def delete( - item_hash: str = typer.Argument(..., help="Item hash to unpersist"), - reason: str = typer.Option("User deletion", help="Reason for deleting the program"), - keep_code: bool = typer.Option(False, help=help_strings.PROGRAM_KEEP_CODE), - private_key: Optional[str] = settings.PRIVATE_KEY_STRING, - private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE, - print_message: bool = typer.Option(False), - verbose: bool = True, - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Item hash to unpersist")], + reason: Annotated[str, typer.Option(help="Reason for deleting the program")] = "User deletion", + keep_code: Annotated[bool, typer.Option(help=help_strings.PROGRAM_KEEP_CODE)] = False, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_message: Annotated[bool, typer.Option(help="Print the message after deletion")] = False, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Delete a program""" @@ -449,11 +439,13 @@ async def delete( @app.command(name="list") async def list_programs( - address: Optional[str] = typer.Option(None, help="Owner address of the programs"), - private_key: Optional[str] = typer.Option(settings.PRIVATE_KEY_STRING, help=help_strings.PRIVATE_KEY), - private_key_file: Optional[Path] = typer.Option(settings.PRIVATE_KEY_FILE, help=help_strings.PRIVATE_KEY_FILE), - json: bool = typer.Option(default=False, help="Print as json instead of rich table"), - debug: bool = False, + address: Annotated[Optional[str], typer.Option(help="Owner address of the programs")] = None, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + json: Annotated[bool, typer.Option(help="Print as json instead of rich table")] = False, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """List all programs associated to an account""" @@ -574,16 +566,15 @@ async def list_programs( @app.command() async def persist( - item_hash: str = typer.Argument(..., help="Item hash to persist"), - keep_prev: bool = typer.Option( - False, - help=help_strings.PROGRAM_KEEP_PREV, - ), - private_key: Optional[str] = settings.PRIVATE_KEY_STRING, - private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE, - print_message: bool = typer.Option(False), - verbose: bool = True, - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Item hash to persist")], + keep_prev: Annotated[bool, typer.Option(help=help_strings.PROGRAM_KEEP_PREV)] = False, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_message: Annotated[bool, typer.Option(help="Print the message after persisting")] = False, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ) -> Optional[str]: """ Recreate a non-persistent program as persistent (item hash will change). The program must be updatable and yours @@ -669,16 +660,15 @@ async def persist( @app.command() async def unpersist( - item_hash: str = typer.Argument(..., help="Item hash to unpersist"), - keep_prev: bool = typer.Option( - False, - help=help_strings.PROGRAM_KEEP_PREV, - ), - private_key: Optional[str] = settings.PRIVATE_KEY_STRING, - private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE, - print_message: bool = typer.Option(False), - verbose: bool = True, - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Item hash to unpersist")], + keep_prev: Annotated[bool, typer.Option(help=help_strings.PROGRAM_KEEP_PREV)] = False, + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + print_message: Annotated[bool, typer.Option(help="Print the message after unpersisting")] = False, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = True, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ) -> Optional[str]: """ Recreate a persistent program as non-persistent (item hash will change). The program must be updatable and yours @@ -764,12 +754,14 @@ async def unpersist( @app.command() async def logs( - item_hash: str = typer.Argument(..., help="Item hash of program"), - private_key: Optional[str] = settings.PRIVATE_KEY_STRING, - private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE, - domain: str = typer.Option(None, help=help_strings.PROMPT_PROGRAM_CRN_URL), - chain: Chain = typer.Option(None, help=help_strings.ADDRESS_CHAIN), - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Item hash of program")], + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + domain: Annotated[Optional[str], typer.Option(help=help_strings.PROMPT_PROGRAM_CRN_URL)] = None, + chain: Annotated[Optional[Chain], typer.Option(help=help_strings.ADDRESS_CHAIN)] = None, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Display the logs of a program @@ -778,9 +770,9 @@ async def logs( setup_logging(debug) account = _load_account(private_key, private_key_file, chain=chain) - domain = sanitize_url(domain or Prompt.ask(help_strings.PROMPT_PROGRAM_CRN_URL)) + domain_ = sanitize_url(domain or Prompt.ask(help_strings.PROMPT_PROGRAM_CRN_URL)) - async with VmClient(account, domain) as client: + async with VmClient(account, domain_) as client: async with client.operate(vm_id=item_hash, operation="logs", method="GET") as response: logger.debug("Request %s %s", response.url, response.status) if response.status != 200: @@ -804,11 +796,13 @@ async def logs( @app.command() async def runtime_checker( - item_hash: str = typer.Argument(..., help="Item hash of the runtime to check"), - private_key: Optional[str] = settings.PRIVATE_KEY_STRING, - private_key_file: Optional[Path] = settings.PRIVATE_KEY_FILE, - verbose: bool = False, - debug: bool = False, + item_hash: Annotated[str, typer.Argument(help="Item hash of the runtime to check")], + private_key: Annotated[Optional[str], typer.Option(help=help_strings.PRIVATE_KEY)] = settings.PRIVATE_KEY_STRING, + private_key_file: Annotated[ + Optional[Path], typer.Option(help=help_strings.PRIVATE_KEY_FILE) + ] = settings.PRIVATE_KEY_FILE, + verbose: Annotated[bool, typer.Option(help="Display additional information")] = False, + debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False, ): """Check versions used by a runtime (distribution, python, nodejs, etc)""" diff --git a/tests/unit/test_commands.py b/tests/unit/test_commands.py index 2a3e1581..458b5198 100644 --- a/tests/unit/test_commands.py +++ b/tests/unit/test_commands.py @@ -316,3 +316,16 @@ def test_file_download_only_info(): ) assert result.exit_code == 0 assert result.return_value.dict()["hash"] == FAKE_STORE_HASH_CONTENT_FILE_CID + + +def test_file_list(): + result = runner.invoke( + app, + [ + "file", + "list", + ], + ) + + assert result.exit_code == 0 + assert "0x" in result.stdout