V2 Draft

fireblocks_sdk/__init__.py

@@ -1,3 +1,3 @@
+from fireblocks_sdk.api_types import *
 from fireblocks_sdk.sdk import FireblocksSDK
 from fireblocks_sdk.sdk_token_provider import SdkTokenProvider
-from fireblocks_sdk.api_types import *

fireblocks_sdk/api_types.py

@@ -1,10 +1,10 @@
-
-class TransferPeerPath(object):
+class TransferPeerPath:
     def __init__(self, peer_type, peer_id):
         """Defines a source or a destination for a transfer
 
         Args:
-            peer_type (str): either VAULT_ACCOUNT, EXCHANGE_ACCOUNT, INTERNAL_WALLET, EXTERNAL_WALLET, FIAT_ACCOUNT, NETWORK_CONNECTION, ONE_TIME_ADDRESS or UNKNOWN_PEER
+            peer_type (str): either VAULT_ACCOUNT, EXCHANGE_ACCOUNT, INTERNAL_WALLET, EXTERNAL_WALLET, FIAT_ACCOUNT,
+                                NETWORK_CONNECTION, ONE_TIME_ADDRESS or UNKNOWN_PEER
             peer_id (str): the account/wallet id
         """
 
@@ -14,18 +14,20 @@ def __init__(self, peer_type, peer_id):
         if peer_id is not None:
             self.id = str(peer_id)
 
+
 class DestinationTransferPeerPath(TransferPeerPath):
     def __init__(self, peer_type, peer_id=None, one_time_address=None):
         """Defines a destination for a transfer
 
         Args:
-            peer_type (str): either VAULT_ACCOUNT, EXCHANGE_ACCOUNT, INTERNAL_WALLET, EXTERNAL_WALLET, FIAT_ACCOUNT, NETWORK_CONNECTION, ONE_TIME_ADDRESS or UNKNOWN_PEER
+            peer_type (str): either VAULT_ACCOUNT, EXCHANGE_ACCOUNT, INTERNAL_WALLET, EXTERNAL_WALLET,
+                            FIAT_ACCOUNT, NETWORK_CONNECTION, ONE_TIME_ADDRESS or UNKNOWN_PEER
             peer_id (str): the account/wallet id
             one_time_address (JSON object): The destination address (and tag) for a non whitelisted address.
         """
         TransferPeerPath.__init__(self, peer_type, peer_id)
 
-        if one_time_address != None:
+        if one_time_address:
             self.oneTimeAddress = one_time_address
 
 
@@ -38,18 +40,19 @@ def __init__(self, peer_type, peer_id=None, one_time_address=None):
 CONTRACT_CALL = "CONTRACT_CALL"
 ONE_TIME_ADDRESS = "ONE_TIME_ADDRESS"
 
-TRANSACTION_TYPES = (TRANSACTION_TRANSFER, TRANSACTION_MINT, TRANSACTION_BURN, TRANSACTION_SUPPLY_TO_COMPOUND, TRANSACTION_REDEEM_FROM_COMPOUND, RAW, CONTRACT_CALL, ONE_TIME_ADDRESS)
+TRANSACTION_TYPES = (TRANSACTION_TRANSFER, TRANSACTION_MINT, TRANSACTION_BURN, TRANSACTION_SUPPLY_TO_COMPOUND,
+                     TRANSACTION_REDEEM_FROM_COMPOUND, RAW, CONTRACT_CALL, ONE_TIME_ADDRESS)
 
 TRANSACTION_STATUS_SUBMITTED = "SUBMITTED"
 TRANSACTION_STATUS_QUEUED = "QUEUED"
-TRANSACTION_STATUS_PENDING_SIGNATURE= "PENDING_SIGNATURE"
+TRANSACTION_STATUS_PENDING_SIGNATURE = "PENDING_SIGNATURE"
 TRANSACTION_STATUS_PENDING_AUTHORIZATION = "PENDING_AUTHORIZATION"
 TRANSACTION_STATUS_PENDING_3RD_PARTY_MANUAL_APPROVAL = "PENDING_3RD_PARTY_MANUAL_APPROVAL"
 TRANSACTION_STATUS_PENDING_3RD_PARTY = "PENDING_3RD_PARTY"
-TRANSACTION_STATUS_PENDING = "PENDING" # Deprecated
+TRANSACTION_STATUS_PENDING = "PENDING"  # Deprecated
 TRANSACTION_STATUS_BROADCASTING = "BROADCASTING"
 TRANSACTION_STATUS_CONFIRMING = "CONFIRMING"
