pyaver.aver_client 

  1import base64
  2import datetime
  3from anchorpy import Provider, Wallet, Program
  4from solana.rpc import types
  5from solana.rpc.async_api import AsyncClient
  6from solana.publickey import PublicKey
  7from solana.transaction import Transaction
  8from solana.rpc.api import Client
  9# from constants import AVER_API_URL_DEVNET, DEVNET_SOLANA_URL, DEFAULT_QUOTE_TOKEN_DEVNET, AVER_PROGRAM_ID_DEVNET_2
 10from .constants import SYS_VAR_CLOCK, get_aver_api_endpoint, get_quote_token, get_solana_endpoint, AVER_PROGRAM_ID
 11from solana.keypair import Keypair
 12from requests import get
 13from spl.token.instructions import get_associated_token_address, create_associated_token_account
 14from .enums import SolanaNetwork
 15from .layouts import CLOCK_STRUCT
 16
 17class AverClient():
 18    """
 19    Aver Client Class
 20
 21    Use AverClient to interact with the Aver Program, Solana network and Aver API
 22    """
 23
 24    connection: AsyncClient
 25    """Solana AsyncClient"""
 26    program: Program
 27    """AnchorPy Program"""
 28    provider: Provider
 29    """AnchorPy Provider"""
 30    aver_api_endpoint: str
 31    """Endpoint is used to make requests to Aver"""
 32    solana_network: SolanaNetwork
 33    """Devnet or Mainnet - must correspond to the network provided in the connection AsyncClient"""
 34    quote_token: PublicKey
 35    """The token mint of the default quote token for markets on this network (i.e. USDC)"""
 36    solana_client: Client
 37    """Solana Client"""
 38    owner: Keypair
 39    """The default payer for transactions on-chain, unless one is specified"""
 40
 41    def __init__(
 42            self, 
 43            program: Program,
 44            aver_api_endpoint: str,
 45            solana_network: SolanaNetwork,
 46        ):
 47        """
 48        Initialises AverClient object. Do not use this function; use AverClient.load() instead
 49
 50        Args:
 51            program (Program): Aver program AnchorPy
 52            aver_api_endpoint (str): Endpoint for Aver API (to be removed soon)
 53            solana_network (SolanaNetwork): Solana network
 54        """
 55        self.connection = program.provider.connection
 56        self.program = program
 57        self.provider = program.provider
 58        self.aver_api_endpoint = aver_api_endpoint
 59        self.solana_network = solana_network
 60        self.quote_token = get_quote_token(solana_network)
 61        self.solana_client = Client(get_solana_endpoint(solana_network))
 62        self.owner = program.provider.wallet.payer
 63
 64    @staticmethod
 65    async def load(
 66            connection: AsyncClient,
 67            owner: Keypair or Wallet, 
 68            opts: types.TxOpts,
 69            network: SolanaNetwork,
 70            program_id: PublicKey = AVER_PROGRAM_ID,
 71        ):
 72            """
 73            Initialises an AverClient object
 74
 75            Args:
 76                connection (AsyncClient): Solana AsyncClient object
 77                owner (KeypairorWallet): Default keypair to pay transaction costs (and rent costs) unless one is otherwise specified for a given transaction.
 78                opts (types.TxOpts): Default options for sending transactions. 
 79                network (SolanaNetwork): Solana network
 80                program_id (PublicKey, optional): Program public key. Defaults to latest AVER_PROGRAM_ID specified in constants.py.
 81
 82            Raises:
 83                Exception: Invalid owner argument
 84
 85            Returns:
 86                AverClient: AverClient
 87            """
 88            if(isinstance(owner, Wallet)):
 89                wallet = owner
 90            elif(isinstance(owner, Keypair)):
 91                wallet = Wallet(owner)
 92            else:
 93                raise Exception('Invalid Owner Passed. Must be Wallet or Keypair')
 94
 95            provider = Provider(
 96                connection,
 97                wallet,
 98                opts
 99            )
