RPC Providers
solana.rpc.providers
special
RPC Providers.
async_base
Async base RPC Provider.
AsyncBaseProvider
Base class for async RPC providers to implement.
Source code in solana/rpc/providers/async_base.py
class AsyncBaseProvider:
"""Base class for async RPC providers to implement."""
async def make_request(self, body: Body, parser: Type[T]) -> T:
"""Make a request ot the rpc endpoint."""
raise NotImplementedError("Providers must implement this method")
make_request(self, body, parser)
async
Make a request ot the rpc endpoint.
Source code in solana/rpc/providers/async_base.py
async def make_request(self, body: Body, parser: Type[T]) -> T:
"""Make a request ot the rpc endpoint."""
raise NotImplementedError("Providers must implement this method")
async_http
Async HTTP RPC Provider.
AsyncHTTPProvider
Async HTTP provider to interact with the http rpc endpoint.
Source code in solana/rpc/providers/async_http.py
class AsyncHTTPProvider(AsyncBaseProvider, _HTTPProviderCore):
"""Async HTTP provider to interact with the http rpc endpoint."""
def __init__(
self,
endpoint: Optional[str] = None,
extra_headers: Optional[Dict[str, str]] = None,
timeout: float = DEFAULT_TIMEOUT,
):
"""Init AsyncHTTPProvider."""
super().__init__(endpoint, extra_headers)
self.session = httpx.AsyncClient(timeout=timeout)
def __str__(self) -> str:
"""String definition for HTTPProvider."""
return f"Async HTTP RPC connection {self.endpoint_uri}"
@handle_async_exceptions(SolanaRpcException, httpx.HTTPError)
async def make_request(self, body: Body, parser: Type[T]) -> T:
"""Make an async HTTP request to an http rpc endpoint."""
raw = await self.make_request_unparsed(body)
return _parse_raw(raw, parser=parser)
async def make_request_unparsed(self, body: Body) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_request(body=body)
raw_response = await self.session.post(**request_kwargs)
return _after_request_unparsed(raw_response)
async def make_batch_request_unparsed(self, reqs: Tuple[Body, ...]) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_batch_request(reqs)
raw_response = await self.session.post(**request_kwargs)
return _after_request_unparsed(raw_response)
@overload
async def make_batch_request(self, reqs: _BodiesTup, parsers: _Tup) -> _RespTup: ...
@overload
async def make_batch_request(self, reqs: _BodiesTup1, parsers: _Tup1) -> _RespTup1: ...
@overload
async def make_batch_request(self, reqs: _BodiesTup2, parsers: _Tup2) -> _RespTup2: ...
@overload
async def make_batch_request(self, reqs: _BodiesTup3, parsers: _Tup3) -> _RespTup3: ...
@overload
async def make_batch_request(self, reqs: _BodiesTup4, parsers: _Tup4) -> _RespTup4: ...
@overload
async def make_batch_request(self, reqs: _BodiesTup5, parsers: _Tup5) -> _RespTup5: ...
async def make_batch_request(self, reqs: Tuple[Body, ...], parsers: _Tuples) -> Tuple[RPCResult, ...]:
"""Make an async HTTP batch request to an http rpc endpoint.
Args:
reqs: A tuple of request objects from ``solders.rpc.requests``.
parsers: A tuple of response classes from ``solders.rpc.responses``.
Note: ``parsers`` should line up with ``reqs``.
Example:
>>> from solana.rpc.providers.async_http import AsyncHTTPProvider
>>> from solders.rpc.requests import GetBlockHeight, GetFirstAvailableBlock
>>> from solders.rpc.responses import GetBlockHeightResp, GetFirstAvailableBlockResp
>>> provider = AsyncHTTPProvider("https://api.devnet.solana.com")
>>> reqs = (GetBlockHeight(), GetFirstAvailableBlock())
>>> parsers = (GetBlockHeightResp, GetFirstAvailableBlockResp)
>>> await provider.make_batch_request(reqs, parsers) # doctest: +SKIP
(GetBlockHeightResp(
158613909,
), GetFirstAvailableBlockResp(
86753592,
))
"""
raw = await self.make_batch_request_unparsed(reqs)
return _parse_raw_batch(raw, parsers)
async def __aenter__(self) -> "AsyncHTTPProvider":
"""Use as a context manager."""
await self.session.__aenter__()
return self
async def __aexit__(self, _exc_type, _exc, _tb):
"""Exits the context manager."""
await self.close()
async def close(self) -> None:
"""Close session."""
await self.session.aclose()
__init__(self, endpoint=None, extra_headers=None, timeout=10)
special
Init AsyncHTTPProvider.
Source code in solana/rpc/providers/async_http.py
def __init__(
self,
endpoint: Optional[str] = None,
extra_headers: Optional[Dict[str, str]] = None,
timeout: float = DEFAULT_TIMEOUT,
):
"""Init AsyncHTTPProvider."""
super().__init__(endpoint, extra_headers)
self.session = httpx.AsyncClient(timeout=timeout)
close(self)
async
Close session.
Source code in solana/rpc/providers/async_http.py
async def close(self) -> None:
"""Close session."""
await self.session.aclose()
make_batch_request(self, reqs, parsers)
async
Make an async HTTP batch request to an http rpc endpoint.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
reqs |
Tuple[Union[solders.rpc.requests.GetAccountInfo, solders.rpc.requests.GetBalance, solders.rpc.requests.GetBlock, solders.rpc.requests.GetBlockHeight, solders.rpc.requests.GetBlockProduction, solders.rpc.requests.GetBlockCommitment, solders.rpc.requests.GetBlocks, solders.rpc.requests.GetBlocksWithLimit, solders.rpc.requests.GetBlockTime, solders.rpc.requests.GetClusterNodes, solders.rpc.requests.GetEpochInfo, solders.rpc.requests.GetEpochSchedule, solders.rpc.requests.GetFeeForMessage, solders.rpc.requests.GetFirstAvailableBlock, solders.rpc.requests.GetGenesisHash, solders.rpc.requests.GetHealth, solders.rpc.requests.GetHighestSnapshotSlot, solders.rpc.requests.GetIdentity, solders.rpc.requests.GetInflationGovernor, solders.rpc.requests.GetInflationRate, solders.rpc.requests.GetInflationReward, solders.rpc.requests.GetLargestAccounts, solders.rpc.requests.GetLatestBlockhash, solders.rpc.requests.GetLeaderSchedule, solders.rpc.requests.GetMaxRetransmitSlot, solders.rpc.requests.GetMaxShredInsertSlot, solders.rpc.requests.GetMinimumBalanceForRentExemption, solders.rpc.requests.GetMultipleAccounts, solders.rpc.requests.GetProgramAccounts, solders.rpc.requests.GetRecentPerformanceSamples, solders.rpc.requests.GetSignaturesForAddress, solders.rpc.requests.GetSignatureStatuses, solders.rpc.requests.GetSlot, solders.rpc.requests.GetSlotLeader, solders.rpc.requests.GetSlotLeaders, solders.rpc.requests.GetStakeActivation, solders.rpc.requests.GetSupply, solders.rpc.requests.GetTokenAccountBalance, solders.rpc.requests.GetTokenAccountsByDelegate, solders.rpc.requests.GetTokenAccountsByOwner, solders.rpc.requests.GetTokenLargestAccounts, solders.rpc.requests.GetTokenSupply, solders.rpc.requests.GetTransaction, solders.rpc.requests.GetTransactionCount, solders.rpc.requests.GetVersion, solders.rpc.requests.GetVoteAccounts, solders.rpc.requests.IsBlockhashValid, solders.rpc.requests.MinimumLedgerSlot, solders.rpc.requests.RequestAirdrop, solders.rpc.requests.SendRawTransaction, solders.rpc.requests.SendLegacyTransaction, solders.rpc.requests.ValidatorExit, solders.rpc.requests.AccountSubscribe, solders.rpc.requests.BlockSubscribe, solders.rpc.requests.LogsSubscribe, solders.rpc.requests.ProgramSubscribe, solders.rpc.requests.SignatureSubscribe, solders.rpc.requests.SlotSubscribe, solders.rpc.requests.SlotsUpdatesSubscribe, solders.rpc.requests.RootSubscribe, solders.rpc.requests.VoteSubscribe, solders.rpc.requests.AccountUnsubscribe, solders.rpc.requests.BlockUnsubscribe, solders.rpc.requests.LogsUnsubscribe, solders.rpc.requests.ProgramUnsubscribe, solders.rpc.requests.SignatureUnsubscribe, solders.rpc.requests.SimulateLegacyTransaction, solders.rpc.requests.SlotUnsubscribe, solders.rpc.requests.SlotsUpdatesUnsubscribe, solders.rpc.requests.RootUnsubscribe, solders.rpc.requests.VoteUnsubscribe], ...] |
A tuple of request objects from |
required |
parsers |
Union[Tuple[Type[~T]], Tuple[Type[~T], Type[~_T1]], Tuple[Type[~T], Type[~_T1], Type[~_T2]], Tuple[Type[~T], Type[~_T1], Type[~_T2], Type[~_T3]], Tuple[Type[~T], Type[~_T1], Type[~_T2], Type[~_T3], Type[~_T4]], Tuple[Type[~T], Type[~_T1], Type[~_T2], Type[~_T3], Type[~_T4], Type[~_T5]]] |
A tuple of response classes from |
required |
Examples:
>>> from solana.rpc.providers.async_http import AsyncHTTPProvider
>>> from solders.rpc.requests import GetBlockHeight, GetFirstAvailableBlock
>>> from solders.rpc.responses import GetBlockHeightResp, GetFirstAvailableBlockResp
>>> provider = AsyncHTTPProvider("https://api.devnet.solana.com")
>>> reqs = (GetBlockHeight(), GetFirstAvailableBlock())
>>> parsers = (GetBlockHeightResp, GetFirstAvailableBlockResp)
>>> await provider.make_batch_request(reqs, parsers)
(GetBlockHeightResp(
158613909,
), GetFirstAvailableBlockResp(
86753592,
))
Source code in solana/rpc/providers/async_http.py
async def make_batch_request(self, reqs: Tuple[Body, ...], parsers: _Tuples) -> Tuple[RPCResult, ...]:
"""Make an async HTTP batch request to an http rpc endpoint.
Args:
reqs: A tuple of request objects from ``solders.rpc.requests``.
parsers: A tuple of response classes from ``solders.rpc.responses``.
Note: ``parsers`` should line up with ``reqs``.
Example:
>>> from solana.rpc.providers.async_http import AsyncHTTPProvider
>>> from solders.rpc.requests import GetBlockHeight, GetFirstAvailableBlock
>>> from solders.rpc.responses import GetBlockHeightResp, GetFirstAvailableBlockResp
>>> provider = AsyncHTTPProvider("https://api.devnet.solana.com")
>>> reqs = (GetBlockHeight(), GetFirstAvailableBlock())
>>> parsers = (GetBlockHeightResp, GetFirstAvailableBlockResp)
>>> await provider.make_batch_request(reqs, parsers) # doctest: +SKIP
(GetBlockHeightResp(
158613909,
), GetFirstAvailableBlockResp(
86753592,
))
"""
raw = await self.make_batch_request_unparsed(reqs)
return _parse_raw_batch(raw, parsers)
make_batch_request_unparsed(self, reqs)
async
Make an async HTTP request to an http rpc endpoint.
Source code in solana/rpc/providers/async_http.py
async def make_batch_request_unparsed(self, reqs: Tuple[Body, ...]) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_batch_request(reqs)
raw_response = await self.session.post(**request_kwargs)
return _after_request_unparsed(raw_response)
make_request_unparsed(self, body)
async
Make an async HTTP request to an http rpc endpoint.
Source code in solana/rpc/providers/async_http.py
async def make_request_unparsed(self, body: Body) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_request(body=body)
raw_response = await self.session.post(**request_kwargs)
return _after_request_unparsed(raw_response)
base
Base RPC Provider.
BaseProvider
Base class for RPC providers to implement.
Source code in solana/rpc/providers/base.py
class BaseProvider:
"""Base class for RPC providers to implement."""
def make_request(self, body: Body, parser: Type[T]) -> T:
"""Make a request to the rpc endpoint."""
raise NotImplementedError("Providers must implement this method")
make_request(self, body, parser)
Make a request to the rpc endpoint.
Source code in solana/rpc/providers/base.py
def make_request(self, body: Body, parser: Type[T]) -> T:
"""Make a request to the rpc endpoint."""
raise NotImplementedError("Providers must implement this method")
core
Helper code for HTTP provider classes.
get_default_endpoint()
Get the default http rpc endpoint.
Source code in solana/rpc/providers/core.py
def get_default_endpoint() -> URI:
"""Get the default http rpc endpoint."""
return URI(os.environ.get("SOLANARPC_HTTP_URI", "http://localhost:8899"))
http
HTTP RPC Provider.
HTTPProvider
HTTP provider to interact with the http rpc endpoint.
Source code in solana/rpc/providers/http.py
class HTTPProvider(BaseProvider, _HTTPProviderCore):
"""HTTP provider to interact with the http rpc endpoint."""
def __str__(self) -> str:
"""String definition for HTTPProvider."""
return f"HTTP RPC connection {self.endpoint_uri}"
@handle_exceptions(SolanaRpcException, httpx.HTTPError)
def make_request(self, body: Body, parser: Type[T]) -> T:
"""Make an HTTP request to an http rpc endpoint."""
raw = self.make_request_unparsed(body)
return _parse_raw(raw, parser=parser)
def make_request_unparsed(self, body: Body) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_request(body=body)
raw_response = httpx.post(**request_kwargs)
return _after_request_unparsed(raw_response)
def make_batch_request_unparsed(self, reqs: Tuple[Body, ...]) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_batch_request(reqs)
raw_response = httpx.post(**request_kwargs)
return _after_request_unparsed(raw_response)
@overload
def make_batch_request(self, reqs: _BodiesTup, parsers: _Tup) -> _RespTup: ...
@overload
def make_batch_request(self, reqs: _BodiesTup1, parsers: _Tup1) -> _RespTup1: ...
@overload
def make_batch_request(self, reqs: _BodiesTup2, parsers: _Tup2) -> _RespTup2: ...
@overload
def make_batch_request(self, reqs: _BodiesTup3, parsers: _Tup3) -> _RespTup3: ...
@overload
def make_batch_request(self, reqs: _BodiesTup4, parsers: _Tup4) -> _RespTup4: ...
@overload
def make_batch_request(self, reqs: _BodiesTup5, parsers: _Tup5) -> _RespTup5: ...
def make_batch_request(self, reqs: Tuple[Body, ...], parsers: _Tuples) -> Tuple[RPCResult, ...]:
"""Make a HTTP batch request to an http rpc endpoint.
Args:
reqs: A tuple of request objects from ``solders.rpc.requests``.
parsers: A tuple of response classes from ``solders.rpc.responses``.
Note: ``parsers`` should line up with ``reqs``.
Example:
>>> from solana.rpc.providers.http import HTTPProvider
>>> from solders.rpc.requests import GetBlockHeight, GetFirstAvailableBlock
>>> from solders.rpc.responses import GetBlockHeightResp, GetFirstAvailableBlockResp
>>> provider = HTTPProvider("https://api.devnet.solana.com")
>>> reqs = (GetBlockHeight(), GetFirstAvailableBlock())
>>> parsers = (GetBlockHeightResp, GetFirstAvailableBlockResp)
>>> provider.make_batch_request(reqs, parsers) # doctest: +SKIP
(GetBlockHeightResp(
158613909,
), GetFirstAvailableBlockResp(
86753592,
))
"""
raw = self.make_batch_request_unparsed(reqs)
return _parse_raw_batch(raw, parsers)
make_batch_request(self, reqs, parsers)
Make a HTTP batch request to an http rpc endpoint.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
reqs |
Tuple[Union[solders.rpc.requests.GetAccountInfo, solders.rpc.requests.GetBalance, solders.rpc.requests.GetBlock, solders.rpc.requests.GetBlockHeight, solders.rpc.requests.GetBlockProduction, solders.rpc.requests.GetBlockCommitment, solders.rpc.requests.GetBlocks, solders.rpc.requests.GetBlocksWithLimit, solders.rpc.requests.GetBlockTime, solders.rpc.requests.GetClusterNodes, solders.rpc.requests.GetEpochInfo, solders.rpc.requests.GetEpochSchedule, solders.rpc.requests.GetFeeForMessage, solders.rpc.requests.GetFirstAvailableBlock, solders.rpc.requests.GetGenesisHash, solders.rpc.requests.GetHealth, solders.rpc.requests.GetHighestSnapshotSlot, solders.rpc.requests.GetIdentity, solders.rpc.requests.GetInflationGovernor, solders.rpc.requests.GetInflationRate, solders.rpc.requests.GetInflationReward, solders.rpc.requests.GetLargestAccounts, solders.rpc.requests.GetLatestBlockhash, solders.rpc.requests.GetLeaderSchedule, solders.rpc.requests.GetMaxRetransmitSlot, solders.rpc.requests.GetMaxShredInsertSlot, solders.rpc.requests.GetMinimumBalanceForRentExemption, solders.rpc.requests.GetMultipleAccounts, solders.rpc.requests.GetProgramAccounts, solders.rpc.requests.GetRecentPerformanceSamples, solders.rpc.requests.GetSignaturesForAddress, solders.rpc.requests.GetSignatureStatuses, solders.rpc.requests.GetSlot, solders.rpc.requests.GetSlotLeader, solders.rpc.requests.GetSlotLeaders, solders.rpc.requests.GetStakeActivation, solders.rpc.requests.GetSupply, solders.rpc.requests.GetTokenAccountBalance, solders.rpc.requests.GetTokenAccountsByDelegate, solders.rpc.requests.GetTokenAccountsByOwner, solders.rpc.requests.GetTokenLargestAccounts, solders.rpc.requests.GetTokenSupply, solders.rpc.requests.GetTransaction, solders.rpc.requests.GetTransactionCount, solders.rpc.requests.GetVersion, solders.rpc.requests.GetVoteAccounts, solders.rpc.requests.IsBlockhashValid, solders.rpc.requests.MinimumLedgerSlot, solders.rpc.requests.RequestAirdrop, solders.rpc.requests.SendRawTransaction, solders.rpc.requests.SendLegacyTransaction, solders.rpc.requests.ValidatorExit, solders.rpc.requests.AccountSubscribe, solders.rpc.requests.BlockSubscribe, solders.rpc.requests.LogsSubscribe, solders.rpc.requests.ProgramSubscribe, solders.rpc.requests.SignatureSubscribe, solders.rpc.requests.SlotSubscribe, solders.rpc.requests.SlotsUpdatesSubscribe, solders.rpc.requests.RootSubscribe, solders.rpc.requests.VoteSubscribe, solders.rpc.requests.AccountUnsubscribe, solders.rpc.requests.BlockUnsubscribe, solders.rpc.requests.LogsUnsubscribe, solders.rpc.requests.ProgramUnsubscribe, solders.rpc.requests.SignatureUnsubscribe, solders.rpc.requests.SimulateLegacyTransaction, solders.rpc.requests.SlotUnsubscribe, solders.rpc.requests.SlotsUpdatesUnsubscribe, solders.rpc.requests.RootUnsubscribe, solders.rpc.requests.VoteUnsubscribe], ...] |
A tuple of request objects from |
required |
parsers |
Union[Tuple[Type[~T]], Tuple[Type[~T], Type[~_T1]], Tuple[Type[~T], Type[~_T1], Type[~_T2]], Tuple[Type[~T], Type[~_T1], Type[~_T2], Type[~_T3]], Tuple[Type[~T], Type[~_T1], Type[~_T2], Type[~_T3], Type[~_T4]], Tuple[Type[~T], Type[~_T1], Type[~_T2], Type[~_T3], Type[~_T4], Type[~_T5]]] |
A tuple of response classes from |
required |
Examples:
>>> from solana.rpc.providers.http import HTTPProvider
>>> from solders.rpc.requests import GetBlockHeight, GetFirstAvailableBlock
>>> from solders.rpc.responses import GetBlockHeightResp, GetFirstAvailableBlockResp
>>> provider = HTTPProvider("https://api.devnet.solana.com")
>>> reqs = (GetBlockHeight(), GetFirstAvailableBlock())
>>> parsers = (GetBlockHeightResp, GetFirstAvailableBlockResp)
>>> provider.make_batch_request(reqs, parsers)
(GetBlockHeightResp(
158613909,
), GetFirstAvailableBlockResp(
86753592,
))
Source code in solana/rpc/providers/http.py
def make_batch_request(self, reqs: Tuple[Body, ...], parsers: _Tuples) -> Tuple[RPCResult, ...]:
"""Make a HTTP batch request to an http rpc endpoint.
Args:
reqs: A tuple of request objects from ``solders.rpc.requests``.
parsers: A tuple of response classes from ``solders.rpc.responses``.
Note: ``parsers`` should line up with ``reqs``.
Example:
>>> from solana.rpc.providers.http import HTTPProvider
>>> from solders.rpc.requests import GetBlockHeight, GetFirstAvailableBlock
>>> from solders.rpc.responses import GetBlockHeightResp, GetFirstAvailableBlockResp
>>> provider = HTTPProvider("https://api.devnet.solana.com")
>>> reqs = (GetBlockHeight(), GetFirstAvailableBlock())
>>> parsers = (GetBlockHeightResp, GetFirstAvailableBlockResp)
>>> provider.make_batch_request(reqs, parsers) # doctest: +SKIP
(GetBlockHeightResp(
158613909,
), GetFirstAvailableBlockResp(
86753592,
))
"""
raw = self.make_batch_request_unparsed(reqs)
return _parse_raw_batch(raw, parsers)
make_batch_request_unparsed(self, reqs)
Make an async HTTP request to an http rpc endpoint.
Source code in solana/rpc/providers/http.py
def make_batch_request_unparsed(self, reqs: Tuple[Body, ...]) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_batch_request(reqs)
raw_response = httpx.post(**request_kwargs)
return _after_request_unparsed(raw_response)
make_request_unparsed(self, body)
Make an async HTTP request to an http rpc endpoint.
Source code in solana/rpc/providers/http.py
def make_request_unparsed(self, body: Body) -> str:
"""Make an async HTTP request to an http rpc endpoint."""
request_kwargs = self._before_request(body=body)
raw_response = httpx.post(**request_kwargs)
return _after_request_unparsed(raw_response)