Moved PbcClient into own package

fin_depo/defi_partisia_blockchain.py

@@ -1,17 +1,10 @@
 """See `PartisiaBlockchainAccountDepoFetcher` for documentation."""
 
-import dataclasses
-import datetime
-import email.utils
-import json
 import logging
-from collections.abc import Mapping
-from decimal import Decimal
-from typing import Any
 
 import fin_defs
+import pbc_client
 import requests
-from frozendict import frozendict
 
 from .data import (
     DepoFetcher,
@@ -25,121 +18,6 @@ from .data import (
 logger = logging.getLogger(__name__)
 
 
-# mainnet: https://reader.partisiablockchain.com
-# testnet: https://node1.testnet.partisiablockchain.com
-
-
-HOSTNAME = 'reader.partisiablockchain.com'
-
-URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local'
-URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global'
-URL_CONTRACT_STATE = 'https://{hostname}/{shard}blockchain/contracts/{address}?requireContractState=false'
-
-
-MPC_DECIMALS = 10000
-
-
-def shard_id_for_address(address: str) -> str:
-    # Very rough implementation
-    if address is None:
-        msg = 'Address must not be None'
-        raise TypeError(msg)
-    if address.endswith('a'):
-        return 'shards/Shard0/'
-    if address.endswith('2'):
-        return 'shards/Shard1/'
-    return 'shards/Shard2/'
-
-
-@dataclasses.dataclass(frozen=True)
-class Balances:
-    update_time: datetime.datetime
-    mpc: Decimal
-    byoc: Mapping[str, Decimal]
-
-
-@dataclasses.dataclass(frozen=True)
-class PbcClient:
-    session: requests.Session
-
-    def get_json(
-        self,
-        url: str,
-        data: Mapping[str, Any] = frozendict(),
-        method='POST',
-    ) -> tuple[Any, datetime.datetime]:
-        headers = {
-            'Content-Type': 'application/json',
-            'Accept': 'application/json',
-        }
-
-        response = self.session.request(
-            method,
-            url,
-            headers=headers,
-            data=json.dumps(data),
-        )
-        response.raise_for_status()
-        date_text = response.headers.get('last-modified') or response.headers.get(
-            'date',
-        )
-        date = email.utils.parsedate_to_datetime(date_text)
-        json_data = response.json()
-        if json_data is None:
-            msg = 'No result data for ' + url
-            raise Exception(msg)
-        return (json_data, date)
-
-    def determine_coins(self) -> list[dict[str, Any]]:
-        data: dict = {'path': []}
-
-        url: str = URL_ACCOUNT_PLUGIN_GLOBAL.format(
-            hostname=HOSTNAME,
-            shard='',
-        )
-
-        json_data, date = self.get_json(url, data=data)
-        return json_data['coins']['coins']
-
-    def get_account_balances(self, address: str) -> Balances:
-        coins = self.determine_coins()
-
-        url = URL_ACCOUNT_PLUGIN.format(
-            hostname=HOSTNAME,
-            shard=shard_id_for_address(address),
-        )
-
-        data: dict = {
-            'path': [
-                {'type': 'field', 'name': 'accounts'},
-                {'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address},
-            ],
-        }
-        account_data, date = self.get_json(url, data=data)
-
-        byoc: dict[str, Decimal] = {}
-        mpc = Decimal(account_data['mpcTokens']) / MPC_DECIMALS
-
-        for coin_idx, amount_data in enumerate(account_data['accountCoins']):
-            coin_data = coins[coin_idx]
-            byoc_balance = Decimal(amount_data['balance'])
-            denominator = Decimal(coin_data['conversionRate']['denominator'])
-            native_balance = byoc_balance / denominator
-            byoc[coin_data['symbol']] = native_balance
-            del coin_idx, coin_data
-
-        return Balances(date, mpc, byoc)
-
-    def get_contract_state(self, address: str) -> tuple[dict, datetime.datetime]:
-        url = URL_CONTRACT_STATE.format(
-            hostname=HOSTNAME,
-            shard=shard_id_for_address(address),
-            address=address,
-        )
-        data: dict = {'path': []}
-        return self.get_json(url, data=data)
-
-
 BYOC_ASSETS = {
     'ETH': fin_defs.WELL_KNOWN_SYMBOLS['ETH'],
     'POLYGON_USDC': fin_defs.WELL_KNOWN_SYMBOLS['USDC'],
@@ -164,7 +42,7 @@ class PartisiaBlockchainAccountDepoFetcher(DepoFetcher):
         self.assert_param('session', requests.Session, session)
         self.assert_param('blockchain_address', str, blockchain_address)
 
-        self.client = PbcClient(session)
+        self.client = pbc_client.PbcClient(session)
         self.blockchain_address = blockchain_address
 
     def get_depo(self) -> DepoGroup:
@@ -188,10 +66,12 @@ class PartisiaBlockchainAccountDepoFetcher(DepoFetcher):
         )
 
     def _get_withdrawals(self) -> list[WithdrawalDetails]:
-        raise UnsupportedOperationException("TODO")
+        msg = 'TODO'
+        raise UnsupportedOperationException(msg)
 
     def _get_deposits(self) -> list[DepositDetails]:
-        raise UnsupportedOperationException("TODO")
+        msg = 'TODO'
+        raise UnsupportedOperationException(msg)
 
     def _get_double_registers(self) -> list[DoubleRegister]:
         double_registers: list[DoubleRegister] = []

requirements.txt

@@ -3,4 +3,5 @@ python-kucoin
 krakenex
 frozendict
 fin-defs @ git+https://gitfub.space/Jmaa/fin-defs.git
+pbc-client @ git+https://gitfub.space/Jmaa/pbc-client.git
 dataclassabc