100            api_endpoint = get_aver_api_endpoint(network)
101            program = await AverClient.load_program(provider, program_id)
102
103            aver_client = AverClient(program, api_endpoint, network)    
104            return aver_client
105
106    @staticmethod
107    async def load_program(
108            provider: Provider,
109            program_id: PublicKey = AVER_PROGRAM_ID
110        ):
111        """
112        Loads Aver Program
113
114        Args:
115            provider (Provider): Provider
116            program_id (PublicKey, optional): Program public key. Defaults to AVER_PROGRAM_ID.
117
118        Raises:
119            Exception: Program IDL not loaded
120
121        Returns:
122            Program: Program
123        """
124        idl = await Program.fetch_idl(program_id,provider)
125        if(idl):
126            program = Program(idl, program_id, provider)
127            return program
128        else:
129            raise Exception('Program IDL not loaded')
130
131
132    async def close(self):
133        """
134        Closes connection.
135
136        Call this in your program's clean-up function(s)
137        """
138        await self.provider.close()
139
140    #TODO - Move to admin
141    async def check_health(self):
142        url = self.aver_api_endpoint + '/health?format=json'
143        aver_health_response = get(url)
144
145        solana_health_response = await self.provider.connection.get_version()
146
147        return {'api': aver_health_response.json(), 'solana': solana_health_response}
148
149
150    async def get_or_create_associated_token_account(
151            self,
152            owner: PublicKey,
153            payer: Keypair,
154            token_mint: PublicKey = None,
155        ):
156        """
157        Attempts to load an Associate Token Account (ATA) or creates one if not found.
158
159        Args:
160            owner (PublicKey): Public key of the owner of Associated Token Account
161            payer (Keypair): Payer of transaction fee and ATA rent (rent is recoverable)
162            token_mint (PublicKey, optional): ATA token mint public key. Defaults to USDC token according to chosen solana network.
163
164        Returns:
165            PublicKey: ATA PublicKey
166        """
167        token_mint = token_mint if token_mint is not None else self.quote_token
168        associated_token_account = get_associated_token_address(owner=owner, mint=token_mint)
169        response = await self.provider.connection.get_token_account_balance(associated_token_account)
170
171        if not 'result' in response.keys(): 
172            tx = Transaction()
173            tx.add(
174                create_associated_token_account(
175                    payer=payer.public_key,
176                    owner=owner,
177                    mint=token_mint,
178                ),
179            )
180            signers = [payer]
181            response = await self.provider.connection.send_transaction(tx, *signers, opts=self.provider.opts)
182            await self.provider.connection.confirm_transaction(response['result'])
183            return associated_token_account
184        else:
185            return associated_token_account
186
187    async def request_lamport_airdrop(
188            self,
189            amount: int,
190            owner: PublicKey
191        ):
192        """
193        Request an airdrop of lamports (SOL). This method is only supported on devnet.
194
195        Args:
196            amount (int): Lamports to airdrop. Note 1 lamport = 10^-9 SOL. Max of 1 SOL (10^9 lamports) applies.
197            owner (PublicKey): Public key of account to be airdropped
198
199        Returns:
200            RPCResponse: Response
201        """
202        return await self.provider.connection.request_airdrop(
203            pubkey=owner, 
204            lamports=amount, 
205            commitment=self.provider.opts
206            )
207    
208
209    async def request_ata_balance(self, ata: PublicKey):
210        """
211        Fetches the balance for an Associated Token Account (ATA). 
212        Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token.
213        For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.
214
215        Args:
216            ata (PublicKey): ATA public key
217
218        Raises:
219            Exception: Error from response
220
221        Returns:
222            int: Token balance
223        """
224        response = await self.provider.connection.get_token_account_balance(ata, self.provider.opts.preflight_commitment)
225        if 'error' in response:
226            raise Exception(response['error'])
227        return int(response['result']['value']['amount'])
228
229    async def request_token_balance(
230            self,
231            mint: PublicKey,
232            owner: PublicKey
233        ):
234        """
235        Fetches a wallet's token balance, given the wallet's owner and the token mint's public key.
236        Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token.
237        For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.
238
239        Args:
240            mint: The public key of the token mint
241            owner: The public key of the wallet
242
243        Returns:
244            int: Token balance
245
246        """
247        ata = get_associated_token_address(owner, mint)
248        return await self.request_ata_balance(ata)
249
250    async def request_lamport_balance(
251            self,
252            owner: PublicKey
253        ):
254        """
255        Fetches Lamport (SOL) balance for a given wallet
256        Note: the value returned is the integer representation of the balance, where a unit is one lamport (=10^-9 SOL)
257        For example, a value of 1,000,000,000 (lamports) = 1 SOL.
258
259        Args:
260            owner (PublicKey): Owner public key
261
262        Raises:
263            Exception: Error from response
264
265        Returns:
266            int: Lamport balance
267        """
268        response = await self.provider.connection.get_balance(owner, self.provider.opts.preflight_commitment)
269        if 'error' in response:
270            raise Exception(response['error'])
271        return response['result']['value']
272
273    async def get_system_clock_datetime(
274            self
275        ):
276        """
277        Loads current solana system datetime
278
279        Returns:
280            datetime: Current Solana Clock Datetime
281        """
282        raw_data = await self.connection.get_account_info(SYS_VAR_CLOCK)
283        data = raw_data["result"]["value"]["data"][0]
284        parsed_data = CLOCK_STRUCT.parse(base64.decodebytes(data.encode("ascii")))
285        return datetime.datetime.utcfromtimestamp(parsed_data['unix_timestamp'])
286
class AverClient: 
 18class AverClient():
 19    """
 20    Aver Client Class
 21
 22    Use AverClient to interact with the Aver Program, Solana network and Aver API
 23    """
 24
 25    connection: AsyncClient
 26    """Solana AsyncClient"""
 27    program: Program
 28    """AnchorPy Program"""
 29    provider: Provider
 30    """AnchorPy Provider"""
 31    aver_api_endpoint: str
 32    """Endpoint is used to make requests to Aver"""
 33    solana_network: SolanaNetwork
 34    """Devnet or Mainnet - must correspond to the network provided in the connection AsyncClient"""
 35    quote_token: PublicKey
 36    """The token mint of the default quote token for markets on this network (i.e. USDC)"""
 37    solana_client: Client
 38    """Solana Client"""
 39    owner: Keypair
 40    """The default payer for transactions on-chain, unless one is specified"""
 41
 42    def __init__(
 43            self, 
 44            program: Program,
 45            aver_api_endpoint: str,
 46            solana_network: SolanaNetwork,
 47        ):
 48        """
 49        Initialises AverClient object. Do not use this function; use AverClient.load() instead
 50
 51        Args:
 52            program (Program): Aver program AnchorPy
 53            aver_api_endpoint (str): Endpoint for Aver API (to be removed soon)
 54            solana_network (SolanaNetwork): Solana network
 55        """
 56        self.connection = program.provider.connection
 57        self.program = program
 58        self.provider = program.provider
 59        self.aver_api_endpoint = aver_api_endpoint
 60        self.solana_network = solana_network
 61        self.quote_token = get_quote_token(solana_network)
 62        self.solana_client = Client(get_solana_endpoint(solana_network))
 63        self.owner = program.provider.wallet.payer
 64
 65    @staticmethod
 66    async def load(
 67            connection: AsyncClient,
 68            owner: Keypair or Wallet, 
 69            opts: types.TxOpts,
 70            network: SolanaNetwork,
 71            program_id: PublicKey = AVER_PROGRAM_ID,
 72        ):
 73            """
 74            Initialises an AverClient object
 75
 76            Args:
 77                connection (AsyncClient): Solana AsyncClient object
 78                owner (KeypairorWallet): Default keypair to pay transaction costs (and rent costs) unless one is otherwise specified for a given transaction.
 79                opts (types.TxOpts): Default options for sending transactions. 
 80                network (SolanaNetwork): Solana network
 81                program_id (PublicKey, optional): Program public key. Defaults to latest AVER_PROGRAM_ID specified in constants.py.
 82
 83            Raises:
 84                Exception: Invalid owner argument
 85
 86            Returns:
 87                AverClient: AverClient
 88            """
 89            if(isinstance(owner, Wallet)):
 90                wallet = owner
 91            elif(isinstance(owner, Keypair)):
 92                wallet = Wallet(owner)
 93            else:
 94                raise Exception('Invalid Owner Passed. Must be Wallet or Keypair')
 95
 96            provider = Provider(
 97                connection,
 98                wallet,
 99                opts
100            )
101            api_endpoint = get_aver_api_endpoint(network)
102            program = await AverClient.load_program(provider, program_id)
103
104            aver_client = AverClient(program, api_endpoint, network)    
105            return aver_client
106
107    @staticmethod
108    async def load_program(
109            provider: Provider,
110            program_id: PublicKey = AVER_PROGRAM_ID
111        ):
112        """
113        Loads Aver Program
114
115        Args:
116            provider (Provider): Provider
117            program_id (PublicKey, optional): Program public key. Defaults to AVER_PROGRAM_ID.
118
119        Raises:
120            Exception: Program IDL not loaded
121
122        Returns:
123            Program: Program
124        """
125        idl = await Program.fetch_idl(program_id,provider)
126        if(idl):
127            program = Program(idl, program_id, provider)
128            return program
129        else:
130            raise Exception('Program IDL not loaded')
131
132
133    async def close(self):
134        """
135        Closes connection.
136
137        Call this in your program's clean-up function(s)
138        """
139        await self.provider.close()
140
141    #TODO - Move to admin
142    async def check_health(self):
143        url = self.aver_api_endpoint + '/health?format=json'
144        aver_health_response = get(url)
145
146        solana_health_response = await self.provider.connection.get_version()
147
148        return {'api': aver_health_response.json(), 'solana': solana_health_response}
149
150
151    async def get_or_create_associated_token_account(
152            self,
153            owner: PublicKey,
154            payer: Keypair,
155            token_mint: PublicKey = None,
156        ):
157        """
158        Attempts to load an Associate Token Account (ATA) or creates one if not found.
159
160        Args:
161            owner (PublicKey): Public key of the owner of Associated Token Account
162            payer (Keypair): Payer of transaction fee and ATA rent (rent is recoverable)
163            token_mint (PublicKey, optional): ATA token mint public key. Defaults to USDC token according to chosen solana network.
164
165        Returns:
166            PublicKey: ATA PublicKey
167        """
168        token_mint = token_mint if token_mint is not None else self.quote_token
169        associated_token_account = get_associated_token_address(owner=owner, mint=token_mint)
170        response = await self.provider.connection.get_token_account_balance(associated_token_account)
171
172        if not 'result' in response.keys(): 
173            tx = Transaction()
174            tx.add(
175                create_associated_token_account(
176                    payer=payer.public_key,
177                    owner=owner,
178                    mint=token_mint,
179                ),
180            )
181            signers = [payer]
182            response = await self.provider.connection.send_transaction(tx, *signers, opts=self.provider.opts)
183            await self.provider.connection.confirm_transaction(response['result'])
184            return associated_token_account
185        else:
186            return associated_token_account
187
188    async def request_lamport_airdrop(
189            self,
190            amount: int,
191            owner: PublicKey
192        ):
193        """
194        Request an airdrop of lamports (SOL). This method is only supported on devnet.
195
196        Args:
197            amount (int): Lamports to airdrop. Note 1 lamport = 10^-9 SOL. Max of 1 SOL (10^9 lamports) applies.
198            owner (PublicKey): Public key of account to be airdropped
199
200        Returns:
201            RPCResponse: Response
202        """
203        return await self.provider.connection.request_airdrop(
204            pubkey=owner, 
205            lamports=amount, 
206            commitment=self.provider.opts
207            )
208    
209
210    async def request_ata_balance(self, ata: PublicKey):
211        """
212        Fetches the balance for an Associated Token Account (ATA). 
213        Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token.
214        For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.
215
216        Args:
217            ata (PublicKey): ATA public key
218
219        Raises:
220            Exception: Error from response
221
222        Returns:
223            int: Token balance
224        """
225        response = await self.provider.connection.get_token_account_balance(ata, self.provider.opts.preflight_commitment)
226        if 'error' in response:
227            raise Exception(response['error'])
228        return int(response['result']['value']['amount'])
229
230    async def request_token_balance(
231            self,
232            mint: PublicKey,
233            owner: PublicKey
234        ):
235        """
236        Fetches a wallet's token balance, given the wallet's owner and the token mint's public key.
237        Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token.
238        For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.
239
240        Args:
241            mint: The public key of the token mint
242            owner: The public key of the wallet
243
244        Returns:
245            int: Token balance
246
247        """
248        ata = get_associated_token_address(owner, mint)
249        return await self.request_ata_balance(ata)
250
251    async def request_lamport_balance(
252            self,
253            owner: PublicKey
254        ):
255        """
256        Fetches Lamport (SOL) balance for a given wallet
257        Note: the value returned is the integer representation of the balance, where a unit is one lamport (=10^-9 SOL)
258        For example, a value of 1,000,000,000 (lamports) = 1 SOL.
259
260        Args:
261            owner (PublicKey): Owner public key
262
263        Raises:
264            Exception: Error from response
265
266        Returns:
267            int: Lamport balance
268        """
269        response = await self.provider.connection.get_balance(owner, self.provider.opts.preflight_commitment)
270        if 'error' in response:
271            raise Exception(response['error'])
272        return response['result']['value']
273
274    async def get_system_clock_datetime(
275            self
276        ):
277        """
278        Loads current solana system datetime
279
280        Returns:
281            datetime: Current Solana Clock Datetime
282        """
283        raw_data = await self.connection.get_account_info(SYS_VAR_CLOCK)
284        data = raw_data["result"]["value"]["data"][0]
285        parsed_data = CLOCK_STRUCT.parse(base64.decodebytes(data.encode("ascii")))
286        return datetime.datetime.utcfromtimestamp(parsed_data['unix_timestamp'])

