cli: add simple CLI to use ONEKEY platform API

Author: martonilles

onekey_client/cli/__init__.py

@@ -0,0 +1,77 @@
+import sys
+
+import click
+import httpx
+
+from onekey_client import Client
+from .firmware_upload import upload_firmware
+from .misc import list_tenants, get_tenant_token
+
+
+@click.group()
+@click.option(
+    "--api-url",
+    default="https://app.eu.onekey.com/api",
+    show_default=True,
+    help="ONEKEY platform API endpoint",
+)
+@click.option(
+    "--disable-tls-verify",
+    default=False,
+    show_default=True,
+    help="Disable verifying server certificate, use only for testing",
+    is_flag=True,
+)
+@click.option(
+    "--email", help="Email to authenticate on the ONEKEY platform", required=True
+)
+@click.option(
+    "--password",
+    hide_input=True,
+    required=True,
+    prompt=True,
+    help="Password to authenticate on the ONEKEY platform",
+)
+@click.option(
+    "--tenant", "tenant_name", required=True, help="Tenant name on ONEKEY platform"
+)
+@click.pass_context
+def cli(ctx, api_url, disable_tls_verify, email, password, tenant_name):
+    client = Client(api_url=api_url, disable_tls_verify=disable_tls_verify)
+    try:
+        client.login(email, password)
+    except httpx.HTTPStatusError as e:
+        if e.response.status_code == httpx.codes.UNAUTHORIZED:
+            click.echo(f"Authentication failed on {email} @ {api_url}")
+            sys.exit(1)
+        else:
+            click.echo(
+                f"Error connecting to ONEKEY platform: '{api_url}', error: {e.response.status_code}"
+            )
+            sys.exit(2)
+
+    try:
+        tenant = client.get_tenant(tenant_name)
+    except KeyError:
+        click.echo(f"Invalid tenant: {tenant_name}")
+        tenants = client.get_all_tenants()
+        click.echo("Available tenants:")
+        for tenant in tenants:
+            click.echo(f"- {tenant.name} ({tenant.id}")
+        sys.exit(3)
+
+    client.use_tenant(tenant)
+    ctx.obj = client
+
+
+cli.add_command(list_tenants)
+cli.add_command(get_tenant_token)
+cli.add_command(upload_firmware)
+
+
+def main():
+    cli(auto_envvar_prefix="ONEKEY")
+
+
+if __name__ == "__main__":
+    main()

onekey_client/cli/cli.py

@@ -0,0 +1,74 @@
+import sys
+from pathlib import Path
+from typing import Optional
+
+import click
+
+from onekey_client import FirmwareMetadata, Client
+from onekey_client.errors import QueryError
+
+
+@click.command()
+@click.option(
+    "--product", "product_name", required=True, help="Product name to add the firmware"
+)
+@click.option(
+    "--vendor", "vendor_name", required=True, help="Vendor name to add the firmware"
+)
+@click.option(
+    "--product-group",
+    "product_group_name",
+    default="Default",
+    show_default=True,
+    required=True,
+    help="Product group name to add the firmware",
+)
+@click.option("--version", help="Firmware version")
+@click.option("--name", help="Firmware name")
+@click.argument("filename", type=click.Path(exists=True, path_type=Path))
+@click.pass_obj
+def upload_firmware(
+    client: Client,
+    product_name: str,
+    vendor_name: str,
+    product_group_name: str,
+    version: Optional[str],
+    name: Optional[str],
+    filename: Path,
+):
+    """Uploads a firmware to the ONEKEY platform"""
+
+    product_groups = client.get_product_groups()
+
+    try:
+        product_group_id = product_groups[product_group_name]
+    except KeyError:
+        click.echo(f"Missing product group: {product_group_name}")
+        click.echo("Available product groups:")
+        for pg in product_groups.keys():
+            click.echo(f"- {pg}")
+        sys.exit(10)
+
+    if name is None:
+        name = (
+            f"{vendor_name}-{product_name}-{filename.name}"
+            if version is None
+            else f"{vendor_name}-{product_name}-{version}"
+        )
+
+    metadata = FirmwareMetadata(
+        name=name,
+        vendor_name=vendor_name,
+        product_name=product_name,
+        product_group_id=product_group_id,
+        version=version,
+    )
+
+    try:
+        res = client.upload_firmware(metadata, filename, enable_monitoring=False)
+        click.echo(res["id"])
+    except QueryError as e:
+        click.echo("Error during firmware upload:")
+        for error in e._errors:
+            click.echo(f"- {error['message']}")
+        sys.exit(11)

onekey_client/cli/firmware_upload.py

@@ -0,0 +1,23 @@
+import json
+
+import click
+
+from onekey_client import Client
+
+
+@click.command()
+@click.pass_obj
+def list_tenants(client: Client):
+    """List available tenants"""
+
+    tenants = client.get_all_tenants()
+    for tenant in tenants:
+        click.echo(f"{tenant.name} ({tenant.id}")
+
+
+@click.command()
+@click.pass_obj
+def get_tenant_token(client: Client):
+    """Get tenant specific Bearer token"""
+
+    click.echo(json.dumps(client.get_auth_headers()))

onekey_client/cli/misc.py

@@ -33,3 +33,6 @@ black = "^22.3.0"
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
+
+[tool.poetry.scripts]
+onekey = "onekey_client.cli.cli:main"