-TRANSACTION_STATUS_CONFIRMED = "CONFIRMED" # Deprecated
+TRANSACTION_STATUS_CONFIRMED = "CONFIRMED"  # Deprecated
 TRANSACTION_STATUS_COMPLETED = "COMPLETED"
 TRANSACTION_STATUS_PENDING_AML_SCREENING = "PENDING_AML_SCREENING"
 TRANSACTION_STATUS_PARTIALLY_COMPLETED = "PARTIALLY_COMPLETED"
@@ -61,25 +64,25 @@ def __init__(self, peer_type, peer_id=None, one_time_address=None):
 TRANSACTION_STATUS_BLOCKED = "BLOCKED"
 
 TRANSACTION_STATUS_TYPES = (TRANSACTION_STATUS_SUBMITTED,
-TRANSACTION_STATUS_QUEUED,
-TRANSACTION_STATUS_PENDING_SIGNATURE,
-TRANSACTION_STATUS_PENDING_AUTHORIZATION,
-TRANSACTION_STATUS_PENDING_3RD_PARTY_MANUAL_APPROVAL,
-TRANSACTION_STATUS_PENDING_3RD_PARTY,
-TRANSACTION_STATUS_PENDING,
-TRANSACTION_STATUS_BROADCASTING,
-TRANSACTION_STATUS_CONFIRMING,
-TRANSACTION_STATUS_CONFIRMED,
-TRANSACTION_STATUS_COMPLETED,
-TRANSACTION_STATUS_PENDING_AML_SCREENING,
-TRANSACTION_STATUS_PARTIALLY_COMPLETED,
-TRANSACTION_STATUS_CANCELLING,
-TRANSACTION_STATUS_CANCELLED,
-TRANSACTION_STATUS_REJECTED,
-TRANSACTION_STATUS_FAILED,
-TRANSACTION_STATUS_TIMEOUT,
-TRANSACTION_STATUS_BLOCKED
-)
+                            TRANSACTION_STATUS_QUEUED,
+                            TRANSACTION_STATUS_PENDING_SIGNATURE,
+                            TRANSACTION_STATUS_PENDING_AUTHORIZATION,
+                            TRANSACTION_STATUS_PENDING_3RD_PARTY_MANUAL_APPROVAL,
+                            TRANSACTION_STATUS_PENDING_3RD_PARTY,
+                            TRANSACTION_STATUS_PENDING,
+                            TRANSACTION_STATUS_BROADCASTING,
+                            TRANSACTION_STATUS_CONFIRMING,
+                            TRANSACTION_STATUS_CONFIRMED,
+                            TRANSACTION_STATUS_COMPLETED,
+                            TRANSACTION_STATUS_PENDING_AML_SCREENING,
+                            TRANSACTION_STATUS_PARTIALLY_COMPLETED,
+                            TRANSACTION_STATUS_CANCELLING,
+                            TRANSACTION_STATUS_CANCELLED,
+                            TRANSACTION_STATUS_REJECTED,
+                            TRANSACTION_STATUS_FAILED,
+                            TRANSACTION_STATUS_TIMEOUT,
+                            TRANSACTION_STATUS_BLOCKED
+                            )
 
 VAULT_ACCOUNT = "VAULT_ACCOUNT"
 EXCHANGE_ACCOUNT = "EXCHANGE_ACCOUNT"
@@ -90,7 +93,9 @@ def __init__(self, peer_type, peer_id=None, one_time_address=None):
 NETWORK_CONNECTION = "NETWORK_CONNECTION"
 COMPOUND = "COMPOUND"
 
-PEER_TYPES = (VAULT_ACCOUNT, EXCHANGE_ACCOUNT, INTERNAL_WALLET, EXTERNAL_WALLET, UNKNOWN_PEER, FIAT_ACCOUNT, NETWORK_CONNECTION, COMPOUND, ONE_TIME_ADDRESS)
+PEER_TYPES = (
+    VAULT_ACCOUNT, EXCHANGE_ACCOUNT, INTERNAL_WALLET, EXTERNAL_WALLET, UNKNOWN_PEER, FIAT_ACCOUNT, NETWORK_CONNECTION,
+    COMPOUND, ONE_TIME_ADDRESS)
 
 MPC_ECDSA_SECP256K1 = "MPC_ECDSA_SECP256K1"
 MPC_EDDSA_ED25519 = "MPC_EDDSA_ED25519"
@@ -183,6 +188,7 @@ class FireblocksApiException(Exception):
         message: explanation of the error
         error_code: error code of the error
     """
+
     def __init__(self, message="Fireblocks SDK error", error_code=None):
         self.message = message
         self.error_code = error_code
@@ -195,7 +201,8 @@ class PagedVaultAccountsRequestFilters(object):
     Args
         name_prefix (string, optional): Vault account name prefix
         name_suffix (string, optional): Vault account name suffix
-        min_amount_threshold (number, optional):  The minimum amount for asset to have in order to be included in the results
+        min_amount_threshold (number, optional):  The minimum amount for asset to have in order to be included
+        in the results
         asset_id (string, optional): The asset symbol
         order_by (ASC/DESC, optional): Order of results by vault creation time (default: DESC)
         limit (number, optional): Results page size
@@ -207,7 +214,9 @@ class PagedVaultAccountsRequestFilters(object):
         - You should only insert 'before' or 'after' (or none of them), but not both
         - For default and max 'limit' values please see: https://docs.fireblocks.com/api/swagger-ui/#/
     """