Aver Client Class

Use AverClient to interact with the Aver Program, Solana network and Aver API

AverClient( program: anchorpy.program.core.Program, aver_api_endpoint: str, solana_network: pyaver.enums.SolanaNetwork) 
42    def __init__(
43            self, 
44            program: Program,
45            aver_api_endpoint: str,
46            solana_network: SolanaNetwork,
47        ):
48        """
49        Initialises AverClient object. Do not use this function; use AverClient.load() instead
50
51        Args:
52            program (Program): Aver program AnchorPy
53            aver_api_endpoint (str): Endpoint for Aver API (to be removed soon)
54            solana_network (SolanaNetwork): Solana network
55        """
56        self.connection = program.provider.connection
57        self.program = program
58        self.provider = program.provider
59        self.aver_api_endpoint = aver_api_endpoint
60        self.solana_network = solana_network
61        self.quote_token = get_quote_token(solana_network)
62        self.solana_client = Client(get_solana_endpoint(solana_network))
63        self.owner = program.provider.wallet.payer

Initialises AverClient object. Do not use this function; use AverClient.load() instead

Args
  • program (Program): Aver program AnchorPy
  • aver_api_endpoint (str): Endpoint for Aver API (to be removed soon)
  • solana_network (SolanaNetwork): Solana network
