Skip to content

Commit

Permalink
Wrap all cmd args in Annotated + minor fix on file list (#339)
Browse files Browse the repository at this point in the history
- Wrap all command args in Annotated to allow to easily import CLI as a lib
  • Loading branch information
philogicae authored Feb 18, 2025
1 parent fbae1c4 commit 41575ac
Show file tree
Hide file tree
Showing 8 changed files with 518 additions and 448 deletions.
56 changes: 32 additions & 24 deletions src/aleph_client/commands/account.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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."""

Expand Down Expand Up @@ -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).
Expand Down Expand Up @@ -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."""

Expand All @@ -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.
Expand All @@ -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."""

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)"""

Expand Down
50 changes: 29 additions & 21 deletions src/aleph_client/commands/domain.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand Down
96 changes: 54 additions & 42 deletions src/aleph_client/commands/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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."""

Expand All @@ -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."""

Expand Down Expand Up @@ -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."""

Expand Down Expand Up @@ -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."""

Expand Down Expand Up @@ -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)
Expand Down
Loading

0 comments on commit 41575ac

Please sign in to comment.