-    def __init__(self, name_prefix=None, name_suffix=None, min_amount_threshold=None, asset_id=None, order_by=None, limit=None, before=None, after=None):
+
+    def __init__(self, name_prefix=None, name_suffix=None, min_amount_threshold=None, asset_id=None, order_by=None,
+                 limit=None, before=None, after=None):
         self.name_prefix = name_prefix
         self.name_suffix = name_suffix
         self.min_amount_threshold = min_amount_threshold

fireblocks_sdk/common/wrappers.py

@@ -0,0 +1,24 @@
+import functools
+from typing import TypeVar
+
+from fireblocks_sdk.entities.deserializable import Deserializable
+
+T = TypeVar('T')
+
+
+def response_deserializer(response_type: T) -> T:
+    def inner(func) -> T:
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs) -> T:
+            return_value = func(*args, **kwargs)
+            if isinstance(return_value, list):
+                return [response_type.deserialize(item) for item in return_value]
+
+            if not issubclass(response_type, Deserializable):
+                return return_value
+
+            return response_type.deserialize(return_value)
+
+        return wrapper
+
+    return inner

fireblocks_sdk/connectors/rest.py

@@ -0,0 +1,66 @@
+import json
+import platform
+from typing import Dict, Union
+
+import pkg_resources
+import requests
+
+from fireblocks_sdk import FireblocksApiException
+from fireblocks_sdk.entities.api_response import ApiResponse
+from fireblocks_sdk.sdk_token_provider import SdkTokenProvider
+
+SDK_BUILD_PLATFORM = pkg_resources.get_build_platform()
+SDK_VERSION = pkg_resources.get_distribution("fireblocks_sdk").version
+
+SDK_PACKAGE = 'fireblocks-sdk-py'
+SDK_USER_AGENT = f'{SDK_PACKAGE}/{SDK_VERSION}; Python {platform.python_version()} ({SDK_BUILD_PLATFORM})'
+
+class RestConnector:
+    def __init__(self, token_provider: SdkTokenProvider, base_url: str, api_key: str, timeout: int) -> None:
+        self.token_provider = token_provider
+        self.base_url = base_url
+        self.api_key = api_key
+        self.timeout = timeout
+
+    def _generate_headers(self, path, body: Union[Dict, None] = None, idempotency_key: str = None):
+        token = self.token_provider.sign_jwt(path, json.dumps(body) if body else json.dumps({}))
+        headers = {
+            "X-API-Key": self.api_key,
+            "Authorization": f"Bearer {token}",
+            "User-Agent": SDK_USER_AGENT
+        }
+
+        if idempotency_key:
+            headers.update({"Idempotency-Key": idempotency_key})
+
+        return headers
+
+    def get(self, path) -> ApiResponse:
+        response = requests.get(self.base_url + path,
+                                headers=self._generate_headers(path), timeout=self.timeout)
+        return self.handle_response(response)
+
+    def delete(self, path) -> ApiResponse:
+        response = requests.delete(
+            self.base_url + path, headers=self._generate_headers(path), timeout=self.timeout)
+        return self.handle_response(response)
+
+    def post(self, path, body=Union[Dict, None], idempotency_key=None) -> ApiResponse:
+        response = requests.post(self.base_url + path, headers=self._generate_headers(path, body, idempotency_key),
+                                 json=body or {}, timeout=self.timeout)
+        return self.handle_response(response)
+
+    def put(self, path, body: Union[Dict, None] = None) -> ApiResponse:
+        headers = self._generate_headers(path, body)
+        headers.update({"Content-Type": "application/json"})
+        response = requests.put(self.base_url + path, headers=headers,
+                                data=json.dumps(body), timeout=self.timeout)
+        return self.handle_response(response)
+
+    @staticmethod
+    def handle_response(response) -> ApiResponse:
+        if response.status_code >= 300:
+            raise FireblocksApiException(
+                f"Got an error from fireblocks server: {response.content}")
+
+        return ApiResponse(response.status_code, response.json(), response.headers)

fireblocks_sdk/demo/run_demo.py

