Feature: credits buy commands

pyproject.toml

@@ -30,18 +30,18 @@ dynamic = [ "version" ]
 dependencies = [
   "aiodns==3.2",
   "aiohttp==3.11.13",
-  "aleph-message>=1.0.1",
-  "aleph-sdk-python>=2.0.5",
-  "base58==2.1.1",               # Needed now as default with _load_account changement
+  "aleph-message>=1.0.4",
+  "aleph-sdk-python @ git+https://github.com/aleph-im/aleph-sdk-python@1yam-credits-system",
+  "base58==2.1.1",                                                                           # Needed now as default with _load_account changement
   "click<8.2",
-  "py-sr25519-bindings==0.2",    # Needed for DOT signatures
+  "py-sr25519-bindings==0.2",                                                                # Needed for DOT signatures
   "pydantic>=2",
   "pygments==2.19.1",
-  "pynacl==1.5",                 # Needed now as default with _load_account changement
+  "pynacl==1.5",                                                                             # Needed now as default with _load_account changement
   "python-magic==0.4.27",
   "rich==13.9.*",
   "setuptools>=65.5",
-  "substrate-interface==1.7.11", # Needed for DOT signatures
+  "substrate-interface==1.7.11",                                                             # Needed for DOT signatures
   "textual==0.73",
   "typer==0.15.2",
 ]

src/aleph_client/__main__.py

@@ -6,6 +6,7 @@
     about,
     account,
     aggregate,
+    credit,
     domain,
     files,
     instance,
@@ -30,9 +31,11 @@
 app.add_typer(files.app, name="file", help="Manage files (upload and pin on IPFS) on aleph.im & twentysix.cloud")
 app.add_typer(program.app, name="program", help="Manage programs (micro-VMs) on aleph.im & twentysix.cloud")
 app.add_typer(instance.app, name="instance", help="Manage instances (VMs) on aleph.im & twentysix.cloud")
+app.add_typer(credit.app, name="credits", help="Credits commmands on aleph.im")
 app.add_typer(domain.app, name="domain", help="Manage custom domain (DNS) on aleph.im & twentysix.cloud")
 app.add_typer(node.app, name="node", help="Get node info on aleph.im & twentysix.cloud")
 app.add_typer(about.app, name="about", help="Display the informations of Aleph CLI")
+
 app.command("pricing")(pricing.prices_for_service)
 
 if __name__ == "__main__":

src/aleph_client/commands/account.py

@@ -9,6 +9,7 @@
 
 import aiohttp
 import typer
+from aleph.sdk import AlephHttpClient
 from aleph.sdk.account import _load_account
 from aleph.sdk.chains.common import generate_key
 from aleph.sdk.chains.solana import parse_private_key as parse_solana_private_key
@@ -334,12 +335,26 @@ async def balance(
                 ),
             ]
 
+            try:
+                # Fetch user Credits
+                async with AlephHttpClient() as client:
+                    credits_balance = await client.get_credit_balance(address)
+                    infos += [
+                        Text("\nCredits:"),
+                        Text.from_markup(
+                            f"[bright_cyan] {displayable_amount(credits_balance.credits, decimals=2)}[/bright_cyan]"
+                        ),
+                    ]
+            except Exception as e:
+                # In the case we call on ccn that does not support credits yet
+                logger.warning(f"Failed to fetch credits balance: {e}")
+
             # Get vouchers and add them to Account Info panel
             vouchers = await voucher_manager.get_all(address=address)
             if vouchers:
                 voucher_names = [voucher.name for voucher in vouchers]
                 infos += [
-                    Text("\n\nVouchers:"),
+                    Text("\nVouchers:"),
                     Text.from_markup(f"\n [bright_cyan]{', '.join(voucher_names)}[/bright_cyan]"),
                 ]
 

src/aleph_client/commands/credit.py

