diff --git a/CHANGELOG.md b/CHANGELOG.md index e78d7fa..27a5f86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +### Added + +- Support list historical balances and list transactions function for address. + ## [0.0.4] - 2024-10-1 ### Added diff --git a/cdp/address.py b/cdp/address.py index f3eadee..90dcdd0 100644 --- a/cdp/address.py +++ b/cdp/address.py @@ -1,3 +1,4 @@ +from collections.abc import Iterator from decimal import Decimal from cdp.asset import Asset @@ -5,6 +6,8 @@ from cdp.balance_map import BalanceMap from cdp.cdp import Cdp from cdp.faucet_transaction import FaucetTransaction +from cdp.historical_balance import HistoricalBalance +from cdp.transaction import Transaction class Address: @@ -98,6 +101,33 @@ def balances(self): return BalanceMap.from_models(response.data) + def historical_balances(self, asset_id) -> Iterator[HistoricalBalance]: + """List historical balances. + + Args: + asset_id (str): The asset ID. + + Returns: + Iterator[HistoricalBalance]: An iterator of HistoricalBalance objects. + + Raises: + Exception: If there's an error listing the historical balances. + + """ + return HistoricalBalance.list(network_id=self.network_id, address_id=self.address_id, asset_id=asset_id) + + def transactions(self) -> Iterator[Transaction]: + """List transactions of the address. + + Returns: + Iterator[Transaction]: An iterator of Transaction objects. + + Raises: + Exception: If there's an error listing the transactions. + + """ + return Transaction.list(network_id=self.network_id, address_id=self.address_id) + def __str__(self) -> str: """Return a string representation of the Address.""" return f"Address: (address_id: {self.address_id}, network_id: {self.network_id})" diff --git a/cdp/api_clients.py b/cdp/api_clients.py index a63b0e1..b93c943 100644 --- a/cdp/api_clients.py +++ b/cdp/api_clients.py @@ -1,11 +1,13 @@ from cdp.cdp_api_client import CdpApiClient from cdp.client.api.addresses_api import AddressesApi from cdp.client.api.assets_api import AssetsApi +from cdp.client.api.balance_history_api import BalanceHistoryApi from cdp.client.api.contract_invocations_api import ContractInvocationsApi from cdp.client.api.external_addresses_api import ExternalAddressesApi from cdp.client.api.networks_api import NetworksApi from cdp.client.api.smart_contracts_api import SmartContractsApi from cdp.client.api.trades_api import TradesApi +from cdp.client.api.transaction_history_api import TransactionHistoryApi from cdp.client.api.transfers_api import TransfersApi from cdp.client.api.wallets_api import WalletsApi @@ -46,6 +48,8 @@ def __init__(self, cdp_client: CdpApiClient) -> None: self._trades: TradesApi | None = None self._contract_invocations: ContractInvocationsApi | None = None self._smart_contracts: SmartContractsApi | None = None + self._balance_history: BalanceHistoryApi | None = None + self._transaction_history: TransactionHistoryApi | None = None @property def wallets(self) -> WalletsApi: @@ -167,6 +171,21 @@ def contract_invocations(self) -> ContractInvocationsApi: self._contract_invocations = ContractInvocationsApi(api_client=self._cdp_client) return self._contract_invocations + @property + def balance_history(self) -> BalanceHistoryApi: + """Get the BalanceHistoryApi client instance. + + Returns: + BalanceHistoryApi: The BalanceHistoryApi client instance. + + Note: + This property lazily initializes the BalanceHistoryApi client on first access. + + """ + if self._balance_history is None: + self._balance_history = BalanceHistoryApi(api_client=self._cdp_client) + return self._balance_history + @property def smart_contracts(self) -> SmartContractsApi: """Get the SmartContractsApi client instance. @@ -181,3 +200,18 @@ def smart_contracts(self) -> SmartContractsApi: if self._smart_contracts is None: self._smart_contracts = SmartContractsApi(api_client=self._cdp_client) return self._smart_contracts + + @property + def transaction_history(self) -> TransactionHistoryApi: + """Get the TransactionHistoryApi client instance. + + Returns: + TransactionHistoryApi: The TransactionHistoryApi client instance. + + Note: + This property lazily initializes the TransactionHistoryApi client on first access. + + """ + if self._transaction_history is None: + self._transaction_history = TransactionHistoryApi(api_client=self._cdp_client) + return self._transaction_history diff --git a/cdp/historical_balance.py b/cdp/historical_balance.py new file mode 100644 index 0000000..f58c98f --- /dev/null +++ b/cdp/historical_balance.py @@ -0,0 +1,129 @@ +from collections.abc import Iterator +from decimal import Decimal + +from cdp.asset import Asset +from cdp.cdp import Cdp +from cdp.client.models.address_historical_balance_list import AddressHistoricalBalanceList +from cdp.client.models.historical_balance import HistoricalBalance as HistoricalBalanceModel + + +class HistoricalBalance: + """A class representing a balance.""" + + def __init__(self, amount: Decimal, asset: Asset, block_height: str, block_hash: str): + """Initialize the Balance class. + + Args: + amount (Decimal): The amount. + asset (Asset): The asset. + block_height (str): the block height where the balance is in. + block_hash (str): the block hash where the balance is in. + + """ + self._amount = amount + self._asset = asset + self._block_height = block_height + self._block_hash = block_hash + + @classmethod + def from_model(cls, model: HistoricalBalanceModel) -> "HistoricalBalance": + """Create a Balance instance from a model. + + Args: + model (BalanceModel): The model representing the balance. + asset_id (Optional[str]): The asset ID. + + Returns: + Balance: The Balance instance. + + """ + asset = Asset.from_model(model.asset) + + return cls( + amount=asset.from_atomic_amount(model.amount), + asset=asset, + block_height=model.block_height, + block_hash=model.block_hash + ) + + @classmethod + def list(cls, network_id: str, address_id: str, asset_id: str) -> Iterator["HistoricalBalance"]: + """List historical balances of an address of an asset. + + Args: + network_id (str): The ID of the network to list historical balance for. + address_id (str): The ID of the address to list historical balance for. + asset_id(str): The asset ID to list historical balance. + + Returns: + Iterator[Transaction]: An iterator of HistoricalBalance objects. + + Raises: + Exception: If there's an error listing the historical_balances. + + """ + page = None + while True: + response: AddressHistoricalBalanceList = Cdp.api_clients.balance_history.list_address_historical_balance( + network_id=network_id, + address_id=address_id, + asset_id=Asset.primary_denomination(asset_id), + limit=100, + page=page, + ) + + for model in response.data: + yield cls.from_model(model) + + if not response.has_more: + break + + page = response.next_page + + @property + def amount(self) -> Decimal: + """Get the amount. + + Returns: + Decimal: The amount. + + """ + return self._amount + + @property + def asset(self) -> Asset: + """Get the asset. + + Returns: + Asset: The asset. + + """ + return self._asset + + @property + def block_height(self) -> str: + """Get the block height. + + Returns: + str: The block height. + + """ + return self._block_height + + @property + def block_hash(self) -> str: + """Get the block hash. + + Returns: + str: The block hash. + + """ + return self._block_hash + + def __str__(self) -> str: + """Return a string representation of the Balance.""" + return f"HistoricalBalance: (amount: {self.amount}, asset: {self.asset}, block_height: {self.block_height}, block_hash: {self.block_hash})" + + def __repr__(self) -> str: + """Return a string representation of the Balance.""" + return str(self) diff --git a/cdp/transaction.py b/cdp/transaction.py index 816ad1a..46e7f2c 100644 --- a/cdp/transaction.py +++ b/cdp/transaction.py @@ -1,11 +1,14 @@ import json +from collections.abc import Iterator from enum import Enum from eth_account.signers.local import LocalAccount from eth_account.typed_transactions import DynamicFeeTransaction from web3 import Web3 +from cdp.cdp import Cdp from cdp.client.models import Transaction as TransactionModel +from cdp.client.models.address_transaction_list import AddressTransactionList class Transaction: @@ -52,6 +55,38 @@ def __init__(self, model: TransactionModel): self._raw: DynamicFeeTransaction | None = None self._signature: str | None = model.signed_payload + @classmethod + def list(cls, network_id: str, address_id: str) -> Iterator["Transaction"]: + """List transactions of the address. + + Args: + network_id (str): The ID of the network to list transaction for. + address_id (str): The ID of the address to list transaction for. + + Returns: + Iterator[Transaction]: An iterator of Transaction objects. + + Raises: + Exception: If there's an error listing the transactions. + + """ + page = None + while True: + response: AddressTransactionList = Cdp.api_clients.transaction_history.list_address_transactions( + network_id=network_id, + address_id=address_id, + limit=1, + page=page, + ) + + for model in response.data: + yield cls(model) + + if not response.has_more: + break + + page = response.next_page + @property def unsigned_payload(self) -> str: """Get the unsigned payload.""" diff --git a/tests/factories/historical_balance_factory.py b/tests/factories/historical_balance_factory.py new file mode 100644 index 0000000..aabd4b2 --- /dev/null +++ b/tests/factories/historical_balance_factory.py @@ -0,0 +1,41 @@ +import pytest + +from cdp.client.models.historical_balance import HistoricalBalance as HistoricalBalanceModel +from cdp.historical_balance import HistoricalBalance + + +@pytest.fixture +def historical_balance_model_factory(asset_model_factory): + """Create and return a factory for creating HistoricalBalance fixtures.""" + + def _create_historical_balance_model( + amount="1000000000000000000", + network_id="base-sepolia", + asset_id="eth", + decimals=18, + block_hash="0xblockhash", + block_height="12345" + ): + asset_model = asset_model_factory(network_id, asset_id, decimals) + return HistoricalBalanceModel( + amount=amount, block_hash=block_hash, block_height=block_height, asset=asset_model) + + return _create_historical_balance_model + + +@pytest.fixture +def historical_balance_factory(asset_factory): + """Create and return a factory for creating HistoricalBalance fixtures.""" + + def _create_historical_balance( + amount="1000000000000000000", + network_id="base-sepolia", + asset_id="eth", + decimals=18, + block_hash="0xblockhash", + block_height="12345" + ): + asset = asset_factory(network_id=network_id, asset_id=asset_id, decimals=decimals) + return HistoricalBalance(amount, asset, block_height, block_hash) + + return _create_historical_balance diff --git a/tests/test_address.py b/tests/test_address.py index b88f781..90a3a48 100644 --- a/tests/test_address.py +++ b/tests/test_address.py @@ -8,6 +8,8 @@ from cdp.client.exceptions import ApiException from cdp.errors import ApiError from cdp.faucet_transaction import FaucetTransaction +from cdp.historical_balance import HistoricalBalance +from cdp.transaction import Transaction def test_address_initialization(address_factory): @@ -157,6 +159,81 @@ def test_address_balances_api_error(mock_api_clients, address_factory): address.balances() +@patch("cdp.Cdp.api_clients") +def test_address_historical_balances(mock_api_clients, address_factory, historical_balance_model_factory): + """Test the historical_balances method of an Address.""" + address = address_factory() + historical_balance_model = historical_balance_model_factory() + + mock_list_historical_balances = Mock() + mock_list_historical_balances.return_value = Mock(data=[historical_balance_model], has_more=False) + mock_api_clients.balance_history.list_address_historical_balance = mock_list_historical_balances + + historical_balances = address.historical_balances("eth") + + assert len(list(historical_balances)) == 1 + assert all(isinstance(h, HistoricalBalance) for h in historical_balances) + mock_list_historical_balances.assert_called_once_with( + network_id=address.network_id, + address_id=address.address_id, + asset_id="eth", + limit=100, + page=None + ) + + +@patch("cdp.Cdp.api_clients") +def test_address_historical_balances_error(mock_api_clients, address_factory): + """Test the historical_balances method of an Address raises an error when the API call fails.""" + address = address_factory() + + mock_list_historical_balances = Mock() + err = ApiException(500, "boom") + mock_list_historical_balances.side_effect = ApiError(err, code="boom", message="boom") + mock_api_clients.balance_history.list_address_historical_balance = mock_list_historical_balances + + with pytest.raises(ApiError): + historical_balances = address.historical_balances("eth") + next(historical_balances) + + +@patch("cdp.Cdp.api_clients") +def test_address_transactions(mock_api_clients, address_factory, transaction_model_factory): + """Test the list transactions method of an Address.""" + address = address_factory() + onchain_transaction_model = transaction_model_factory() + + mock_list_transactions = Mock() + mock_list_transactions.return_value = Mock(data=[onchain_transaction_model], has_more=False) + mock_api_clients.transaction_history.list_address_transactions = mock_list_transactions + + transactions = address.transactions() + + assert len(list(transactions)) == 1 + assert all(isinstance(t, Transaction) for t in transactions) + mock_list_transactions.assert_called_once_with( + network_id=address.network_id, + address_id=address.address_id, + limit=1, + page=None + ) + + +@patch("cdp.Cdp.api_clients") +def test_address_transactions_error(mock_api_clients, address_factory): + """Test the list transactions method of an Address raises an error when the API call fails.""" + address = address_factory() + + mock_list_transactions = Mock() + err = ApiException(500, "boom") + mock_list_transactions.side_effect = ApiError(err, code="boom", message="boom") + mock_api_clients.transaction_history.list_address_transactions = mock_list_transactions + + with pytest.raises(ApiError): + transactions = address.transactions() + next(transactions) + + def test_address_str_representation(address_factory): """Test the str representation of an Address.""" address = address_factory() diff --git a/tests/test_historical_balance.py b/tests/test_historical_balance.py new file mode 100644 index 0000000..a50f126 --- /dev/null +++ b/tests/test_historical_balance.py @@ -0,0 +1,86 @@ +from decimal import Decimal +from unittest.mock import Mock, patch + +import pytest + +from cdp.asset import Asset +from cdp.client.exceptions import ApiException +from cdp.errors import ApiError +from cdp.historical_balance import HistoricalBalance + + +def test_historical_balance_initialization(asset_factory): + """Test historical_balance initialization.""" + asset = asset_factory(asset_id="eth", decimals=18) + + historical_balance = HistoricalBalance(Decimal("1"), asset, "12345", "0xblockhash") + assert historical_balance.amount == Decimal("1") + assert historical_balance.asset == asset + + +def test_historical_balance_from_model(historical_balance_model_factory): + """Test historical_balance from model.""" + historical_balance_model = historical_balance_model_factory() + + balance = HistoricalBalance.from_model(historical_balance_model) + assert balance.amount == Decimal("1") + assert isinstance(balance.asset, Asset) + assert balance.asset.asset_id == "eth" + + +def test_historical_balance_amount(historical_balance_factory): + """Test historical balance amount.""" + historical_balance = historical_balance_factory(amount=1.5) + + assert historical_balance.amount == Decimal("1.5") + + +def test_historical_balance_str_representation(historical_balance_factory): + """Test historical balance string representation.""" + historical_balance = historical_balance_factory(amount=1.5) + assert ( + str(historical_balance) + == "HistoricalBalance: (amount: 1.5, asset: Asset: (asset_id: eth, network_id: base-sepolia, contract_address: None, decimals: 18), block_height: 12345, block_hash: 0xblockhash)" + ) + + +def test_historical_balance_repr(historical_balance_factory): + """Test historical balance repr.""" + historical_balance = historical_balance_factory(amount=1.5) + assert ( + repr(historical_balance) + == "HistoricalBalance: (amount: 1.5, asset: Asset: (asset_id: eth, network_id: base-sepolia, contract_address: None, decimals: 18), block_height: 12345, block_hash: 0xblockhash)" + ) + + +@patch("cdp.Cdp.api_clients") +def test_list_historical_balances(mock_api_clients, historical_balance_model_factory): + """Test the historical_balances method.""" + mock_list_historical_balances = Mock() + mock_list_historical_balances.return_value = Mock(data=[historical_balance_model_factory()], has_more=False) + mock_api_clients.balance_history.list_address_historical_balance = mock_list_historical_balances + + historical_balances = HistoricalBalance.list(network_id="test-network-id", address_id="0xaddressid", asset_id="eth") + + assert len(list(historical_balances)) == 1 + assert all(isinstance(h, HistoricalBalance) for h in historical_balances) + mock_list_historical_balances.assert_called_once_with( + network_id="test-network-id", + address_id="0xaddressid", + asset_id="eth", + limit=100, + page=None + ) + + +@patch("cdp.Cdp.api_clients") +def test_list_historical_balances_error(mock_api_clients): + """Test the historical_balances method getting api error.""" + mock_list_historical_balances = Mock() + err = ApiException(500, "boom") + mock_list_historical_balances.side_effect = ApiError(err, code="boom", message="boom") + mock_api_clients.balance_history.list_address_historical_balance = mock_list_historical_balances + + with pytest.raises(ApiError): + historical_balances = HistoricalBalance.list(network_id="test-network-id", address_id="0xaddressid", asset_id="eth") + next(historical_balances) diff --git a/tests/test_transaction.py b/tests/test_transaction.py index 2186fb2..ea8b58b 100644 --- a/tests/test_transaction.py +++ b/tests/test_transaction.py @@ -1,6 +1,10 @@ +from unittest.mock import Mock, patch + import pytest +from cdp.client.exceptions import ApiException from cdp.client.models.transaction import Transaction as TransactionModel +from cdp.errors import ApiError from cdp.transaction import Transaction @@ -142,3 +146,32 @@ def test_repr(transaction_factory): expected_repr = "Transaction: (transaction_hash: 0xtransactionhash, status: complete)" assert repr(transaction) == expected_repr + + +@patch("cdp.Cdp.api_clients") +def test_list_transactions(mock_api_clients, transaction_model_factory): + """Test the listing of transactions.""" + mock_list_transactions = Mock() + mock_list_transactions.return_value = Mock(data=[transaction_model_factory()], has_more=False) + mock_api_clients.transaction_history.list_address_transactions = mock_list_transactions + + transactions = Transaction.list(network_id="test-network-id", address_id="0xaddressid") + + assert len(list(transactions)) == 1 + assert all(isinstance(t, Transaction) for t in transactions) + mock_list_transactions.assert_called_once_with( + network_id="test-network-id", address_id="0xaddressid", limit=1, page=None + ) + + +@patch("cdp.Cdp.api_clients") +def test_list_transactions_error(mock_api_clients): + """Test the listing of transactions getting api error.""" + mock_list_transactions = Mock() + err = ApiException(500, "boom") + mock_list_transactions.side_effect = ApiError(err, code="boom", message="boom") + mock_api_clients.transaction_history.list_address_transactions = mock_list_transactions + + with pytest.raises(ApiError): + transactions = Transaction.list(network_id="test-network-id", address_id="0xaddressid") + next(transactions)