connection: solana.rpc.async_api.AsyncClient

Solana AsyncClient

program: anchorpy.program.core.Program

AnchorPy Program

provider: anchorpy.provider.Provider

AnchorPy Provider

aver_api_endpoint: str

Endpoint is used to make requests to Aver

solana_network: pyaver.enums.SolanaNetwork

Devnet or Mainnet - must correspond to the network provided in the connection AsyncClient

quote_token: solana.publickey.PublicKey

The token mint of the default quote token for markets on this network (i.e. USDC)

solana_client: solana.rpc.api.Client

Solana Client

owner: solana.keypair.Keypair

The default payer for transactions on-chain, unless one is specified

@staticmethod
async def load( connection: solana.rpc.async_api.AsyncClient, owner: solana.keypair.Keypair, opts: solana.rpc.types.TxOpts, network: pyaver.enums.SolanaNetwork, program_id: solana.publickey.PublicKey = 6q5ZGhEj6kkmEjuyCXuH4x8493bpi9fNzvy9L8hX83HQ) 
 65    @staticmethod
 66    async def load(
 67            connection: AsyncClient,
 68            owner: Keypair or Wallet, 
 69            opts: types.TxOpts,
 70            network: SolanaNetwork,
 71            program_id: PublicKey = AVER_PROGRAM_ID,
 72        ):
 73            """
 74            Initialises an AverClient object
 75
 76            Args:
 77                connection (AsyncClient): Solana AsyncClient object
 78                owner (KeypairorWallet): Default keypair to pay transaction costs (and rent costs) unless one is otherwise specified for a given transaction.
 79                opts (types.TxOpts): Default options for sending transactions. 
 80                network (SolanaNetwork): Solana network
 81                program_id (PublicKey, optional): Program public key. Defaults to latest AVER_PROGRAM_ID specified in constants.py.
 82
 83            Raises:
 84                Exception: Invalid owner argument
 85
 86            Returns:
 87                AverClient: AverClient
 88            """
 89            if(isinstance(owner, Wallet)):
 90                wallet = owner
 91            elif(isinstance(owner, Keypair)):
 92                wallet = Wallet(owner)
 93            else:
 94                raise Exception('Invalid Owner Passed. Must be Wallet or Keypair')
 95
 96            provider = Provider(
 97                connection,
 98                wallet,
 99                opts
100            )
101            api_endpoint = get_aver_api_endpoint(network)
102            program = await AverClient.load_program(provider, program_id)
103
104            aver_client = AverClient(program, api_endpoint, network)    
105            return aver_client

