feat: Allow partial fetching from debank (#269)

Author: Vinsho

blockapi/test/v2/api/debank/test_debank.py

@@ -2,6 +2,7 @@
 
 from blockapi.utils.address import make_checksum_address
 from blockapi.v2.api import DebankApi
+from blockapi.v2.models import FetchResult
 
 
 def test_build_balance_request_url(debank_api):
@@ -57,6 +58,79 @@ def test_build_portfolio_request_url(debank_api):
     )
 
 
+def test_build_token_list_for_chain_request_url(debank_api):
+    url = debank_api._build_request_url(
+        'get_token_list_for_chain',
+        address='0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca',
+        chain_id='eth',
+        is_all=True,
+    )
+    assert (
+        url
+        == 'https://pro-openapi.debank.com/v1/user/token_list?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&chain_id=eth&is_all=True'
+    )
+
+
+def test_build_protocol_for_address_request_url(debank_api):
+    url = debank_api._build_request_url(
+        'get_protocol_for_address',
+        address='0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca',
+        protocol_id='yflink',
+    )
+    assert (
+        url
+        == 'https://pro-openapi.debank.com/v1/user/protocol?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&protocol_id=yflink'
+    )
+
+
+def test_parse_pool_for_address_wraps_single_object(
+    debank_api, yflink_protocol_response_raw, portfolio_response, requests_mock
+):
+    # /v1/user/protocol returns a single protocol object (dict, not list).
+    # parse_pool_for_address must wrap it so DebankPortfolioParser can iterate.
+    requests_mock.get(
+        'https://pro-openapi.debank.com/v1/protocol/all_list',
+        text=yflink_protocol_response_raw,
+    )
+    requests_mock.get(
+        'https://pro-openapi.debank.com/v1/user/protocol?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&protocol_id=avax_traderjoexyz_lending',
+        json=portfolio_response,
+    )
+    debank_api._protocol_cache.invalidate()
+    fetched = debank_api.fetch_protocol_for_address(
+        '0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca',
+        'avax_traderjoexyz_lending',
+    )
+    parsed = debank_api.parse_pool_for_address(fetched)
+    assert parsed.errors is None
+    assert len(parsed.data) > 0
+    assert (
+        parsed.data[0].pool_info.pool_id == '0xdc13687554205e5b89ac783db14bb5bba4a1edac'
+    )
+
+
+def test_parse_pool_for_address_handles_empty_response(debank_api, protocol_cache):
+    # When Debank returns null/None for an address with no positions in this protocol.
+    protocol_cache.update({})
+    parsed = debank_api.parse_pool_for_address(FetchResult(status_code=200, data=None))
+    assert parsed.errors is None
+    assert parsed.data == []
+
+
+def test_fetch_token_list_for_chain_uses_api_key(
+    debank_api, protocol_cache, requests_mock
+):
+    req = requests_mock.get(
+        'https://pro-openapi.debank.com/v1/user/token_list?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&chain_id=eth&is_all=True',
+        text='[]',
+    )
+    protocol_cache.update({})
+    debank_api.fetch_token_list_for_chain(
+        '0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca', 'eth'
+    )
+    assert req.last_request.headers.get('AccessKey') == 'dummy-key'
+
+
 def test_error_response_returns_empty_balances(
     debank_api, protocol_cache, error_response_raw, requests_mock
 ):

blockapi/v2/api/debank.py

@@ -735,6 +735,12 @@ class DebankApi(CustomizableBlockchainApi, BalanceMixin, IPortfolio):
         'get_protocols': '/v1/protocol/all_list',
         'usage': '/v1/account/units',
         'get_complex_app_list': '/v1/user/complex_app_list?id={address}',
+        'get_token_list_for_chain': (
+            '/v1/user/token_list?id={address}&chain_id={chain_id}&is_all={is_all}'
+        ),
+        'get_protocol_for_address': (
+            '/v1/user/protocol?id={address}&protocol_id={protocol_id}'
+        ),
     }
 
     default_protocol_cache = DebankProtocolCache()
@@ -783,6 +789,23 @@ def fetch_debank_apps(self, address: str) -> FetchResult:
             address=address,
         )
 
+    def fetch_token_list_for_chain(self, address: str, chain_id: str) -> FetchResult:
+        return self.get_data(
+            'get_token_list_for_chain',
+            headers=self._headers,
+            address=address,
+            chain_id=chain_id,
+            is_all=self._is_all,
+        )
+
+    def fetch_protocol_for_address(self, address: str, protocol_id: str) -> FetchResult:
+        return self.get_data(
+            'get_protocol_for_address',
+            headers=self._headers,
+            address=address,
+            protocol_id=protocol_id,
+        )
+
     def fetch_protocols(self) -> FetchResult:
         return self.get_data(
             'get_protocols',
@@ -809,6 +832,17 @@ def parse_pools(self, fetch_result: FetchResult) -> ParseResult:
         self._maybe_update_protocols()
         return ParseResult(data=self._portfolio_parser.parse(fetch_result.data))
 
+    def parse_pool_for_address(self, fetch_result: FetchResult) -> ParseResult:
+        if error := self._get_error(fetch_result.data):
+            return ParseResult(errors=[error])
+
+        self._maybe_update_protocols()
+        data = fetch_result.data
+        # /v1/user/protocol returns a single protocol object; wrap in a list
+        # so the portfolio parser (which iterates) handles it uniformly.
+        wrapped = [data] if isinstance(data, dict) else (data or [])
+        return ParseResult(data=self._portfolio_parser.parse(wrapped))
+
     def get_protocols(self) -> Dict[str, Protocol]:
         response = self.get('get_protocols', headers=self._headers)
         if self._has_error(response):