From ce6fb340fb351145e99daa3d7e11d2393482a66d Mon Sep 17 00:00:00 2001 From: Jason Paulos Date: Thu, 6 Jun 2024 13:59:55 -0400 Subject: [PATCH] Pass extra args to underlying `kmd_request` function --- algosdk/kmd.py | 81 +++++++++++++++++++++++++------------------------- 1 file changed, 41 insertions(+), 40 deletions(-) diff --git a/algosdk/kmd.py b/algosdk/kmd.py index 70dc0edd..a2432cf1 100644 --- a/algosdk/kmd.py +++ b/algosdk/kmd.py @@ -1,5 +1,6 @@ import base64 import json +from typing import Any import urllib.error from urllib import parse from urllib.request import Request, urlopen @@ -66,7 +67,7 @@ def kmd_request(self, method, requrl, params=None, data=None, timeout=30): raise error.KMDHTTPError(e) return json.loads(resp.read().decode("utf-8")) - def versions(self): + def versions(self, **kwargs: Any): """ Get kmd versions. @@ -74,9 +75,9 @@ def versions(self): str[]: list of versions """ req = "/versions" - return self.kmd_request("GET", req)["versions"] + return self.kmd_request("GET", req, **kwargs)["versions"] - def list_wallets(self): + def list_wallets(self, **kwargs: Any): """ List all wallets hosted on node. @@ -84,14 +85,14 @@ def list_wallets(self): dict[]: list of dictionaries containing wallet information """ req = "/wallets" - res = self.kmd_request("GET", req) + res = self.kmd_request("GET", req, **kwargs) if "wallets" in res: return res["wallets"] else: return [] def create_wallet( - self, name, pswd, driver_name="sqlite", master_deriv_key=None + self, name, pswd, driver_name="sqlite", master_deriv_key=None, **kwargs: Any ): """ Create a new wallet. @@ -113,9 +114,9 @@ def create_wallet( } if master_deriv_key: query["master_derivation_key"] = master_deriv_key - return self.kmd_request("POST", req, data=query)["wallet"] + return self.kmd_request("POST", req, data=query, **kwargs)["wallet"] - def get_wallet(self, handle): + def get_wallet(self, handle, **kwargs: Any): """ Get wallet information. @@ -127,9 +128,9 @@ def get_wallet(self, handle): """ req = "/wallet/info" query = {"wallet_handle_token": handle} - return self.kmd_request("POST", req, data=query)["wallet_handle"] + return self.kmd_request("POST", req, data=query, **kwargs)["wallet_handle"] - def init_wallet_handle(self, id, password): + def init_wallet_handle(self, id, password, **kwargs: Any): """ Initialize a handle for the wallet. @@ -142,9 +143,9 @@ def init_wallet_handle(self, id, password): """ req = "/wallet/init" query = {"wallet_id": id, "wallet_password": password} - return self.kmd_request("POST", req, data=query)["wallet_handle_token"] + return self.kmd_request("POST", req, data=query, **kwargs)["wallet_handle_token"] - def release_wallet_handle(self, handle): + def release_wallet_handle(self, handle, **kwargs: Any): """ Deactivate the handle for the wallet. @@ -156,10 +157,10 @@ def release_wallet_handle(self, handle): """ req = "/wallet/release" query = {"wallet_handle_token": handle} - result = self.kmd_request("POST", req, data=query) + result = self.kmd_request("POST", req, data=query, **kwargs) return result == {} - def renew_wallet_handle(self, handle): + def renew_wallet_handle(self, handle, **kwargs: Any): """ Renew the wallet handle. @@ -171,9 +172,9 @@ def renew_wallet_handle(self, handle): """ req = "/wallet/renew" query = {"wallet_handle_token": handle} - return self.kmd_request("POST", req, data=query)["wallet_handle"] + return self.kmd_request("POST", req, data=query, **kwargs)["wallet_handle"] - def rename_wallet(self, id, password, new_name): + def rename_wallet(self, id, password, new_name, **kwargs: Any): """ Rename the wallet. @@ -191,9 +192,9 @@ def rename_wallet(self, id, password, new_name): "wallet_password": password, "wallet_name": new_name, } - return self.kmd_request("POST", req, data=query)["wallet"] + return self.kmd_request("POST", req, data=query, **kwargs)["wallet"] - def export_master_derivation_key(self, handle, password): + def export_master_derivation_key(self, handle, password, **kwargs: Any): """ Get the wallet's master derivation key. @@ -206,10 +207,10 @@ def export_master_derivation_key(self, handle, password): """ req = "/master-key/export" query = {"wallet_handle_token": handle, "wallet_password": password} - result = self.kmd_request("POST", req, data=query) + result = self.kmd_request("POST", req, data=query, **kwargs) return result["master_derivation_key"] - def import_key(self, handle, private_key): + def import_key(self, handle, private_key, **kwargs: Any): """ Import an account into the wallet. @@ -222,9 +223,9 @@ def import_key(self, handle, private_key): """ req = "/key/import" query = {"wallet_handle_token": handle, "private_key": private_key} - return self.kmd_request("POST", req, data=query)["address"] + return self.kmd_request("POST", req, data=query, **kwargs)["address"] - def export_key(self, handle, password, address): + def export_key(self, handle, password, address, **kwargs: Any): """ Return an account private key. @@ -242,9 +243,9 @@ def export_key(self, handle, password, address): "wallet_password": password, "address": address, } - return self.kmd_request("POST", req, data=query)["private_key"] + return self.kmd_request("POST", req, data=query, **kwargs)["private_key"] - def generate_key(self, handle, display_mnemonic=True): + def generate_key(self, handle, display_mnemonic=True, **kwargs: Any): """ Generate a key in the wallet. @@ -258,9 +259,9 @@ def generate_key(self, handle, display_mnemonic=True): """ req = "/key" query = {"wallet_handle_token": handle} - return self.kmd_request("POST", req, data=query)["address"] + return self.kmd_request("POST", req, data=query, **kwargs)["address"] - def delete_key(self, handle, password, address): + def delete_key(self, handle, password, address, **kwargs: Any): """ Delete a key in the wallet. @@ -278,10 +279,10 @@ def delete_key(self, handle, password, address): "wallet_password": password, "address": address, } - result = self.kmd_request("DELETE", req, data=query) + result = self.kmd_request("DELETE", req, data=query, **kwargs) return result == {} - def list_keys(self, handle): + def list_keys(self, handle, **kwargs: Any): """ List all keys in the wallet. @@ -294,12 +295,12 @@ def list_keys(self, handle): req = "/key/list" query = {"wallet_handle_token": handle} - result = self.kmd_request("POST", req, data=query) + result = self.kmd_request("POST", req, data=query, **kwargs) if result: return result["addresses"] return [] - def sign_transaction(self, handle, password, txn, signing_address=None): + def sign_transaction(self, handle, password, txn, signing_address=None, **kwargs: Any): """ Sign a transaction. @@ -322,11 +323,11 @@ def sign_transaction(self, handle, password, txn, signing_address=None): } if signing_address: query["public_key"] = signing_address - result = self.kmd_request("POST", req, data=query) + result = self.kmd_request("POST", req, data=query, **kwargs) result = result["signed_transaction"] return encoding.msgpack_decode(result) - def list_multisig(self, handle): + def list_multisig(self, handle, **kwargs: Any): """ List all multisig accounts in the wallet. @@ -338,12 +339,12 @@ def list_multisig(self, handle): """ req = "/multisig/list" query = {"wallet_handle_token": handle} - result = self.kmd_request("POST", req, data=query) + result = self.kmd_request("POST", req, data=query, **kwargs) if result == {}: return [] return result["addresses"] - def import_multisig(self, handle, multisig): + def import_multisig(self, handle, multisig, **kwargs: Any): """ Import a multisig account into the wallet. @@ -364,9 +365,9 @@ def import_multisig(self, handle, multisig): for s in multisig.subsigs ], } - return self.kmd_request("POST", req, data=query)["address"] + return self.kmd_request("POST", req, data=query, **kwargs)["address"] - def export_multisig(self, handle, address): + def export_multisig(self, handle, address, **kwargs: Any): """ Export a multisig account. @@ -379,7 +380,7 @@ def export_multisig(self, handle, address): """ req = "/multisig/export" query = {"wallet_handle_token": handle, "address": address} - result = self.kmd_request("POST", req, data=query) + result = self.kmd_request("POST", req, data=query, **kwargs) pks = result["pks"] pks = [encoding.encode_address(base64.b64decode(p)) for p in pks] msig = transaction.Multisig( @@ -387,7 +388,7 @@ def export_multisig(self, handle, address): ) return msig - def delete_multisig(self, handle, password, address): + def delete_multisig(self, handle, password, address, **kwargs: Any): """ Delete a multisig account. @@ -405,10 +406,10 @@ def delete_multisig(self, handle, password, address): "wallet_password": password, "address": address, } - result = self.kmd_request("DELETE", req, data=query) + result = self.kmd_request("DELETE", req, data=query, **kwargs) return result == {} - def sign_multisig_transaction(self, handle, password, public_key, mtx): + def sign_multisig_transaction(self, handle, password, public_key, mtx, **kwargs: Any): """ Sign a multisig transaction for the given public key. @@ -439,7 +440,7 @@ def sign_multisig_transaction(self, handle, password, public_key, mtx): signer = base64.b64encode(encoding.decode_address(mtx.auth_addr)) query["signer"] = signer.decode() - result = self.kmd_request("POST", req, data=query)["multisig"] + result = self.kmd_request("POST", req, data=query, **kwargs)["multisig"] msig = encoding.msgpack_decode(result) mtx.multisig = msig return mtx