@@ -0,0 +1,136 @@
+import logging
+from pathlib import Path
+from typing import Annotated, Optional
+
+import typer
+from aiohttp import ClientResponseError
+from aleph.sdk import AlephHttpClient
+from aleph.sdk.account import _load_account
+from aleph.sdk.conf import settings
+from aleph.sdk.query.filters import CreditsFilter
+from aleph.sdk.types import AccountFromPrivateKey
+from aleph.sdk.utils import displayable_amount
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+from rich.text import Text
+
+from aleph_client.commands import help_strings
+from aleph_client.commands.utils import setup_logging
+from aleph_client.utils import AsyncTyper
+
+logger = logging.getLogger(__name__)
+app = AsyncTyper(no_args_is_help=True)
+console = Console()
+
+
+@app.command()
+async def show(
+    address: Annotated[
+        str,
+        typer.Argument(help="Address of the wallet you want to check / None if you want check your current accounts"),
+    ] = "",
+    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="Display as json")] = False,
+    debug: Annotated[bool, typer.Option()] = False,
+):
+    """Display the numbers of credits for a specific address."""
+
+    setup_logging(debug)
+
+    account: AccountFromPrivateKey = _load_account(private_key, private_key_file)
+
+    if account and not address:
+        address = account.get_address()
+
+    if address:
+        async with AlephHttpClient(api_server=settings.API_HOST) as client:
+            credit = await client.get_credit_balance(address=address)
+            if json:
+                typer.echo(credit.model_dump_json(indent=4))
+            else:
+                infos = [
+                    Text.from_markup(f"Address: [bright_cyan]{address}[/bright_cyan]\n"),
+                    Text("Credits:"),
+                    Text.from_markup(f"[bright_cyan] {displayable_amount(credit.credits, decimals=2)}[/bright_cyan]"),
+                ]
+                console.print(
+                    Panel(
+                        Text.assemble(*infos),
+                        title="Credits Infos",
+                        border_style="bright_cyan",
+                        expand=False,
+                        title_align="left",
+                    )
+                )
+    else:
+        typer.echo("Error: Please provide either a private key, private key file, or an address.")
+
+
+@app.command(name="list")
+async def list_credits(
+    page_size: Annotated[int, typer.Option(help="Numbers of element per page")] = 100,
+    page: Annotated[int, typer.Option(help="Current Page")] = 1,
+    min_balance: Annotated[
+        Optional[int], typer.Option(help="Minimum balance required to be taken into account")
+    ] = None,
+    json: Annotated[bool, typer.Option(help="Display as json")] = False,
+):
+    try:
+        async with AlephHttpClient(api_server=settings.API_HOST) as client:
+            credit_filter = CreditsFilter(min_balance=min_balance) if min_balance else None
+            filtered_credits = await client.get_credits(credit_filter=credit_filter, page_size=page_size, page=page)
+            if json:
+                typer.echo(filtered_credits.model_dump_json(indent=4))
+            else:
+                table = Table(title="Credits Information", border_style="white")
+                table.add_column("Address", style="bright_cyan")
+                table.add_column("Credits", justify="right", style="bright_cyan")
+
+                for credit in filtered_credits.credit_balances:
+                    table.add_row(credit.address, f"{displayable_amount(credit.credits, decimals=2)}")
+
+                # Add pagination footer
+                pagination_info = Text.assemble(
+                    f"Page: {filtered_credits.pagination_page} of {filtered_credits.pagination_total} | ",
+                    f"Items per page: {filtered_credits.pagination_per_page} | ",
+                    f"Page size: {filtered_credits.pagination_total}",
+                )
+                table.caption = pagination_info
+
+                console.print(table)
+    except ClientResponseError as e:
+        typer.echo("Failed to retrieve credits.")
+        raise (e)
+
+
+@app.command(name="buy")
+async def buy_credits(
+    debug: Annotated[bool, typer.Option()] = False,
+):
+    """Purchase Aleph credits through the payment website."""
+    setup_logging(debug)
+
+    # Payment URL
+    payment_url = "https://app.aleph.cloud/console/"
+
+    infos = [
+        Text.from_markup("To purchase Aleph credits, visit:"),
+        Text.from_markup(
+            f"\n\n[bright_yellow][link={payment_url}]{payment_url}[/link][/bright_yellow]",
+            style="italic",
+        ),
+    ]
+
+    console.print(
+        Panel(
+            Text.assemble(*infos),
+            title="Buy Credits",
+            border_style="bright_cyan",
+            expand=False,
+            title_align="left",
+        )
+    )