Initialises an AverClient object

Args
  • connection (AsyncClient): Solana AsyncClient object
  • owner (KeypairorWallet): Default keypair to pay transaction costs (and rent costs) unless one is otherwise specified for a given transaction.
  • opts (types.TxOpts): Default options for sending transactions.
  • network (SolanaNetwork): Solana network
  • program_id (PublicKey, optional): Program public key. Defaults to latest AVER_PROGRAM_ID specified in constants.py.
Raises
  • Exception: Invalid owner argument
Returns

AverClient: AverClient

@staticmethod
async def load_program( provider: anchorpy.provider.Provider, program_id: solana.publickey.PublicKey = 6q5ZGhEj6kkmEjuyCXuH4x8493bpi9fNzvy9L8hX83HQ) 
107    @staticmethod
108    async def load_program(
109            provider: Provider,
110            program_id: PublicKey = AVER_PROGRAM_ID
111        ):
112        """
113        Loads Aver Program
114
115        Args:
116            provider (Provider): Provider
117            program_id (PublicKey, optional): Program public key. Defaults to AVER_PROGRAM_ID.
118
119        Raises:
120            Exception: Program IDL not loaded
121
122        Returns:
123            Program: Program
124        """
125        idl = await Program.fetch_idl(program_id,provider)
126        if(idl):
127            program = Program(idl, program_id, provider)
128            return program
129        else:
130            raise Exception('Program IDL not loaded')