@@ -0,0 +1,95 @@
+"""
+This is a demo runner to experience with all the SDK functions from the CLI
+In order to use it, please install "inquirer" that is not included as a requirement
+
+pip3 install inquirer
+
+Before running the file, make sure you set API_SERVER_ADDRESS, PRIVATE_KEY_PATH and USER_ID
+to the correct values corresponding the environment you are working with.
+"""
+
+import inspect
+import json
+import os
+from typing import Dict, List
+
+import inquirer
+from inquirer.themes import GreenPassion
+
+from fireblocks_sdk import FireblocksSDK
+
+API_SERVER_ADDRESS = os.getenv('API_SERVER_ADDRESS', 'https://api.fireblocks.io')
+PRIVATE_KEY_PATH = os.getenv('PRIVATE_KEY_PATH', '')
+USER_ID = os.getenv('USER_ID', '')
+
+class DemoRunner:
+    def __init__(self, api_server: str, key_path: str, user_id: str) -> None:
+        private_key = open(key_path, 'r').read()
+        self.client = FireblocksSDK(private_key, user_id, api_server)
+        self.client_methods = self._get_client_methods()
+
+    def run(self):
+        while True:
+            action_question = [
+                inquirer.List('action',
+                              message="What would you like to do?",
+                              choices=self.client_methods.keys()),
+            ]
+
+            action_answer = inquirer.prompt(action_question, theme=GreenPassion())
+            requested_action = str(action_answer['action'])
+            print(f'needed params: {self.client_methods[requested_action]}')
+
+            params_questions = []
+            for param in self.client_methods[requested_action]:
+                params_questions.append(inquirer.Text(f'{param}', f'{param}'))
+
+            entered_params = []
+            if params_questions:
+                entered_params = inquirer.prompt(params_questions, theme=GreenPassion())
+                for k, v in entered_params.items():
+                    if v == '':
+                        entered_params[k] = None
+
+            print(f'entered_params: {entered_params}')
+
+            service, method = requested_action.split('::')
+
+            if entered_params:
+                return_value = getattr(getattr(self.client, service), method)(
+                    *entered_params.values() if entered_params else None)
+            else:
+                return_value = getattr(getattr(self.client, service), method)()
+
+            self.print_obj(return_value)
+            continue
+
+    def print_obj(self, obj):
+        if isinstance(obj, list):
+            for i in range(len(obj)):
+                self.print_obj(obj[i])
+        elif isinstance(obj, Dict):
+            print(json.dumps(obj))
+        elif isinstance(obj, object):
+            print(obj.__dict__)
+
+    def _get_client_methods(self) -> Dict[str, List[str]]:
+        services = ['vault', 'exchange', 'contracts', 'wallets', 'gas_station', 'transfer_tickets', 'transactions',
+                    'web_hooks']
+        methods: Dict[str, List[str]] = {}
+        for service in services:
+            client_service = getattr(self.client, service)
+            object_methods = [f'{service}::{method_name}' for method_name in dir(client_service)
+                              if callable(getattr(client_service, method_name)) and not method_name.startswith('_')]
+
+            for m in object_methods:
+                service, method = m.split('::')
+                method_instance = getattr(getattr(self.client, service), method)
+                methods[m] = list(inspect.signature(method_instance).parameters.keys())
+
+        return methods
+
+
+if __name__ == '__main__':
+    runner = DemoRunner(API_SERVER_ADDRESS, PRIVATE_KEY_PATH, USER_ID)
+    runner.run()

fireblocks_sdk/entities/allocated_balances.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import Dict, Union
+
+
+class AllocatedBalances:
+    def __init__(self, allocation_id: str, total: str, available: str, third_party_account_id: Union[str, None],
+                 affiliation: Union[str, None], virtual_type: Union[str, None], pending: Union[str, None],
+                 frozen: Union[str, None], locked: Union[str, None]) -> None:
+        self.virtual_type = virtual_type
+        self.total = total
+        self.available = available
+        self.pending = pending
+        self.frozen = frozen
+        self.locked = locked
+        self.affiliation = affiliation
+        self.third_party_account_id = third_party_account_id
+        self.allocation_id = allocation_id
+
+    @classmethod
+    def deserialize(cls, data: Dict[str, any]) -> AllocatedBalances:
+        return cls(
+            data.get('allocationId'),
+            data.get('total'),
+            data.get('available'),
+            data.get('thirdPartyAccountId'),
+            data.get('affiliation'),
+            data.get('virtualType'),
+            data.get('pending'),
+            data.get('frozen'),
+            data.get('locked')
+        )

fireblocks_sdk/entities/api_response.py

@@ -0,0 +1,10 @@
+from typing import Dict, TypeVar, Generic
+
+T = TypeVar('T')
+
+
+class ApiResponse(Generic[T]):
+    def __init__(self, status_code: int, content, extras: Dict[str, str]) -> None:
+        self.status_code = status_code
+        self.content = content
+        self.extras = extras