Envoy CLI and Cleanup Command

envoy/__main__.py

@@ -0,0 +1,93 @@
+import os
+import argparse
+
+from .cli.status import STATUS_ARGS
+from .cli.cleanup import CLEANUP_ARGS
+
+from . import connect
+from .exceptions import CommandError
+
+from dotenv import load_dotenv
+
+
+# Global arguments for all commands
+load_dotenv()
+GLOBAL_ARGS = {
+    ("-u", "--url"): {
+        "help": "the url for the admin API of your Envoy node",
+        "default": os.environ.get("ENVOY_URL", None),
+        "type": str,
+        "metavar": "URL",
+    },
+    ("-c", "--client-id"): {
+        "help": "the client id of the API key to use for authentication",
+        "default": os.environ.get("ENVOY_CLIENT_ID", None),
+        "type": str,
+        "metavar": "ID",
+    },
+    ("-s", "--secret"): {
+        "help": "the client secret of the API key to use for authentication",
+        "default": os.environ.get("ENVOY_CLIENT_SECRET", None),
+        "type": str,
+        "metavar": "SECRET",
+    },
+    ("-t", "--timeout"): {
+        "help": "the number of seconds to wait for a response before timing out",
+        "default": 10.0,
+        "type": float,
+        "metavar": "SEC",
+    },
+}
+
+
+# Define the CLI commands and arguments
+COMMANDS = {
+    "status": STATUS_ARGS,
+    "cleanup": CLEANUP_ARGS,
+}
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        prog="envoy",
+        description="API client for TRISA Envoy nodes and common commands",
+        epilog="Report bugs to https://github.com/trisacrypto/pyenvoy/issues",
+    )
+
+    # Add global arguments to the parser
+    for pargs, kwargs in GLOBAL_ARGS.items():
+        if isinstance(pargs, str):
+            pargs = (pargs,)
+        parser.add_argument(*pargs, **kwargs)
+
+    # Add subparsers for each command
+    subparsers = parser.add_subparsers(title="commands", required=True)
+    for cmd, cargs in COMMANDS.items():
+        subparser = subparsers.add_parser(cmd, description=cargs["description"])
+        for pargs, kwargs in cargs.get("args", {}).items():
+            if isinstance(pargs, str):
+                pargs = (pargs,)
+            subparser.add_argument(*pargs, **kwargs)
+        subparser.set_defaults(func=cargs["func"])
+
+    try:
+        args = parser.parse_args()
+        if args.url is None or args.client_id is None or args.secret is None:
+            parser.error(
+                "missing required client configuration: url, client id, and/or secret"
+            )
+
+        client = connect(
+            url=args.url,
+            client_id=args.client_id,
+            client_secret=args.secret,
+            timeout=args.timeout,
+        )
+
+        args.func(client, args)
+    except CommandError as e:
+        parser.error(str(e))
+
+
+if __name__ == "__main__":
+    main()

envoy/cli/cleanup.py