Loads Aver Program

Args
  • provider (Provider): Provider
  • program_id (PublicKey, optional): Program public key. Defaults to AVER_PROGRAM_ID.
Raises
  • Exception: Program IDL not loaded
Returns

Program: Program

async def close(self) 
133    async def close(self):
134        """
135        Closes connection.
136
137        Call this in your program's clean-up function(s)
138        """
139        await self.provider.close()

Closes connection.

Call this in your program's clean-up function(s)

async def check_health(self) 
142    async def check_health(self):
143        url = self.aver_api_endpoint + '/health?format=json'
144        aver_health_response = get(url)
145
146        solana_health_response = await self.provider.connection.get_version()
147
148        return {'api': aver_health_response.json(), 'solana': solana_health_response}
async def get_or_create_associated_token_account( self, owner: solana.publickey.PublicKey, payer: solana.keypair.Keypair, token_mint: solana.publickey.PublicKey = None) 
151    async def get_or_create_associated_token_account(
152            self,
153            owner: PublicKey,
154            payer: Keypair,
155            token_mint: PublicKey = None,
156        ):
157        """
158        Attempts to load an Associate Token Account (ATA) or creates one if not found.
159
160        Args:
161            owner (PublicKey): Public key of the owner of Associated Token Account
162            payer (Keypair): Payer of transaction fee and ATA rent (rent is recoverable)
163            token_mint (PublicKey, optional): ATA token mint public key. Defaults to USDC token according to chosen solana network.
164
165        Returns:
166            PublicKey: ATA PublicKey
167        """
168        token_mint = token_mint if token_mint is not None else self.quote_token
169        associated_token_account = get_associated_token_address(owner=owner, mint=token_mint)
170        response = await self.provider.connection.get_token_account_balance(associated_token_account)
171
172        if not 'result' in response.keys(): 
173            tx = Transaction()
174            tx.add(
175                create_associated_token_account(
176                    payer=payer.public_key,
177                    owner=owner,
178                    mint=token_mint,
179                ),
180            )
181            signers = [payer]
182            response = await self.provider.connection.send_transaction(tx, *signers, opts=self.provider.opts)
183            await self.provider.connection.confirm_transaction(response['result'])
184            return associated_token_account
185        else:
186            return associated_token_account

Attempts to load an Associate Token Account (ATA) or creates one if not found.

Args
  • owner (PublicKey): Public key of the owner of Associated Token Account
  • payer (Keypair): Payer of transaction fee and ATA rent (rent is recoverable)
  • token_mint (PublicKey, optional): ATA token mint public key. Defaults to USDC token according to chosen solana network.
Returns

PublicKey: ATA PublicKey

