From 99cbaeab0be3ab8dd54b9cf61e88f846baa6b7b3 Mon Sep 17 00:00:00 2001 From: asteiner Date: Wed, 22 Jan 2025 08:34:01 +0100 Subject: [PATCH 1/2] add rpc method getTokenLargestAccounts --- solathon/client.py | 19 +++++++++++++++++++ solathon/core/types/__init__.py | 21 +++++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/solathon/client.py b/solathon/client.py index d3f3b47..96970a4 100644 --- a/solathon/client.py +++ b/solathon/client.py @@ -12,6 +12,8 @@ Commitment, LargestAccounts, LargestAccountsType, + LargestTokenAccounts, + LargestTokenAccountsType, PubKeyIdentity, PubKeyIdentityType, RPCResponse, @@ -415,6 +417,21 @@ def get_largest_accounts( if self.clean_response: return [LargestAccounts(account) for account in response["value"]] return response + + def get_largest_token_accounts( + self, mint: Text + ) -> RPCResponse[List[LargestTokenAccountsType]] | List[LargestTokenAccounts]: + """ + Returns the token largest accounts. + + Returns: + RPCResponse: The response from the RPC endpoint. + """ + + response = self.build_and_send_request("getTokenLargestAccounts", [mint]) + if self.clean_response: + return [LargestTokenAccounts(account) for account in response["value"]] + return response def get_leader_schedule( self, @@ -730,6 +747,8 @@ def build_and_send_request( res: RPCResponse = self.http.send(data) if self.clean_response: if "error" in res: + print(res['error']['code']) + print(res['error']['message']) raise RPCRequestError( f"Failed to fetch data from RPC endpoint. Error {res['error']['code']}: {res['error']['message']}" ) diff --git a/solathon/core/types/__init__.py b/solathon/core/types/__init__.py index 85b96cc..8f59a10 100644 --- a/solathon/core/types/__init__.py +++ b/solathon/core/types/__init__.py @@ -69,6 +69,27 @@ class LargestAccounts: def __init__(self, response: LargestAccountsType) -> None: self.lamports = response['lamports'] self.address = response['address'] + +class LargestTokenAccountsType(TypedDict): + ''' + JSON Response type of Largest Token Accounts received by RPC + ''' + address: set + amount: int + decimals: int + uiAmount: float + uiAmountString: str + +class LargestTokenAccounts: + ''' + Convert Largest Token Accounts JSON to Class + ''' + def __init__(self, response: LargestTokenAccountsType) -> None: + self.address = response['address'] + self.amount = response['amount'] + self.decimals = response['decimals'] + self.uiAmount = response['uiAmount'] + self.uiAmountString = response['uiAmountString'] class RecentPerformanceSamplesType(TypedDict): ''' From 7fc9aa4dce88e1a5cc608162ba98c16c678b690e Mon Sep 17 00:00:00 2001 From: asteiner Date: Wed, 22 Jan 2025 08:36:03 +0100 Subject: [PATCH 2/2] remove print --- solathon/client.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/solathon/client.py b/solathon/client.py index 96970a4..458ea37 100644 --- a/solathon/client.py +++ b/solathon/client.py @@ -747,8 +747,6 @@ def build_and_send_request( res: RPCResponse = self.http.send(data) if self.clean_response: if "error" in res: - print(res['error']['code']) - print(res['error']['message']) raise RPCRequestError( f"Failed to fetch data from RPC endpoint. Error {res['error']['code']}: {res['error']['message']}" )