@@ -0,0 +1,104 @@
+"""
+This module implements the `cleanup` command for the `envoy` CLI.
+"""
+import re
+
+from .prompt import confirm
+from ..exceptions import CommandError, NotFound
+
+
+def is_ulid(value: str) -> bool:
+    return re.match(r'^[0-9A-HJ-NP-TV-Z]{26}$', value) is not None
+
+
+def cleanup(client, args):
+    # Get Counterparty ID and Confirm Counterparty Selection
+    counterparty_id = None
+    if is_ulid(args.counterparty):
+        try:
+            counterparty = client.counterparties.detail(args.counterparty)
+            counterparty_id = counterparty["id"]
+        except NotFound:
+            raise CommandError(f"counterparty '{args.counterparty}' not found")
+    else:
+        counterparty = client.counterparties.search(args.counterparty)
+        if len(counterparty) == 0:
+            raise CommandError(f"counterparty '{args.counterparty}' not found")
+        else:
+            counterparty = counterparty[0]
+            counterparty_id = counterparty["id"]
+
+        if not args.yes and not confirm(f"continue with {counterparty['name']} ({counterparty_id})?"):
+            raise CommandError("could not identify counterparty")
+
+        counterparty_id = counterparty["id"]
+
+    # Get transactions and confirm transaction selection
+    # TODO: use `/v1/counterparties/{counterparty_id}/transfers` when implemented
+    # TODO: handle pagination when implemented
+    transactions = client.transactions.list(
+        params={
+            "status": args.status,
+        }
+    )
+
+    # Filter transactions by counterparty and status
+    transactions = [
+        tx for tx in transactions
+        if tx["counterparty_id"] == counterparty_id and tx["status"] == args.status
+    ]
+
+    # Confirm moving forward with the operation
+    method = "archive" if not args.delete else "delete"
+    if not args.yes and not confirm(f"{method} {len(transactions)} {args.status} transactions for {counterparty['name']} ({counterparty_id})?"):
+        raise CommandError(f"{method} operation cancelled")
+
+    for tx in transactions:
+        if args.dry_run:
+            print(" ".join([
+                f"dry run: {method} transfer {tx['id']} (status {tx['status']})",
+                f"with {tx['counterparty']}: {tx['amount']} {tx['virtual_asset']}",
+                f"from {tx['originator_address']} to {tx['beneficiary_address']}"
+            ]))
+        elif args.delete:
+            client.transactions.delete(tx["id"])
+        else:
+            tx = client.transactions.detail(tx["id"])
+            tx.archive()
+
+
+CLEANUP_ARGS = {
+    "description": "batch archive or delete transactions by counterparty and status",
+    "func": cleanup,
+    "args": {
+        ("-s", "--status"): {
+            "help": "the status of the transactions to cleanup",
+            "default": "review",
+            "type": str,
+            "choices": [
+                "draft", "pending", "repair", "review",
+                "rejected", "accepted", "completed"
+            ],
+        },
+        ("-c", "--counterparty"): {
+            "help": "the counterparty name or ID of the transactions to cleanup",
+            "type": str,
+            "required": True,
+        },
+        ("-D", "--delete"): {
+            "help": "delete the transactions instead of archiving them",
+            "action": "store_true",
+            "default": False,
+        },
+        ("-d", "--dry-run"): {
+            "help": "do not perform the operation, print the affected transactions",
+            "action": "store_true",
+            "default": False,
+        },
+        ("-y", "--yes"): {
+            "help": "answer yes to all confirmation prompts",
+            "action": "store_true",
+            "default": False,
+        },
+    },
+}

envoy/cli/prompt.py

@@ -0,0 +1,14 @@
+from ..exceptions import CommandError
+
+
+def confirm(prompt) -> bool:
+    for _ in range(3):
+        response = input(f"{prompt} [y/N]: ").strip().lower()
+        if response in {'yes', 'y'}:
+            return True
+        elif response in {'no', 'n'}:
+            return False
+        else:
+            print("Invalid response. Please enter 'y'/'yes' or 'n'/'no'.")
+
+    raise CommandError("could not confirm operation")

envoy/cli/status.py

@@ -0,0 +1,32 @@
+"""
+This module implements the `status` command for the `envoy` CLI.
+"""
+
+import json
+
+
+def status(client, args):
+    status = client.status()
+
+    if args.output == "json":
+        print(json.dumps(status, indent=2))
+    elif args.output == "text":
+        print(f"Status:  {status['status']}")
+        print(f"Uptime:  {status['uptime']}")
+        print(f"Version: {status['version']}")
+    else:
+        print(status["status"])
+
+
+STATUS_ARGS = {
+    "description": "reports the status of your Envoy node",
+    "func": status,
+    "args": {
+        ("-o", "--output"): {
+            "help": "the output format to use",
+            "default": "json",
+            "type": str,
+            "choices": ["json", "text"],
+        },
+    },
+}

envoy/client.py

@@ -13,6 +13,7 @@
 from platform import python_version
 from envoy.version import get_version
 
+from typing import Optional
 from requests import Response
 from email.message import Message
 from collections import namedtuple
@@ -23,13 +24,13 @@
 from envoy.credentials import Credentials
 from envoy.exceptions import AuthenticationError, ServerError, ClientError, NotFound
 