async def request_lamport_airdrop(self, amount: int, owner: solana.publickey.PublicKey) 
188    async def request_lamport_airdrop(
189            self,
190            amount: int,
191            owner: PublicKey
192        ):
193        """
194        Request an airdrop of lamports (SOL). This method is only supported on devnet.
195
196        Args:
197            amount (int): Lamports to airdrop. Note 1 lamport = 10^-9 SOL. Max of 1 SOL (10^9 lamports) applies.
198            owner (PublicKey): Public key of account to be airdropped
199
200        Returns:
201            RPCResponse: Response
202        """
203        return await self.provider.connection.request_airdrop(
204            pubkey=owner, 
205            lamports=amount, 
206            commitment=self.provider.opts
207            )

Request an airdrop of lamports (SOL). This method is only supported on devnet.

Args
  • amount (int): Lamports to airdrop. Note 1 lamport = 10^-9 SOL. Max of 1 SOL (10^9 lamports) applies.
  • owner (PublicKey): Public key of account to be airdropped
Returns

RPCResponse: Response

async def request_ata_balance(self, ata: solana.publickey.PublicKey) 
210    async def request_ata_balance(self, ata: PublicKey):
211        """
212        Fetches the balance for an Associated Token Account (ATA). 
213        Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token.
214        For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.
215
216        Args:
217            ata (PublicKey): ATA public key
218
219        Raises:
220            Exception: Error from response
221
222        Returns:
223            int: Token balance
224        """
225        response = await self.provider.connection.get_token_account_balance(ata, self.provider.opts.preflight_commitment)
226        if 'error' in response:
227            raise Exception(response['error'])
228        return int(response['result']['value']['amount'])

Fetches the balance for an Associated Token Account (ATA). Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token. For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.

Args
  • ata (PublicKey): ATA public key
Raises
  • Exception: Error from response
Returns

int: Token balance

async def request_token_balance( self, mint: solana.publickey.PublicKey, owner: solana.publickey.PublicKey) 
230    async def request_token_balance(
231            self,
232            mint: PublicKey,
233            owner: PublicKey
234        ):
235        """
236        Fetches a wallet's token balance, given the wallet's owner and the token mint's public key.
237        Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token.
238        For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.
239
240        Args:
241            mint: The public key of the token mint
242            owner: The public key of the wallet
243
244        Returns:
245            int: Token balance
246
247        """
248        ata = get_associated_token_address(owner, mint)
249        return await self.request_ata_balance(ata)

Fetches a wallet's token balance, given the wallet's owner and the token mint's public key. Note: the value returned is the integer representation of the balance, where a unit is the smallest possible increment of the token. For example, USDC is a 6 decimal place token, so a value of 1,000,000 here = 1 USDC.

Args
  • mint: The public key of the token mint
  • owner: The public key of the wallet
Returns

int: Token balance

async def request_lamport_balance(self, owner: solana.publickey.PublicKey) 
251    async def request_lamport_balance(
252            self,
253            owner: PublicKey
254        ):
255        """
256        Fetches Lamport (SOL) balance for a given wallet
257        Note: the value returned is the integer representation of the balance, where a unit is one lamport (=10^-9 SOL)
258        For example, a value of 1,000,000,000 (lamports) = 1 SOL.
259
260        Args:
261            owner (PublicKey): Owner public key
262
263        Raises:
264            Exception: Error from response
265
266        Returns:
267            int: Lamport balance
268        """
269        response = await self.provider.connection.get_balance(owner, self.provider.opts.preflight_commitment)
270        if 'error' in response:
271            raise Exception(response['error'])
272        return response['result']['value']

Fetches Lamport (SOL) balance for a given wallet Note: the value returned is the integer representation of the balance, where a unit is one lamport (=10^-9 SOL) For example, a value of 1,000,000,000 (lamports) = 1 SOL.

Args
  • owner (PublicKey): Owner public key
Raises
  • Exception: Error from response
Returns

int: Lamport balance

async def get_system_clock_datetime(self) 
274    async def get_system_clock_datetime(
275            self
276        ):
277        """
278        Loads current solana system datetime
279
280        Returns:
281            datetime: Current Solana Clock Datetime
282        """
283        raw_data = await self.connection.get_account_info(SYS_VAR_CLOCK)
284        data = raw_data["result"]["value"]["data"][0]
285        parsed_data = CLOCK_STRUCT.parse(base64.decodebytes(data.encode("ascii")))
286        return datetime.datetime.utcfromtimestamp(parsed_data['unix_timestamp'])

Loads current solana system datetime

Returns

datetime: Current Solana Clock Datetime