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")

    async def is_connected(self) -> bool:
        """Health check."""
        raise NotImplementedError("Providers must implement this method")

is_connected(self) async

Health check.

Source code in solana/rpc/providers/async_base.py

async def is_connected(self) -> bool:
    """Health check."""
    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 is_connected(self) -> bool:
        """Health check."""
        try:
            response = await self.session.get(self.health_uri)
            response.raise_for_status()
        except (IOError, httpx.HTTPError) as err:
            self.logger.error("Health check failed with error: %s", str(err))
            return False

        return response.status_code == httpx.codes.OK

    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()

is_connected(self) async

Health check.

Source code in solana/rpc/providers/async_http.py

async def is_connected(self) -> bool:
    """Health check."""
    try:
        response = await self.session.get(self.health_uri)
        response.raise_for_status()
    except (IOError, httpx.HTTPError) as err:
        self.logger.error("Health check failed with error: %s", str(err))
        return False

    return response.status_code == httpx.codes.OK

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 solders.rpc.requests.	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 solders.rpc.responses. Note: parsers should line up with reqs.	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")

    def is_connected(self) -> bool:
        """Health check."""
        raise NotImplementedError("Providers must implement this method")

is_connected(self)

Health check.

Source code in solana/rpc/providers/base.py

def is_connected(self) -> bool:
    """Health check."""
    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)

    def is_connected(self) -> bool:
        """Health check."""
        try:
            response = httpx.get(self.health_uri)
            response.raise_for_status()
        except (IOError, httpx.HTTPError) as err:
            self.logger.error("Health check failed with error: %s", str(err))
            return False

        return response.status_code == httpx.codes.OK

is_connected(self)

Health check.

Source code in solana/rpc/providers/http.py

def is_connected(self) -> bool:
    """Health check."""
    try:
        response = httpx.get(self.health_uri)
        response.raise_for_status()
    except (IOError, httpx.HTTPError) as err:
        self.logger.error("Health check failed with error: %s", str(err))
        return False

    return response.status_code == httpx.codes.OK

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 solders.rpc.requests.	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 solders.rpc.responses. Note: parsers should line up with reqs.	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)