diff --git a/fireblocks_sdk/__init__.py b/fireblocks_sdk/__init__.py index c8c9aa3..45c3ff9 100644 --- a/fireblocks_sdk/__init__.py +++ b/fireblocks_sdk/__init__.py @@ -1,3 +1,4 @@ from fireblocks_sdk.sdk import FireblocksSDK +from fireblocks_sdk.ncw_sdk import FireblocksNCW from fireblocks_sdk.sdk_token_provider import SdkTokenProvider -from fireblocks_sdk.api_types import * \ No newline at end of file +from fireblocks_sdk.api_types import * diff --git a/fireblocks_sdk/ncw_sdk.py b/fireblocks_sdk/ncw_sdk.py new file mode 100644 index 0000000..b24944c --- /dev/null +++ b/fireblocks_sdk/ncw_sdk.py @@ -0,0 +1,160 @@ +from .sdk import FireblocksSDK + + +class FireblocksNCW: + def __init__(self, sdk: FireblocksSDK): + self.sdk = sdk + self._wallet_url = "/v1/wallets" + + def create_wallet(self): + url = "/v1/wallets" + return self.sdk._post_request(url) + + def get_wallets(self): + return self.sdk._get_request(self._wallet_url) + + def get_wallet(self, wallet_id: str): + url = f"{self._wallet_url}/{wallet_id}" + return self.sdk._get_request(url) + + def enable_wallet(self, wallet_id: str, enabled: bool): + url = f"{self._wallet_url}/{wallet_id}/enable" + body = {"enabled": enabled} + return self.sdk._put_request(url, body) + + def create_wallet_account(self, wallet_id: str): + url = f"{self._wallet_url}/{wallet_id}/accounts" + return self.sdk._post_request(url) + + def get_wallet_accounts( + self, + wallet_id: str, + page_cursor: str = None, + page_size: int = None, + sort: str = None, + order: str = None, + enabled: bool = None, + ): + url = f"{self._wallet_url}/{wallet_id}/accounts" + query_params = {} + + if page_cursor: + query_params["pageCursor"] = page_cursor + + if page_size: + query_params["pageSize"] = page_size + + if sort: + query_params["sort"] = sort + + if order: + query_params["order"] = order + + if enabled: + query_params["enabled"] = enabled + + return self.sdk._get_request(url, query_params=query_params) + + def get_wallet_account(self, wallet_id: str, account_id: str): + url = f"{self._wallet_url}/{wallet_id}/accounts/{account_id}" + return self.sdk._get_request(url) + + def get_wallet_assets( + self, + wallet_id: str, + account_id: str, + page_cursor: str = None, + page_size: int = None, + sort: str = None, + order: str = None, + enabled: bool = None, + ): + url = f"{self._wallet_url}/{wallet_id}/accounts/{account_id}/assets" + query_params = {} + + if page_cursor: + query_params["pageCursor"] = page_cursor + + if page_size: + query_params["pageSize"] = page_size + + if sort: + query_params["sort"] = sort + + if order: + query_params["order"] = order + + if enabled: + query_params["enabled"] = enabled + + return self.sdk._get_request(url, query_params=query_params) + + def get_wallet_asset(self, wallet_id: str, account_id: str, asset_id: str): + url = f"{self._wallet_url}/{wallet_id}/accounts/{account_id}/assets/{asset_id}" + return self.sdk._get_request(url) + + def activate_wallet_asset(self, wallet_id: str, account_id: str, asset_id: str): + url = f"{self._wallet_url}/{wallet_id}/accounts/{account_id}/assets/{asset_id}" + return self.sdk._post_request(url) + + def refresh_wallet_asset_balance( + self, wallet_id: str, account_id: str, asset_id: str + ): + url = f"{self._wallet_url}/{wallet_id}/accounts/{account_id}/assets/{asset_id}/balance" + return self.sdk._put_request(url) + + def get_wallet_asset_balance( + self, wallet_id: str, account_id: str, asset_id: str + ): + url = f"{self._wallet_url}/{wallet_id}/accounts/{account_id}/assets/{asset_id}/balance" + return self.sdk._get_request(url) + + def get_wallet_asset_addresses( + self, + wallet_id: str, + account_id: str, + asset_id: str, + page_cursor: str = None, + page_size: int = None, + sort: str = None, + order: str = None, + enabled: bool = None, + ): + url = f"{self._wallet_url}/{wallet_id}/accounts/{account_id}/assets/{asset_id}/addresses" + query_params = {} + + if page_cursor: + query_params["pageCursor"] = page_cursor + + if page_size: + query_params["pageSize"] = page_size + + if sort: + query_params["sort"] = sort + + if order: + query_params["order"] = order + + if enabled: + query_params["enabled"] = enabled + + return self.sdk._get_request(url, query_params=query_params) + + def get_devices(self, wallet_id: str): + url = f"{self._wallet_url}/{wallet_id}/devices" + return self.sdk._get_request(url) + + def enable_device(self, wallet_id: str, device_id: str, enabled: bool): + url = f"{self._wallet_url}/{wallet_id}/devices/{device_id}" + body = {"enabled": enabled} + + return self.sdk._put_request(url, body) + + def invoke_wallet_rpc(self, wallet_id: str, device_id: str, payload: str): + """ + payload: stringified JSON, message originated in the NCW SDK + """ + url = f"{self._wallet_url}/{wallet_id}/devices/{device_id}/rpc" + body = {"payload": payload} + + return self.sdk._post_request(url, body) diff --git a/fireblocks_sdk/sdk.py b/fireblocks_sdk/sdk.py index 3cf0d6e..1a825b6 100644 --- a/fireblocks_sdk/sdk.py +++ b/fireblocks_sdk/sdk.py @@ -3,15 +3,36 @@ import urllib from importlib.metadata import version from operator import attrgetter -from typing import List, Dict, Any, Optional +from typing import List, Dict import requests -from .api_types import FireblocksApiException, TRANSACTION_TYPES, TRANSACTION_STATUS_TYPES, TransferPeerPath, \ - DestinationTransferPeerPath, TransferTicketTerm, TRANSACTION_TRANSFER, SIGNING_ALGORITHM, UnsignedMessage, \ - FEE_LEVEL, PagedVaultAccountsRequestFilters, TransactionDestination, NFTOwnershipStatusValues, IssueTokenRequest, \ - GetAssetWalletsFilters, TimePeriod, GetOwnedCollectionsSortValue, GetOwnedNftsSortValues, GetNftsSortValues, OrderValues, \ - GetOwnedAssetsSortValues, PolicyRule, GetSmartTransferFilters, NFTsWalletTypeValues +from .api_types import ( + FireblocksApiException, + TRANSACTION_TYPES, + TRANSACTION_STATUS_TYPES, + TransferPeerPath, + DestinationTransferPeerPath, + TransferTicketTerm, + TRANSACTION_TRANSFER, + SIGNING_ALGORITHM, + UnsignedMessage, + FEE_LEVEL, + PagedVaultAccountsRequestFilters, + TransactionDestination, + NFTOwnershipStatusValues, + IssueTokenRequest, + GetAssetWalletsFilters, + TimePeriod, + GetOwnedCollectionsSortValue, + GetOwnedNftsSortValues, + GetNftsSortValues, + OrderValues, + GetOwnedAssetsSortValues, + PolicyRule, + GetSmartTransferFilters, + NFTsWalletTypeValues +) from .sdk_token_provider import SdkTokenProvider @@ -23,20 +44,35 @@ def handle_response(response, page_mode=False): if response.status_code >= 300: if type(response_data) is dict: error_code = response_data.get("code") - raise FireblocksApiException("Got an error from fireblocks server: " + response.text, error_code) + raise FireblocksApiException( + "Got an error from fireblocks server: " + response.text, error_code + ) else: - raise FireblocksApiException("Got an error from fireblocks server: " + response.text) + raise FireblocksApiException( + "Got an error from fireblocks server: " + response.text + ) else: if page_mode: - return {'transactions': response_data, - 'pageDetails': {'prevPage': response.headers.get('prev-page', ''), - 'nextPage': response.headers.get('next-page', '')}} + return { + "transactions": response_data, + "pageDetails": { + "prevPage": response.headers.get("prev-page", ""), + "nextPage": response.headers.get("next-page", ""), + }, + } return response_data class FireblocksSDK(object): - - def __init__(self, private_key, api_key, api_base_url="https://api.fireblocks.io", timeout=None, anonymous_platform=False, seconds_jwt_exp=55): + def __init__( + self, + private_key, + api_key, + api_base_url="https://api.fireblocks.io", + timeout=None, + anonymous_platform=False, + seconds_jwt_exp=55, + ): """Creates a new Fireblocks API Client. Args: @@ -51,16 +87,15 @@ def __init__(self, private_key, api_key, api_base_url="https://api.fireblocks.io self.token_provider = SdkTokenProvider(private_key, api_key, seconds_jwt_exp) self.timeout = timeout self.http_session = requests.Session() - self.http_session.headers.update({ - 'X-API-Key': self.api_key, - 'User-Agent': self._get_user_agent(anonymous_platform) - }) + self.http_session.headers.update( + { + "X-API-Key": self.api_key, + "User-Agent": self._get_user_agent(anonymous_platform), + } + ) def get_linked_tokens(self, limit: int = 100, offset: int = 0): - request_filter = { - "limit": limit, - "offset": offset - } + request_filter = {"limit": limit, "offset": offset} url = f"/v1/tokenization/tokens" return self._get_request(url, query_params=request_filter) @@ -81,8 +116,14 @@ def get_nft(self, id: str): return self._get_request(url) - def get_nfts(self, ids: List[str], page_cursor: str = '', page_size: int = 100, - sort: List[GetNftsSortValues] = None, order: OrderValues = None): + def get_nfts( + self, + ids: List[str], + page_cursor: str = "", + page_size: int = 100, + sort: List[GetNftsSortValues] = None, + order: OrderValues = None, + ): """ Example list: "[1,2,3,4]" @@ -90,23 +131,25 @@ def get_nfts(self, ids: List[str], page_cursor: str = '', page_size: int = 100, url = f"/v1/nfts/tokens" if len(ids) <= 0: - raise FireblocksApiException("Invalid token_ids. Should contain at least 1 token id") + raise FireblocksApiException( + "Invalid token_ids. Should contain at least 1 token id" + ) params = { "ids": ",".join(ids), } if page_cursor: - params['pageCursor'] = page_cursor + params["pageCursor"] = page_cursor if page_size: - params['pageSize'] = page_size + params["pageSize"] = page_size if sort: - params['sort'] = ",".join(sort) + params["sort"] = ",".join(sort) if order: - params['order'] = order.value + params["order"] = order.value return self._get_request(url, query_params=params) @@ -119,7 +162,9 @@ def refresh_nft_metadata(self, id: str): url = "/v1/nfts/tokens/" + id return self._put_request(path=url) - def refresh_nft_ownership_by_vault(self, blockchain_descriptor: str, vault_account_id: str): + def refresh_nft_ownership_by_vault( + self, blockchain_descriptor: str, vault_account_id: str + ): """ :param blockchain_descriptor: @@ -130,10 +175,10 @@ def refresh_nft_ownership_by_vault(self, blockchain_descriptor: str, vault_accou params = {} if blockchain_descriptor: - params['blockchainDescriptor'] = blockchain_descriptor + params["blockchainDescriptor"] = blockchain_descriptor if vault_account_id: - params['vaultAccountId'] = vault_account_id + params["vaultAccountId"] = vault_account_id return self._put_request(url, query_params=params) @@ -149,16 +194,16 @@ def get_owned_nfts(self, blockchain_descriptor: str, vault_account_ids: List[str params = {} if blockchain_descriptor: - params['blockchainDescriptor'] = blockchain_descriptor + params["blockchainDescriptor"] = blockchain_descriptor if vault_account_ids: - params['vaultAccountIds'] = ",".join(vault_account_ids) + params["vaultAccountIds"] = ",".join(vault_account_ids) if ids: - params['ids'] = ",".join(ids) + params["ids"] = ",".join(ids) if collection_ids: - params['collectionIds'] = ",".join(collection_ids) + params["collectionIds"] = ",".join(collection_ids) if ncw_account_ids: params['ncwAccountIds'] = ",".join(ncw_account_ids) @@ -170,22 +215,22 @@ def get_owned_nfts(self, blockchain_descriptor: str, vault_account_ids: List[str params['walletType'] = wallet_type.value if page_cursor: - params['pageCursor'] = page_cursor + params["pageCursor"] = page_cursor if page_size: - params['pageSize'] = page_size + params["pageSize"] = page_size if sort: - params['sort'] = ",".join(sort) + params["sort"] = ",".join(sort) if order: - params['order'] = order.value + params["order"] = order.value if status: - params['status'] = status.value + params["status"] = status.value if search: - params['search'] = search + params["search"] = search return self._get_request(url, query_params=params) @@ -249,15 +294,16 @@ def list_owned_assets(self, search: str = None, status: NFTOwnershipStatusValues params['walletType'] = wallet_type.value if page_cursor: - params['pageCursor'] = page_cursor + params["pageCursor"] = page_cursor if page_size: - params['pageSize'] = page_size + params["pageSize"] = page_size if sort: - params['sort'] = ",".join(sort) + params["sort"] = ",".join(sort) if order: + params['order'] = order return self._get_request(url, query_params=params) @@ -278,7 +324,13 @@ def get_supported_assets(self): return self._get_request("/v1/supported_assets") - def get_vault_accounts(self, name_prefix=None, name_suffix=None, min_amount_threshold=None, assetId=None): + def get_vault_accounts( + self, + name_prefix=None, + name_suffix=None, + min_amount_threshold=None, + assetId=None, + ): """Gets all vault accounts for your tenant Args: @@ -293,23 +345,25 @@ def get_vault_accounts(self, name_prefix=None, name_suffix=None, min_amount_thre params = {} if name_prefix: - params['namePrefix'] = name_prefix + params["namePrefix"] = name_prefix if name_suffix: - params['nameSuffix'] = name_suffix + params["nameSuffix"] = name_suffix if min_amount_threshold is not None: - params['minAmountThreshold'] = min_amount_threshold + params["minAmountThreshold"] = min_amount_threshold if assetId is not None: - params['assetId'] = assetId + params["assetId"] = assetId if params: url = url + "?" + urllib.parse.urlencode(params) return self._get_request(url) - def get_vault_accounts_with_page_info(self, paged_vault_accounts_request_filters: PagedVaultAccountsRequestFilters): + def get_vault_accounts_with_page_info( + self, paged_vault_accounts_request_filters: PagedVaultAccountsRequestFilters + ): """Gets a page of vault accounts for your tenant according to filters given Args: @@ -317,35 +371,53 @@ def get_vault_accounts_with_page_info(self, paged_vault_accounts_request_filters """ url = f"/v1/vault/accounts_paged" - name_prefix, name_suffix, min_amount_threshold, asset_id, order_by, limit, before, after = \ - attrgetter('name_prefix', 'name_suffix', 'min_amount_threshold', 'asset_id', 'order_by', 'limit', 'before', 'after')( - paged_vault_accounts_request_filters) + ( + name_prefix, + name_suffix, + min_amount_threshold, + asset_id, + order_by, + limit, + before, + after, + ) = attrgetter( + "name_prefix", + "name_suffix", + "min_amount_threshold", + "asset_id", + "order_by", + "limit", + "before", + "after", + )( + paged_vault_accounts_request_filters + ) params = {} if name_prefix: - params['namePrefix'] = name_prefix + params["namePrefix"] = name_prefix if name_suffix: - params['nameSuffix'] = name_suffix + params["nameSuffix"] = name_suffix if min_amount_threshold is not None: - params['minAmountThreshold'] = min_amount_threshold + params["minAmountThreshold"] = min_amount_threshold if asset_id is not None: - params['assetId'] = asset_id + params["assetId"] = asset_id if order_by is not None: - params['orderBy'] = order_by + params["orderBy"] = order_by if limit is not None: - params['limit'] = limit + params["limit"] = limit if before is not None: - params['before'] = before + params["before"] = before if after is not None: - params['after'] = after + params["after"] = after if params: url = url + "?" + urllib.parse.urlencode(params) @@ -353,7 +425,7 @@ def get_vault_accounts_with_page_info(self, paged_vault_accounts_request_filters return self._get_request(url) def get_asset_wallets(self, get_vault_wallets_filters: GetAssetWalletsFilters): - """ Optional filters to apply for request + """Optional filters to apply for request Args total_amount_larger_than (number, optional): The minimum amount for asset to have in order to be included in the results @@ -368,28 +440,34 @@ def get_asset_wallets(self, get_vault_wallets_filters: GetAssetWalletsFilters): """ url = f"/v1/vault/asset_wallets" - total_amount_larger_than, asset_id, order_by, limit, before, after = \ - attrgetter('total_amount_larger_than', 'asset_id', 'order_by', 'limit', 'before', 'after')(get_vault_wallets_filters) + total_amount_larger_than, asset_id, order_by, limit, before, after = attrgetter( + "total_amount_larger_than", + "asset_id", + "order_by", + "limit", + "before", + "after", + )(get_vault_wallets_filters) params = {} if total_amount_larger_than is not None: - params['totalAmountLargerThan'] = total_amount_larger_than + params["totalAmountLargerThan"] = total_amount_larger_than if asset_id is not None: - params['assetId'] = asset_id + params["assetId"] = asset_id if order_by is not None: - params['orderBy'] = order_by + params["orderBy"] = order_by if limit is not None: - params['limit'] = limit + params["limit"] = limit if before is not None: - params['before'] = before + params["before"] = before if after is not None: - params['after'] = after + params["after"] = after if params: url = url + "?" + urllib.parse.urlencode(params) @@ -421,14 +499,20 @@ def get_vault_account_asset(self, vault_account_id, asset_id): return self._get_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}") - def refresh_vault_asset_balance(self, vault_account_id, asset_id, idempotency_key=None): + def refresh_vault_asset_balance( + self, vault_account_id, asset_id, idempotency_key=None + ): """Gets a single vault account asset after forcing refresh from the blockchain Args: vault_account_id (string): The id of the requested account asset_id (string): The symbol of the requested asset (e.g BTC, ETH) """ - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}/balance", {}, idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}/balance", + {}, + idempotency_key, + ) def get_deposit_addresses(self, vault_account_id, asset_id): """Gets deposit addresses for an asset in a vault account @@ -437,7 +521,9 @@ def get_deposit_addresses(self, vault_account_id, asset_id): asset_id (string): The symbol of the requested asset (e.g BTC, ETH) """ - return self._get_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses") + return self._get_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses" + ) def get_unspent_inputs(self, vault_account_id, asset_id): """Gets utxo list for an asset in a vault account @@ -446,10 +532,18 @@ def get_unspent_inputs(self, vault_account_id, asset_id): asset_id (string): The symbol of the requested asset (like BTC, DASH and utxo based assets) """ - return self._get_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}/unspent_inputs") + return self._get_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}/unspent_inputs" + ) - def generate_new_address(self, vault_account_id, asset_id, description=None, customer_ref_id=None, - idempotency_key=None): + def generate_new_address( + self, + vault_account_id, + asset_id, + description=None, + customer_ref_id=None, + idempotency_key=None, + ): """Generates a new address for an asset in a vault account Args: @@ -460,11 +554,15 @@ def generate_new_address(self, vault_account_id, asset_id, description=None, cus idempotency_key (str, optional) """ - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses", - {"description": description or '', "customerRefId": customer_ref_id or ''}, - idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses", + {"description": description or "", "customerRefId": customer_ref_id or ""}, + idempotency_key, + ) - def set_address_description(self, vault_account_id, asset_id, address, tag=None, description=None): + def set_address_description( + self, vault_account_id, asset_id, address, tag=None, description=None + ): """Sets the description of an existing address Args: @@ -475,18 +573,28 @@ def set_address_description(self, vault_account_id, asset_id, address, tag=None, description (string, optional): The description to set, or none for no description """ if tag: - return self._put_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses/{address}:{tag}", - {"description": description or ''}) + return self._put_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses/{address}:{tag}", + {"description": description or ""}, + ) else: - return self._put_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses/{address}", - {"description": description or ''}) + return self._put_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses/{address}", + {"description": description or ""}, + ) def get_network_connections(self): """Gets all network connections for your tenant""" return self._get_request("/v1/network_connections") - def create_network_connection(self, local_network_id: str, remote_network_id: str, routing_policy=None, idempotency_key=None): + def create_network_connection( + self, + local_network_id: str, + remote_network_id: str, + routing_policy=None, + idempotency_key=None, + ): """Creates a network connection Args: localNetworkId (str): The local netowrk profile's id @@ -518,22 +626,23 @@ def remove_network_connection(self, connection_id: str): return self._delete_request(f"/v1/network_connections/{connection_id}") - def set_network_connection_routing_policy(self, connection_id: str, routing_policy=None): + def set_network_connection_routing_policy( + self, connection_id: str, routing_policy=None + ): """Sets routing policy for a network connection Args: connection_id (string): The network connection's id routing_policy (routingPolicy): The desired routing policy """ - body = { - "routingPolicy": routing_policy or {} - } + body = {"routingPolicy": routing_policy or {}} - return self._patch_request(f"/v1/network_connections/{connection_id}/set_routing_policy", body) + return self._patch_request( + f"/v1/network_connections/{connection_id}/set_routing_policy", body + ) def get_discoverable_network_ids(self): - """Gets all discoverable network profiles - """ + """Gets all discoverable network profiles""" return self._get_request(f"/v1/network_ids") @@ -544,10 +653,7 @@ def create_network_id(self, name: str, routing_policy=None): routing_policy (routingPolicy): The desired routing policy for the network """ - body = { - "name": name, - "routingPolicy": routing_policy or {} - } + body = {"name": name, "routingPolicy": routing_policy or {}} return self._post_request(f"/v1/network_ids", body) @@ -574,11 +680,11 @@ def set_network_id_discoverability(self, network_id: str, is_discoverable: bool) is_discoverable: (bool) The desired discoverability to set """ - body = { - "isDiscoverable": is_discoverable - } + body = {"isDiscoverable": is_discoverable} - return self._patch_request(f"/v1/network_ids/{network_id}/set_discoverability", body) + return self._patch_request( + f"/v1/network_ids/{network_id}/set_discoverability", body + ) def set_network_id_routing_policy(self, network_id: str, routing_policy): """Sets routing policy for network profile @@ -587,11 +693,11 @@ def set_network_id_routing_policy(self, network_id: str, routing_policy): routing_policy: (routingPolicy) The desired routing policy """ - body = { - "routingPolicy": routing_policy - } + body = {"routingPolicy": routing_policy} - return self._patch_request(f"/v1/network_ids/{network_id}/set_routing_policy", body) + return self._patch_request( + f"/v1/network_ids/{network_id}/set_routing_policy", body + ) def set_network_id_name(self, network_id: str, name: str): """Sets network profile name @@ -600,9 +706,7 @@ def set_network_id_name(self, network_id: str, name: str): name: (str) The desired network profile's name """ - body = { - "name": name - } + body = {"name": name} return self._patch_request(f"/v1/network_ids/{network_id}/set_name", body) @@ -627,9 +731,13 @@ def get_exchange_account_asset(self, exchange_account_id, asset_id): asset_id (string): The asset to transfer """ - return self._get_request(f"/v1/exchange_accounts/{exchange_account_id}/{asset_id}") + return self._get_request( + f"/v1/exchange_accounts/{exchange_account_id}/{asset_id}" + ) - def transfer_to_subaccount(self, exchange_account_id, subaccount_id, asset_id, amount, idempotency_key=None): + def transfer_to_subaccount( + self, exchange_account_id, subaccount_id, asset_id, amount, idempotency_key=None + ): """Transfer to a subaccount from a main exchange account Args: @@ -639,15 +747,17 @@ def transfer_to_subaccount(self, exchange_account_id, subaccount_id, asset_id, a amount (double): The amount to transfer idempotency_key (str, optional) """ - body = { - "subaccountId": subaccount_id, - "amount": amount - } + body = {"subaccountId": subaccount_id, "amount": amount} - return self._post_request(f"/v1/exchange_accounts/{exchange_account_id}/{asset_id}/transfer_to_subaccount", - body, idempotency_key) + return self._post_request( + f"/v1/exchange_accounts/{exchange_account_id}/{asset_id}/transfer_to_subaccount", + body, + idempotency_key, + ) - def transfer_from_subaccount(self, exchange_account_id, subaccount_id, asset_id, amount, idempotency_key=None): + def transfer_from_subaccount( + self, exchange_account_id, subaccount_id, asset_id, amount, idempotency_key=None + ): """Transfer from a subaccount to a main exchange account Args: @@ -657,13 +767,13 @@ def transfer_from_subaccount(self, exchange_account_id, subaccount_id, asset_id, amount (double): The amount to transfer idempotency_key (str, optional) """ - body = { - "subaccountId": subaccount_id, - "amount": amount - } + body = {"subaccountId": subaccount_id, "amount": amount} - return self._post_request(f"/v1/exchange_accounts/{exchange_account_id}/{asset_id}/transfer_from_subaccount", - body, idempotency_key) + return self._post_request( + f"/v1/exchange_accounts/{exchange_account_id}/{asset_id}/transfer_from_subaccount", + body, + idempotency_key, + ) def get_fiat_accounts(self): """Gets all fiat accounts for your tenant""" @@ -691,7 +801,11 @@ def redeem_to_linked_dda(self, account_id, amount, idempotency_key=None): "amount": amount, } - return self._post_request(f"/v1/fiat_accounts/{account_id}/redeem_to_linked_dda", body, idempotency_key) + return self._post_request( + f"/v1/fiat_accounts/{account_id}/redeem_to_linked_dda", + body, + idempotency_key, + ) def deposit_from_linked_dda(self, account_id, amount, idempotency_key=None): """Deposit to a fiat account from a linked DDA @@ -705,11 +819,26 @@ def deposit_from_linked_dda(self, account_id, amount, idempotency_key=None): "amount": amount, } - return self._post_request(f"/v1/fiat_accounts/{account_id}/deposit_from_linked_dda", body, idempotency_key) + return self._post_request( + f"/v1/fiat_accounts/{account_id}/deposit_from_linked_dda", + body, + idempotency_key, + ) - def get_transactions_with_page_info(self, before=0, after=None, status=None, limit=None, txhash=None, - assets=None, source_type=None, source_id=None, dest_type=None, dest_id=None, - next_or_previous_path=None): + def get_transactions_with_page_info( + self, + before=0, + after=None, + status=None, + limit=None, + txhash=None, + assets=None, + source_type=None, + source_id=None, + dest_type=None, + dest_id=None, + next_or_previous_path=None, + ): """Gets a list of transactions matching the given filters or path. Note that "next_or_previous_path" is mutually exclusive with other parameters. If you wish to iterate over the nextPage/prevPage pages, please provide only the "next_or_previous_path" parameter from `pageDetails` response @@ -738,17 +867,44 @@ def get_transactions_with_page_info(self, before=0, after=None, status=None, lim """ if next_or_previous_path is not None: if not next_or_previous_path: - return {'transactions': [], 'pageDetails': {'prevPage': '', 'nextPage': ''}} - index = next_or_previous_path.index('/v1/') + return { + "transactions": [], + "pageDetails": {"prevPage": "", "nextPage": ""}, + } + index = next_or_previous_path.index("/v1/") length = len(next_or_previous_path) - 1 suffix_path = next_or_previous_path[index:length] return self._get_request(suffix_path, True) else: - return self._get_transactions(before, after, status, limit, None, txhash, assets, source_type, source_id, - dest_type, dest_id, True) - - def get_transactions(self, before=0, after=0, status=None, limit=None, order_by=None, txhash=None, - assets=None, source_type=None, source_id=None, dest_type=None, dest_id=None): + return self._get_transactions( + before, + after, + status, + limit, + None, + txhash, + assets, + source_type, + source_id, + dest_type, + dest_id, + True, + ) + + def get_transactions( + self, + before=0, + after=0, + status=None, + limit=None, + order_by=None, + txhash=None, + assets=None, + source_type=None, + source_id=None, + dest_type=None, + dest_id=None, + ): """Gets a list of transactions matching the given filters Args: @@ -771,11 +927,35 @@ def get_transactions(self, before=0, after=0, status=None, limit=None, order_by= NETWORK_CONNECTION, COMPOUND dest_id (str, optional): Only gets transactions with given dest_id """ - return self._get_transactions(before, after, status, limit, order_by, txhash, assets, source_type, source_id, - dest_type, dest_id) + return self._get_transactions( + before, + after, + status, + limit, + order_by, + txhash, + assets, + source_type, + source_id, + dest_type, + dest_id, + ) - def _get_transactions(self, before, after, status, limit, order_by, txhash, assets, source_type, source_id, - dest_type, dest_id, page_mode=False): + def _get_transactions( + self, + before, + after, + status, + limit, + order_by, + txhash, + assets, + source_type, + source_id, + dest_type, + dest_id, + page_mode=False, + ): path = "/v1/transactions" params = {} @@ -783,27 +963,27 @@ def _get_transactions(self, before, after, status, limit, order_by, txhash, asse raise FireblocksApiException("Got invalid transaction type: " + status) if before: - params['before'] = before + params["before"] = before if after: - params['after'] = after + params["after"] = after if status: - params['status'] = status + params["status"] = status if limit: - params['limit'] = limit + params["limit"] = limit if order_by: - params['orderBy'] = order_by + params["orderBy"] = order_by if txhash: - params['txHash'] = txhash + params["txHash"] = txhash if assets: - params['assets'] = assets + params["assets"] = assets if source_type: - params['sourceType'] = source_type + params["sourceType"] = source_type if source_id: - params['sourceId'] = source_id + params["sourceId"] = source_id if dest_type: - params['destType'] = dest_type + params["destType"] = dest_type if dest_id: - params['destId'] = dest_id + params["destId"] = dest_id if params: path = path + "?" + urllib.parse.urlencode(params) @@ -852,8 +1032,7 @@ def get_external_wallet_asset(self, wallet_id, asset_id): return self._get_request(f"/v1/external_wallets/{wallet_id}/{asset_id}") def get_contract_wallets(self): - """Gets all contract wallets for your tenant - """ + """Gets all contract wallets for your tenant""" return self._get_request(f"/v1/contracts") def get_contract_wallet(self, wallet_id): @@ -900,8 +1079,16 @@ def get_fee_for_asset(self, asset_id): return self._get_request(f"/v1/estimate_network_fee?assetId={asset_id}") - def estimate_fee_for_transaction(self, asset_id, amount, source, destination=None, tx_type=TRANSACTION_TRANSFER, - idempotency_key=None, destinations=None): + def estimate_fee_for_transaction( + self, + asset_id, + amount, + source, + destination=None, + tx_type=TRANSACTION_TRANSFER, + idempotency_key=None, + destinations=None, + ): """Estimates transaction fee Args: @@ -919,28 +1106,37 @@ def estimate_fee_for_transaction(self, asset_id, amount, source, destination=Non if not isinstance(source, TransferPeerPath): raise FireblocksApiException( - "Expected transaction source of type TransferPeerPath, but got type: " + type(source)) + "Expected transaction source of type TransferPeerPath, but got type: " + + type(source) + ) body = { "assetId": asset_id, "amount": amount, "source": source.__dict__, - "operation": tx_type + "operation": tx_type, } if destination: - if not isinstance(destination, (TransferPeerPath, DestinationTransferPeerPath)): + if not isinstance( + destination, (TransferPeerPath, DestinationTransferPeerPath) + ): raise FireblocksApiException( - "Expected transaction fee estimation destination of type DestinationTransferPeerPath or TransferPeerPath, but got type: " + type( - destination)) + "Expected transaction fee estimation destination of type DestinationTransferPeerPath or TransferPeerPath, but got type: " + + type(destination) + ) body["destination"] = destination.__dict__ if destinations: if any([not isinstance(x, TransactionDestination) for x in destinations]): - raise FireblocksApiException("Expected destinations of type TransactionDestination") - body['destinations'] = [dest.__dict__ for dest in destinations] + raise FireblocksApiException( + "Expected destinations of type TransactionDestination" + ) + body["destinations"] = [dest.__dict__ for dest in destinations] - return self._post_request("/v1/transactions/estimate_fee", body, idempotency_key) + return self._post_request( + "/v1/transactions/estimate_fee", body, idempotency_key + ) def cancel_transaction_by_id(self, txid, idempotency_key=None): """Cancels the selected transaction @@ -950,9 +1146,13 @@ def cancel_transaction_by_id(self, txid, idempotency_key=None): idempotency_key (str, optional) """ - return self._post_request(f"/v1/transactions/{txid}/cancel", idempotency_key=idempotency_key) + return self._post_request( + f"/v1/transactions/{txid}/cancel", idempotency_key=idempotency_key + ) - def drop_transaction(self, txid, fee_level=None, requested_fee=None, idempotency_key=None): + def drop_transaction( + self, txid, fee_level=None, requested_fee=None, idempotency_key=None + ): """Drops the selected transaction from the blockchain by replacing it with a 0 ETH transaction to itself Args: @@ -969,9 +1169,18 @@ def drop_transaction(self, txid, fee_level=None, requested_fee=None, idempotency if requested_fee: body["requestedFee"] = requested_fee - return self._post_request(f"/v1/transactions/{txid}/drop", body, idempotency_key) + return self._post_request( + f"/v1/transactions/{txid}/drop", body, idempotency_key + ) - def create_vault_account(self, name, hiddenOnUI=False, customer_ref_id=None, autoFuel=False, idempotency_key=None): + def create_vault_account( + self, + name, + hiddenOnUI=False, + customer_ref_id=None, + autoFuel=False, + idempotency_key=None, + ): """Creates a new vault account. Args: @@ -980,11 +1189,7 @@ def create_vault_account(self, name, hiddenOnUI=False, customer_ref_id=None, aut customer_ref_id (str, optional): The ID for AML providers to associate the owner of funds with transactions idempotency_key (str, optional) """ - body = { - "name": name, - "hiddenOnUI": hiddenOnUI, - "autoFuel": autoFuel - } + body = {"name": name, "hiddenOnUI": hiddenOnUI, "autoFuel": autoFuel} if customer_ref_id: body["customerRefId"] = customer_ref_id @@ -998,7 +1203,10 @@ def hide_vault_account(self, vault_account_id, idempotency_key=None): vault_account_id (str): The vault account Id idempotency_key (str, optional) """ - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/hide", idempotency_key=idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/hide", + idempotency_key=idempotency_key, + ) def unhide_vault_account(self, vault_account_id, idempotency_key=None): """Returns the vault account to being visible in the web console @@ -1007,7 +1215,10 @@ def unhide_vault_account(self, vault_account_id, idempotency_key=None): vault_account_id (str): The vault account Id idempotency_key (str, optional) """ - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/unhide", idempotency_key=idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/unhide", + idempotency_key=idempotency_key, + ) def freeze_transaction_by_id(self, txId, idempotency_key=None): """Freezes the selected transaction @@ -1016,7 +1227,9 @@ def freeze_transaction_by_id(self, txId, idempotency_key=None): txId (str): The transaction ID to freeze idempotency_key (str, optional) """ - return self._post_request(f"/v1/transactions/{txId}/freeze", idempotency_key=idempotency_key) + return self._post_request( + f"/v1/transactions/{txId}/freeze", idempotency_key=idempotency_key + ) def unfreeze_transaction_by_id(self, txId, idempotency_key=None): """Unfreezes the selected transaction @@ -1025,7 +1238,9 @@ def unfreeze_transaction_by_id(self, txId, idempotency_key=None): txId (str): The transaction ID to unfreeze idempotency_key (str, optional) """ - return self._post_request(f"/v1/transactions/{txId}/unfreeze", idempotency_key=idempotency_key) + return self._post_request( + f"/v1/transactions/{txId}/unfreeze", idempotency_key=idempotency_key + ) def update_vault_account(self, vault_account_id, name): """Updates a vault account. @@ -1049,7 +1264,10 @@ def create_vault_asset(self, vault_account_id, asset_id, idempotency_key=None): idempotency_key (str, optional) """ - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}", idempotency_key=idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}", + idempotency_key=idempotency_key, + ) def activate_vault_asset(self, vault_account_id, asset_id, idempotency_key=None): """Retry to create a vault asset for a vault asset that failed @@ -1060,10 +1278,14 @@ def activate_vault_asset(self, vault_account_id, asset_id, idempotency_key=None) idempotency_key (str, optional) """ - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/{asset_id}/activate", - idempotency_key=idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/{asset_id}/activate", + idempotency_key=idempotency_key, + ) - def set_vault_account_customer_ref_id(self, vault_account_id, customer_ref_id, idempotency_key=None): + def set_vault_account_customer_ref_id( + self, vault_account_id, customer_ref_id, idempotency_key=None + ): """Sets an AML/KYT customer reference ID for the vault account Args: @@ -1072,11 +1294,20 @@ def set_vault_account_customer_ref_id(self, vault_account_id, customer_ref_id, i idempotency_key (str, optional) """ - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/set_customer_ref_id", - {"customerRefId": customer_ref_id or ''}, idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/set_customer_ref_id", + {"customerRefId": customer_ref_id or ""}, + idempotency_key, + ) - def set_vault_account_customer_ref_id_for_address(self, vault_account_id, asset_id, address, customer_ref_id=None, - idempotency_key=None): + def set_vault_account_customer_ref_id_for_address( + self, + vault_account_id, + asset_id, + address, + customer_ref_id=None, + idempotency_key=None, + ): """Sets an AML/KYT customer reference ID for the given address Args: @@ -1089,7 +1320,9 @@ def set_vault_account_customer_ref_id_for_address(self, vault_account_id, asset_ return self._post_request( f"/v1/vault/accounts/{vault_account_id}/{asset_id}/addresses/{address}/set_customer_ref_id", - {"customerRefId": customer_ref_id or ''}, idempotency_key) + {"customerRefId": customer_ref_id or ""}, + idempotency_key, + ) def create_contract_wallet(self, name, idempotency_key=None): """Creates a new contract wallet @@ -1099,7 +1332,9 @@ def create_contract_wallet(self, name, idempotency_key=None): """ return self._post_request("/v1/contracts", {"name": name}, idempotency_key) - def create_contract_wallet_asset(self, wallet_id, assetId, address, tag=None, idempotency_key=None): + def create_contract_wallet_asset( + self, wallet_id, assetId, address, tag=None, idempotency_key=None + ): """Creates a new contract wallet asset Args: @@ -1108,7 +1343,11 @@ def create_contract_wallet_asset(self, wallet_id, assetId, address, tag=None, id address (str): The wallet address tag (str): (for ripple only) The ripple account tag """ - return self._post_request(f"/v1/contracts/{wallet_id}/{assetId}", {"address": address, "tag": tag}, idempotency_key) + return self._post_request( + f"/v1/contracts/{wallet_id}/{assetId}", + {"address": address, "tag": tag}, + idempotency_key, + ) def create_external_wallet(self, name, customer_ref_id=None, idempotency_key=None): """Creates a new external wallet @@ -1119,8 +1358,11 @@ def create_external_wallet(self, name, customer_ref_id=None, idempotency_key=Non idempotency_key (str, optional) """ - return self._post_request("/v1/external_wallets", {"name": name, "customerRefId": customer_ref_id or ''}, - idempotency_key) + return self._post_request( + "/v1/external_wallets", + {"name": name, "customerRefId": customer_ref_id or ""}, + idempotency_key, + ) def create_internal_wallet(self, name, customer_ref_id=None, idempotency_key=None): """Creates a new internal wallet @@ -1131,10 +1373,15 @@ def create_internal_wallet(self, name, customer_ref_id=None, idempotency_key=Non idempotency_key (str, optional) """ - return self._post_request("/v1/internal_wallets", {"name": name, "customerRefId": customer_ref_id or ''}, - idempotency_key) + return self._post_request( + "/v1/internal_wallets", + {"name": name, "customerRefId": customer_ref_id or ""}, + idempotency_key, + ) - def create_external_wallet_asset(self, wallet_id, asset_id, address, tag=None, idempotency_key=None): + def create_external_wallet_asset( + self, wallet_id, asset_id, address, tag=None, idempotency_key=None + ): """Creates a new asset within an exiting external wallet Args: @@ -1153,7 +1400,9 @@ def create_external_wallet_asset(self, wallet_id, asset_id, address, tag=None, i f"/v1/external_wallets/{wallet_id}/{asset_id}", body, idempotency_key ) - def create_internal_wallet_asset(self, wallet_id, asset_id, address, tag=None, idempotency_key=None): + def create_internal_wallet_asset( + self, wallet_id, asset_id, address, tag=None, idempotency_key=None + ): """Creates a new asset within an exiting internal wallet Args: @@ -1172,11 +1421,32 @@ def create_internal_wallet_asset(self, wallet_id, asset_id, address, tag=None, i f"/v1/internal_wallets/{wallet_id}/{asset_id}", body, idempotency_key ) - def create_transaction(self, asset_id=None, amount=None, source=None, destination=None, fee=None, gas_price=None, - wait_for_status=False, tx_type=TRANSACTION_TRANSFER, note=None, network_fee=None, - customer_ref_id=None, replace_tx_by_hash=None, extra_parameters=None, destinations=None, - fee_level=None, fail_on_low_fee=None, max_fee=None, gas_limit=None, idempotency_key=None, - external_tx_id=None, treat_as_gross_amount=None, force_sweep=None, priority_fee=None): + def create_transaction( + self, + asset_id=None, + amount=None, + source=None, + destination=None, + fee=None, + gas_price=None, + wait_for_status=False, + tx_type=TRANSACTION_TRANSFER, + note=None, + network_fee=None, + customer_ref_id=None, + replace_tx_by_hash=None, + extra_parameters=None, + destinations=None, + fee_level=None, + fail_on_low_fee=None, + max_fee=None, + gas_limit=None, + idempotency_key=None, + external_tx_id=None, + treat_as_gross_amount=None, + force_sweep=None, + priority_fee=None, + ): """Creates a new transaction Args: @@ -1210,7 +1480,9 @@ def create_transaction(self, asset_id=None, amount=None, source=None, destinatio if source: if not isinstance(source, TransferPeerPath): raise FireblocksApiException( - "Expected transaction source of type TransferPeerPath, but got type: " + type(source)) + "Expected transaction source of type TransferPeerPath, but got type: " + + type(source) + ) body = { "waitForStatus": wait_for_status, @@ -1250,10 +1522,13 @@ def create_transaction(self, asset_id=None, amount=None, source=None, destinatio body["note"] = note if destination: - if not isinstance(destination, (TransferPeerPath, DestinationTransferPeerPath)): + if not isinstance( + destination, (TransferPeerPath, DestinationTransferPeerPath) + ): raise FireblocksApiException( - "Expected transaction destination of type DestinationTransferPeerPath or TransferPeerPath, but got type: " + type( - destination)) + "Expected transaction destination of type DestinationTransferPeerPath or TransferPeerPath, but got type: " + + type(destination) + ) body["destination"] = destination.__dict__ if network_fee: @@ -1270,9 +1545,11 @@ def create_transaction(self, asset_id=None, amount=None, source=None, destinatio if destinations: if any([not isinstance(x, TransactionDestination) for x in destinations]): - raise FireblocksApiException("Expected destinations of type TransactionDestination") + raise FireblocksApiException( + "Expected destinations of type TransactionDestination" + ) - body['destinations'] = [dest.__dict__ for dest in destinations] + body["destinations"] = [dest.__dict__ for dest in destinations] if extra_parameters: body["extraParameters"] = extra_parameters @@ -1344,7 +1621,9 @@ def delete_external_wallet_asset(self, wallet_id, asset_id): return self._delete_request(f"/v1/external_wallets/{wallet_id}/{asset_id}") - def set_customer_ref_id_for_internal_wallet(self, wallet_id, customer_ref_id=None, idempotency_key=None): + def set_customer_ref_id_for_internal_wallet( + self, wallet_id, customer_ref_id=None, idempotency_key=None + ): """Sets an AML/KYT customer reference ID for the specific internal wallet Args: @@ -1353,10 +1632,15 @@ def set_customer_ref_id_for_internal_wallet(self, wallet_id, customer_ref_id=Non idempotency_key (str, optional) """ - return self._post_request(f"/v1/internal_wallets/{wallet_id}/set_customer_ref_id", - {"customerRefId": customer_ref_id or ''}, idempotency_key) + return self._post_request( + f"/v1/internal_wallets/{wallet_id}/set_customer_ref_id", + {"customerRefId": customer_ref_id or ""}, + idempotency_key, + ) - def set_customer_ref_id_for_external_wallet(self, wallet_id, customer_ref_id=None, idempotency_key=None): + def set_customer_ref_id_for_external_wallet( + self, wallet_id, customer_ref_id=None, idempotency_key=None + ): """Sets an AML/KYT customer reference ID for the specific external wallet Args: @@ -1365,15 +1649,20 @@ def set_customer_ref_id_for_external_wallet(self, wallet_id, customer_ref_id=Non idempotency_key (str, optional) """ - return self._post_request(f"/v1/external_wallets/{wallet_id}/set_customer_ref_id", - {"customerRefId": customer_ref_id or ''}, idempotency_key) + return self._post_request( + f"/v1/external_wallets/{wallet_id}/set_customer_ref_id", + {"customerRefId": customer_ref_id or ""}, + idempotency_key, + ) def get_transfer_tickets(self): """Gets all transfer tickets of your tenant""" return self._get_request("/v1/transfer_tickets") - def create_transfer_ticket(self, terms, external_ticket_id=None, description=None, idempotency_key=None): + def create_transfer_ticket( + self, terms, external_ticket_id=None, description=None, idempotency_key=None + ): """Creates a new transfer ticket Args: @@ -1392,9 +1681,11 @@ def create_transfer_ticket(self, terms, external_ticket_id=None, description=Non body["description"] = description if any([not isinstance(x, TransferTicketTerm) for x in terms]): - raise FireblocksApiException("Expected Tranfer Assist ticket's term of type TranferTicketTerm") + raise FireblocksApiException( + "Expected Tranfer Assist ticket's term of type TranferTicketTerm" + ) - body['terms'] = [term.__dict__ for term in terms] + body["terms"] = [term.__dict__ for term in terms] return self._post_request(f"/v1/transfer_tickets", body, idempotency_key) @@ -1425,9 +1716,13 @@ def cancel_transfer_ticket(self, ticket_id, idempotency_key=None): idempotency_key (str, optional) """ - return self._post_request(f"/v1/transfer_tickets/{ticket_id}/cancel", idempotency_key=idempotency_key) + return self._post_request( + f"/v1/transfer_tickets/{ticket_id}/cancel", idempotency_key=idempotency_key + ) - def execute_ticket_term(self, ticket_id, term_id, source=None, idempotency_key=None): + def execute_ticket_term( + self, ticket_id, term_id, source=None, idempotency_key=None + ): """Initiate a transfer ticket transaction Args: @@ -1441,12 +1736,20 @@ def execute_ticket_term(self, ticket_id, term_id, source=None, idempotency_key=N if source: if not isinstance(source, TransferPeerPath): raise FireblocksApiException( - "Expected ticket term source Of type TransferPeerPath, but got type: " + type(source)) + "Expected ticket term source Of type TransferPeerPath, but got type: " + + type(source) + ) body["source"] = source.__dict__ - return self._post_request(f"/v1/transfer_tickets/{ticket_id}/{term_id}/transfer", body, idempotency_key) + return self._post_request( + f"/v1/transfer_tickets/{ticket_id}/{term_id}/transfer", + body, + idempotency_key, + ) - def set_confirmation_threshold_for_txid(self, txid, required_confirmations_number, idempotency_key=None): + def set_confirmation_threshold_for_txid( + self, txid, required_confirmations_number, idempotency_key=None + ): """Set the required number of confirmations for transaction Args: @@ -1455,13 +1758,15 @@ def set_confirmation_threshold_for_txid(self, txid, required_confirmations_numbe idempotency_key (str, optional) """ - body = { - "numOfConfirmations": required_confirmations_number - } + body = {"numOfConfirmations": required_confirmations_number} - return self._post_request(f"/v1/transactions/{txid}/set_confirmation_threshold", body, idempotency_key) + return self._post_request( + f"/v1/transactions/{txid}/set_confirmation_threshold", body, idempotency_key + ) - def set_confirmation_threshold_for_txhash(self, txhash, required_confirmations_number, idempotency_key=None): + def set_confirmation_threshold_for_txhash( + self, txhash, required_confirmations_number, idempotency_key=None + ): """Set the required number of confirmations for transaction by txhash Args: @@ -1470,11 +1775,11 @@ def set_confirmation_threshold_for_txhash(self, txhash, required_confirmations_n idempotency_key (str, optional) """ - body = { - "numOfConfirmations": required_confirmations_number - } + body = {"numOfConfirmations": required_confirmations_number} - return self._post_request(f"/v1/txHash/{txhash}/set_confirmation_threshold", body, idempotency_key) + return self._post_request( + f"/v1/txHash/{txhash}/set_confirmation_threshold", body, idempotency_key + ) def get_public_key_info(self, algorithm, derivation_path, compressed=None): """Get the public key information @@ -1494,7 +1799,9 @@ def get_public_key_info(self, algorithm, derivation_path, compressed=None): url += f"&compressed={compressed}" return self._get_request(url) - def get_public_key_info_for_vault_account(self, asset_id, vault_account_id, change, address_index, compressed=None): + def get_public_key_info_for_vault_account( + self, asset_id, vault_account_id, change, address_index, compressed=None + ): """Get the public key information for a vault account Args: @@ -1511,8 +1818,15 @@ def get_public_key_info_for_vault_account(self, asset_id, vault_account_id, chan return self._get_request(url) - def allocate_funds_to_private_ledger(self, vault_account_id, asset, allocation_id, amount, - treat_as_gross_amount=None, idempotency_key=None): + def allocate_funds_to_private_ledger( + self, + vault_account_id, + asset, + allocation_id, + amount, + treat_as_gross_amount=None, + idempotency_key=None, + ): """Allocate funds from your default balance to a private ledger Args: @@ -1526,11 +1840,19 @@ def allocate_funds_to_private_ledger(self, vault_account_id, asset, allocation_i url = f"/v1/vault/accounts/{vault_account_id}/{asset}/lock_allocation" - return self._post_request(url, {"allocationId": allocation_id, "amount": amount, - "treatAsGrossAmount": treat_as_gross_amount or False}, idempotency_key) + return self._post_request( + url, + { + "allocationId": allocation_id, + "amount": amount, + "treatAsGrossAmount": treat_as_gross_amount or False, + }, + idempotency_key, + ) - def deallocate_funds_from_private_ledger(self, vault_account_id, asset, allocation_id, amount, - idempotency_key=None): + def deallocate_funds_from_private_ledger( + self, vault_account_id, asset, allocation_id, amount, idempotency_key=None + ): """deallocate funds from a private ledger to your default balance Args: @@ -1543,7 +1865,9 @@ def deallocate_funds_from_private_ledger(self, vault_account_id, asset, allocati url = f"/v1/vault/accounts/{vault_account_id}/{asset}/release_allocation" - return self._post_request(url, {"allocationId": allocation_id, "amount": amount}, idempotency_key) + return self._post_request( + url, {"allocationId": allocation_id, "amount": amount}, idempotency_key + ) def get_gas_station_info(self, asset_id=None): """Get configuration and status of the Gas Station account" @@ -1559,7 +1883,9 @@ def get_gas_station_info(self, asset_id=None): return self._get_request(url) - def set_gas_station_configuration(self, gas_threshold, gas_cap, max_gas_price=None, asset_id=None): + def set_gas_station_configuration( + self, gas_threshold, gas_cap, max_gas_price=None, asset_id=None + ): """Set configuration of the Gas Station account Args: @@ -1577,27 +1903,29 @@ def set_gas_station_configuration(self, gas_threshold, gas_cap, max_gas_price=No body = { "gasThreshold": gas_threshold, "gasCap": gas_cap, - "maxGasPrice": max_gas_price + "maxGasPrice": max_gas_price, } return self._put_request(url, body) - def get_vault_assets_balance(self, account_name_prefix=None, account_name_suffix=None): + def get_vault_assets_balance( + self, account_name_prefix=None, account_name_suffix=None + ): """Gets vault assets accumulated balance - Args: - account_name_prefix (string, optional): Vault account name prefix - account_name_suffix (string, optional): Vault account name suffix + Args: + account_name_prefix (string, optional): Vault account name prefix + account_name_suffix (string, optional): Vault account name suffix """ url = f"/v1/vault/assets" params = {} if account_name_prefix: - params['accountNamePrefix'] = account_name_prefix + params["accountNamePrefix"] = account_name_prefix if account_name_suffix: - params['accountNameSuffix'] = account_name_suffix + params["accountNameSuffix"] = account_name_suffix if params: url = url + "?" + urllib.parse.urlencode(params) @@ -1607,8 +1935,8 @@ def get_vault_assets_balance(self, account_name_prefix=None, account_name_suffix def get_vault_balance_by_asset(self, asset_id=None): """Gets vault accumulated balance by asset - Args: - asset_id (str, optional): The asset symbol (e.g BTC, ETH) + Args: + asset_id (str, optional): The asset symbol (e.g BTC, ETH) """ url = f"/v1/vault/assets" @@ -1617,7 +1945,9 @@ def get_vault_balance_by_asset(self, asset_id=None): return self._get_request(url) - def create_raw_transaction(self, raw_message, source=None, asset_id=None, note=None): + def create_raw_transaction( + self, raw_message, source=None, asset_id=None, note=None + ): """Creates a new raw transaction with the specified parameters Args: @@ -1629,17 +1959,26 @@ def create_raw_transaction(self, raw_message, source=None, asset_id=None, note=N if asset_id is None: if raw_message.algorithm not in SIGNING_ALGORITHM: - raise Exception("Got invalid signing algorithm type: " + raw_message.algorithm) + raise Exception( + "Got invalid signing algorithm type: " + raw_message.algorithm + ) if not all([isinstance(x, UnsignedMessage) for x in raw_message.messages]): raise FireblocksApiException("Expected messages of type UnsignedMessage") raw_message.messages = [message.__dict__ for message in raw_message.messages] - return self.create_transaction(asset_id, source=source, tx_type="RAW", - extra_parameters={"rawMessageData": raw_message.__dict__}, note=note) + return self.create_transaction( + asset_id, + source=source, + tx_type="RAW", + extra_parameters={"rawMessageData": raw_message.__dict__}, + note=note, + ) - def get_max_spendable_amount(self, vault_account_id, asset_id, manual_signing=False): + def get_max_spendable_amount( + self, vault_account_id, asset_id, manual_signing=False + ): """Get max spendable amount per asset and vault. Args: @@ -1670,18 +2009,20 @@ def set_auto_fuel(self, vault_account_id, auto_fuel, idempotency_key=None): auto_fuel (boolean): The new value for the autoFuel flag idempotency_key (str, optional) """ - body = { - "autoFuel": auto_fuel - } + body = {"autoFuel": auto_fuel} - return self._post_request(f"/v1/vault/accounts/{vault_account_id}/set_auto_fuel", body, idempotency_key) + return self._post_request( + f"/v1/vault/accounts/{vault_account_id}/set_auto_fuel", + body, + idempotency_key, + ) def validate_address(self, asset_id, address): """Gets vault accumulated balance by asset - Args: - asset_id (str): The asset symbol (e.g XRP, EOS) - address (str): The address to be verified + Args: + asset_id (str): The asset symbol (e.g XRP, EOS) + address (str): The address to be verified """ url = f"/v1/transactions/validate_address/{asset_id}/{address}" @@ -1692,7 +2033,9 @@ def resend_webhooks(self): return self._post_request("/v1/webhooks/resend") - def resend_transaction_webhooks_by_id(self, tx_id, resend_created, resend_status_updated): + def resend_transaction_webhooks_by_id( + self, tx_id, resend_created, resend_status_updated + ): """Resend webhooks of transaction Args: @@ -1702,7 +2045,7 @@ def resend_transaction_webhooks_by_id(self, tx_id, resend_created, resend_status """ body = { "resendCreated": resend_created, - "resendStatusUpdated": resend_status_updated + "resendStatusUpdated": resend_status_updated, } return self._post_request(f"/v1/webhooks/resend/{tx_id}", body) @@ -1714,77 +2057,6 @@ def get_users(self): return self._get_request(url) - def get_users(self) -> List[Dict[str, Any]]: - """ - Gets all Users for your tenant - """ - - url = "/v1/users" - - return self._get_request(url) - - def get_users_groups(self) -> List[Dict[str, Any]]: - """ - Gets all Users Groups for your tenant - """ - - url = "/v1/users_groups" - - return self._get_request(url) - - def get_users_group(self, id: str) -> Dict[str, Any]: - """ - Gets a Users Group by ID - @param id: The ID of the User - """ - - url = f"/v1/users_groups/{id}" - - return self._get_request(url) - - def create_user_group(self, group_name: str, member_ids: Optional[List[str]] = None) -> Dict[str, Any]: - """ - Creates a new Users Group - @param group_name: The name of the Users Group - @param member_ids: The ids of the Users Group members - """ - - url = "/v1/users_groups" - - body = { - "groupName": group_name, - "memberIds": member_ids - } - - return self._post_request(url, body) - - def update_user_group(self, id: str, group_name: Optional[str] = None, member_ids: Optional[List[str]] = None) -> Dict[str, Any]: - """ - Updates a Users Group - @param id: The ID of the Users Group - @param group_name: The name of the Users Group - @param member_ids: The ids of the Users Group members - """ - - url = f"/v1/users_groups/{id}" - - body = { - "groupName": group_name, - "memberIds": member_ids - } - - return self._put_request(url, body) - - def delete_user_group(self, id: str) -> None: - """ - Deletes a Users Group - @param id: The ID of the Users Group - """ - - url = f"/v1/users_groups/{id}" - - return self._delete_request(url) - def get_off_exchanges(self): """ Get your connected off exchanges virtual accounts @@ -1801,7 +2073,7 @@ def get_audit_logs(self, time_period: TimePeriod): url = "/v1/audits" - return self._get_request(url, query_params={'timePeriod': time_period.value}) + return self._get_request(url, query_params={"timePeriod": time_period.value}) def get_off_exchange_by_id(self, off_exchange_id): """ @@ -1825,7 +2097,9 @@ def settle_off_exchange_by_id(self, off_exchange_id, idempotency_key=None): return self._post_request(url, {}, idempotency_key) - def set_fee_payer_configuration(self, base_asset, fee_payer_account_id, idempotency_key=None): + def set_fee_payer_configuration( + self, base_asset, fee_payer_account_id, idempotency_key=None + ): """ Setting fee payer configuration for base asset :param base_asset: ID of the base asset you want to configure fee payer for (for example: SOL) @@ -1835,9 +2109,7 @@ def set_fee_payer_configuration(self, base_asset, fee_payer_account_id, idempote url = f"/v1/fee_payer/{base_asset}" - body = { - "feePayerAccountId": fee_payer_account_id - } + body = {"feePayerAccountId": fee_payer_account_id} return self._post_request(url, body, idempotency_key) @@ -1861,8 +2133,9 @@ def remove_fee_payer_configuration(self, base_asset): return self._delete_request(url) - def get_web3_connections(self, pageCursor=None, pageSize=None, sort=None, - filter=None, order=None): + def get_web3_connections( + self, pageCursor=None, pageSize=None, sort=None, filter=None, order=None + ): """ Get all signer connections of the current user :return: Array of sessions @@ -1872,17 +2145,25 @@ def get_web3_connections(self, pageCursor=None, pageSize=None, sort=None, url = "/v1/connections" optional_params = ["pageCursor", "pageSize", "sort", "filter", "order"] - query_params = {param: method_param.get(param) for param in optional_params - if method_param.get(param)} + query_params = { + param: method_param.get(param) + for param in optional_params + if method_param.get(param) + } if query_params: url = url + "?" + urllib.parse.urlencode(query_params) return self._get_request(url) - def create_web3_connection(self, vault_account_id: str, uri: str, - chain_ids: List[str], fee_level: str = "MEDIUM", - idempotency_key: str = None): + def create_web3_connection( + self, + vault_account_id: str, + uri: str, + chain_ids: List[str], + fee_level: str = "MEDIUM", + idempotency_key: str = None, + ): """ Initiate a new signer connection :param vault_account_id: The id of the requested account @@ -1899,7 +2180,7 @@ def create_web3_connection(self, vault_account_id: str, uri: str, "vaultAccountId": int(vault_account_id), "feeLevel": fee_level, "uri": uri, - "chainIds": chain_ids + "chainIds": chain_ids, } return self._post_request(url, payload, idempotency_key) @@ -1913,9 +2194,7 @@ def submit_web3_connection(self, session_id: str, approve: bool): url = f"/v1/connections/wc/{session_id}" - body = { - "approve": approve - } + body = {"approve": approve} return self._put_request(url, body) @@ -2301,31 +2580,31 @@ def _get_request(self, path, page_mode=False, query_params: Dict = None): if query_params: path = path + "?" + urllib.parse.urlencode(query_params) token = self.token_provider.sign_jwt(path) - headers = { - "Authorization": f"Bearer {token}" - } - response = self.http_session.get(self.base_url + path, headers=headers, timeout=self.timeout) + headers = {"Authorization": f"Bearer {token}"} + response = self.http_session.get( + self.base_url + path, headers=headers, timeout=self.timeout + ) return handle_response(response, page_mode) def _delete_request(self, path): token = self.token_provider.sign_jwt(path) - headers = { - "Authorization": f"Bearer {token}" - } - response = self.http_session.delete(self.base_url + path, headers=headers, timeout=self.timeout) + headers = {"Authorization": f"Bearer {token}"} + response = self.http_session.delete( + self.base_url + path, headers=headers, timeout=self.timeout + ) return handle_response(response) def _post_request(self, path, body=None, idempotency_key=None): body = body or {} token = self.token_provider.sign_jwt(path, body) - headers = { - "Authorization": f"Bearer {token}" - } + headers = {"Authorization": f"Bearer {token}"} if idempotency_key is not None: headers["Idempotency-Key"] = idempotency_key - response = self.http_session.post(self.base_url + path, headers=headers, json=body, timeout=self.timeout) + response = self.http_session.post( + self.base_url + path, headers=headers, json=body, timeout=self.timeout + ) return handle_response(response) def _put_request(self, path, body=None, query_params=None): @@ -2336,26 +2615,33 @@ def _put_request(self, path, body=None, query_params=None): token = self.token_provider.sign_jwt(path, body) headers = { "Authorization": f"Bearer {token}", - "Content-Type": "application/json" + "Content-Type": "application/json", } - response = self.http_session.put(self.base_url + path, headers=headers, data=json.dumps(body), timeout=self.timeout) + response = self.http_session.put( + self.base_url + path, + headers=headers, + data=json.dumps(body), + timeout=self.timeout, + ) return handle_response(response) def _patch_request(self, path, body=None): body = body or {} token = self.token_provider.sign_jwt(path, body) - headers = { - "Authorization": f"Bearer {token}" - } - response = self.http_session.patch(self.base_url + path, headers=headers, json=body, timeout=self.timeout) + headers = {"Authorization": f"Bearer {token}"} + response = self.http_session.patch( + self.base_url + path, headers=headers, json=body, timeout=self.timeout + ) return handle_response(response) @staticmethod def _get_user_agent(anonymous_platform): user_agent = f"fireblocks-sdk-py/{version('fireblocks_sdk')}" if not anonymous_platform: - user_agent += f' ({platform.system()} {platform.release()}; ' \ - f'{platform.python_implementation()} {platform.python_version()}; ' \ - f'{platform.machine()})' + user_agent += ( + f" ({platform.system()} {platform.release()}; " + f"{platform.python_implementation()} {platform.python_version()}; " + f"{platform.machine()})" + ) return user_agent