-from envoy.accounts import Accounts
-from envoy.transactions import Transactions
-from envoy.counterparties import Counterparties
 from envoy.users import Users
 from envoy.apikeys import APIKeys
+from envoy.accounts import Accounts
 from envoy.utilities import Utilities
 from envoy.auditlogs import AuditLogs
+from envoy.transactions import Transactions
+from envoy.counterparties import Counterparties
 
 try:
     from json import JSONDecodeError
@@ -100,6 +101,8 @@ def __init__(
         self.client_secret = client_secret or os.environ.get(ENV_CLIENT_SECRET, None)
 
         url = url or os.environ.get(ENV_URL, "")
+        if not url:
+            raise ClientError("no envoy url or host specified")
 
         self._creds = None
         self._host = parse_url_host(url)
@@ -159,7 +162,7 @@ def prefix(self):
         return self._prefix
 
     @staticmethod
-    def get_version(short: bool = None) -> str:
+    def get_version(short: Optional[bool] = None) -> str:
         """Returns the Client version."""
         if short is None:
             return get_version()
@@ -171,8 +174,8 @@ def status(self):
 
     def get(
         self,
-        *endpoint: tuple[str],
-        params: dict = None,
+        *endpoint,
+        params: Optional[dict] = None,
         require_authentication: bool = True,
     ):
         self._pre_flight(require_authentication)
@@ -195,8 +198,8 @@ def get(
     def post(
         self,
         data,
-        *endpoint: tuple[str],
-        params: dict = None,
+        *endpoint,
+        params: Optional[dict] = None,
         require_authentication: bool = True,
     ):
         self._pre_flight(require_authentication)
@@ -220,8 +223,8 @@ def post(
     def put(
         self,
         data,
-        *endpoint: tuple[str],
-        params: dict = None,
+        *endpoint,
+        params: Optional[dict] = None,
         require_authentication: bool = True,
     ):
         self._pre_flight(require_authentication)
@@ -244,8 +247,8 @@ def put(
 
     def delete(
         self,
-        *endpoint: tuple[str],
-        params: dict = None,
+        *endpoint,
+        params: Optional[dict] = None,
         require_authentication: bool = True,
     ):
         self._pre_flight(require_authentication)
@@ -318,7 +321,7 @@ def handle(self, rep: Response):
         else:
             raise ValueError(f"unhandled status code {rep.status_code}")
 
-    def _make_endpoint(self, *endpoint: tuple[str], params: dict = None) -> str:
+    def _make_endpoint(self, *endpoint, params: Optional[dict] = None) -> str:
         """
         Creates an API endpoint from the specified resource endpoint, adding the api
         version identifier to the path to construct a valid Envoy URL.

envoy/exceptions.py

@@ -9,6 +9,12 @@ class EnvoyError(Exception):
     """
 
 
+class CommandError(EnvoyError):
+    """
+    Capture command line errors that do not need a traceback.
+    """
+
+
 class ServerError(EnvoyError):
     """
     An error that is raised when the server returns a 500 status code.

envoy/transactions.py

@@ -148,6 +148,17 @@ def send_prepared(self, prepared):
             parent=self,
         )
 
+    def archive(self):
+        return Record(
+            self.client.post(
+                None,
+                *self._endpoint(),
+                "archive",
+                require_authentication=True,
+            ),
+            parent=self,
+        )
+
     def export(self, f: TextIO, params: dict = None):
         """
         Export the transactions CSV file to the file-like object, f. This performs a

envoy/version.py

@@ -5,8 +5,8 @@
 # Module version and package information
 __version_info__ = {
     "major": 1,
-    "minor": 2,
-    "micro": 1,
+    "minor": 3,
+    "micro": 0,
     "releaselevel": "final",
     "post": 0,
     "serial": 0,

setup.py

@@ -123,7 +123,11 @@ def get_description_type(path=PKG_DESCRIBE):
         "packages": find_packages(where=PROJECT, exclude=EXCLUDES),
         "package_data": {},
         "zip_safe": True,
-        "entry_points": {},
+        "entry_points": {
+            "console_scripts": [
+                "envoy=envoy.__main__:main",
+            ],
+        },
         "install_requires": list(get_requires()),
         "python_requires": ">=3.10, <4",
     }