Skip to content

RPC Providers

solana.rpc.async_http_provider

Async HTTP RPC Provider.

AsyncHTTPProvider

Async HTTP provider to interact with the http rpc endpoint.

Source code in src/solana/rpc/async_http_provider.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class AsyncHTTPProvider:
    """Async HTTP provider to interact with the http rpc endpoint."""

    logger = logging.getLogger("solana.rpc.async_http_provider")

    def __init__(
        self,
        endpoint: str | None = None,
        extra_headers: dict[str, str] | None = None,
        timeout: float | None = None,
        proxy: str | None = None,
        rate_limit: float = 0,
        max_connections: int | None = None,
        max_keepalive_connections: int | None = None,
        keepalive_expiry: float | None = None,
        http2: bool = True,
        max_transport_retries: int = DEFAULT_MAX_TRANSPORT_RETRIES,
    ):
        """Init AsyncHTTPProvider.

        Args:
            endpoint: URL of the RPC endpoint.
            extra_headers: Extra headers to pass for HTTP request.
            timeout: HTTP request timeout in seconds. ``None`` uses the httpx2 default.
            proxy: Proxy URL to pass to the HTTP client.
            rate_limit: Maximum requests per second. ``0`` (default) disables rate limiting.
            max_connections: Maximum number of concurrent connections. ``None`` uses the httpx2 default.
            max_keepalive_connections: Maximum number of idle keep-alive connections. ``None`` uses the
                httpx2 default.
            keepalive_expiry: Idle keep-alive connection expiry in seconds. ``None`` uses the httpx2
                default.
            http2: Enable HTTP/2 support.
            max_transport_retries: Maximum number of times to retry httpx2 transport errors.
        """
        if max_transport_retries < 0:
            raise ValueError("max_transport_retries must be non-negative")
        self.endpoint_uri = get_default_endpoint() if not endpoint else URI(endpoint)
        self.timeout = timeout
        self.extra_headers = extra_headers
        self._max_transport_retries = max_transport_retries
        client_kwargs: dict[str, Any] = {"http2": http2}
        if timeout is not None:
            client_kwargs["timeout"] = httpx2.Timeout(timeout)
        if proxy is not None:
            client_kwargs["proxy"] = proxy
        if any(
            value is not None
            for value in (
                max_connections,
                max_keepalive_connections,
                keepalive_expiry,
            )
        ):
            client_kwargs["limits"] = httpx2.Limits(
                max_connections=max_connections,
                max_keepalive_connections=max_keepalive_connections,
                keepalive_expiry=keepalive_expiry,
            )
        self.session = httpx2.AsyncClient(**client_kwargs)
        self._limiter: AsyncLimiter | None = AsyncLimiter(rate_limit, time_period=1) if rate_limit > 0 else None

    def __str__(self) -> str:
        """String definition for HTTPProvider."""
        return f"Async HTTP RPC connection {self.endpoint_uri}"

    @handle_async_exceptions(SolanaRpcException, httpx2.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)

    @handle_async_exceptions(SolanaRpcException, httpx2.HTTPError)
    async def make_request_unparsed(self, body: JsonRpcRequestSerializer) -> str:
        """Make an async HTTP request to an http rpc endpoint."""
        headers = {"Content-Type": "application/json"}
        if self.extra_headers:
            headers.update(self.extra_headers)
        raw_response = await self._post_request(body.to_json(), headers)
        return _after_request_unparsed(raw_response)

    async def _post_request(self, content: str, headers: dict[str, str]) -> httpx2.Response:
        retries_left = self._max_transport_retries
        while True:
            try:
                return await self._post_request_once(content, headers)
            except _RETRYABLE_TRANSPORT_EXCEPTIONS as exc:
                if retries_left <= 0:
                    raise
                retries_left -= 1
                self.logger.debug(
                    "Retrying HTTP RPC request after %s: %s",
                    exc.__class__.__name__,
                    exc,
                )

    async def _post_request_once(self, content: str, headers: dict[str, str]) -> httpx2.Response:
        if self._limiter is not None:
            async with self._limiter:
                return await self.session.post(self.endpoint_uri, content=content, headers=headers)
        return await self.session.post(self.endpoint_uri, content=content, headers=headers)

    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__(endpoint=None, extra_headers=None, timeout=None, proxy=None, rate_limit=0, max_connections=None, max_keepalive_connections=None, keepalive_expiry=None, http2=True, max_transport_retries=DEFAULT_MAX_TRANSPORT_RETRIES)

Init AsyncHTTPProvider.

Parameters:

Name Type Description Default
endpoint str | None

URL of the RPC endpoint.

 None
extra_headers dict[str, str] | None

Extra headers to pass for HTTP request.

 None
timeout float | None

HTTP request timeout in seconds. None uses the httpx2 default.

 None
proxy str | None

Proxy URL to pass to the HTTP client.

 None
rate_limit float

Maximum requests per second. 0 (default) disables rate limiting.

 0
max_connections int | None

Maximum number of concurrent connections. None uses the httpx2 default.

 None
max_keepalive_connections int | None

Maximum number of idle keep-alive connections. None uses the httpx2 default.

 None
keepalive_expiry float | None

Idle keep-alive connection expiry in seconds. None uses the httpx2 default.

 None
http2 bool

Enable HTTP/2 support.

 True
max_transport_retries int

Maximum number of times to retry httpx2 transport errors.

 DEFAULT_MAX_TRANSPORT_RETRIES
Source code in src/solana/rpc/async_http_provider.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __init__(
    self,
    endpoint: str | None = None,
    extra_headers: dict[str, str] | None = None,
    timeout: float | None = None,
    proxy: str | None = None,
    rate_limit: float = 0,
    max_connections: int | None = None,
    max_keepalive_connections: int | None = None,
    keepalive_expiry: float | None = None,
    http2: bool = True,
    max_transport_retries: int = DEFAULT_MAX_TRANSPORT_RETRIES,
):
    """Init AsyncHTTPProvider.

    Args:
        endpoint: URL of the RPC endpoint.
        extra_headers: Extra headers to pass for HTTP request.
        timeout: HTTP request timeout in seconds. ``None`` uses the httpx2 default.
        proxy: Proxy URL to pass to the HTTP client.
        rate_limit: Maximum requests per second. ``0`` (default) disables rate limiting.
        max_connections: Maximum number of concurrent connections. ``None`` uses the httpx2 default.
        max_keepalive_connections: Maximum number of idle keep-alive connections. ``None`` uses the
            httpx2 default.
        keepalive_expiry: Idle keep-alive connection expiry in seconds. ``None`` uses the httpx2
            default.
        http2: Enable HTTP/2 support.
        max_transport_retries: Maximum number of times to retry httpx2 transport errors.
    """
    if max_transport_retries < 0:
        raise ValueError("max_transport_retries must be non-negative")
    self.endpoint_uri = get_default_endpoint() if not endpoint else URI(endpoint)
    self.timeout = timeout
    self.extra_headers = extra_headers
    self._max_transport_retries = max_transport_retries
    client_kwargs: dict[str, Any] = {"http2": http2}
    if timeout is not None:
        client_kwargs["timeout"] = httpx2.Timeout(timeout)
    if proxy is not None:
        client_kwargs["proxy"] = proxy
    if any(
        value is not None
        for value in (
            max_connections,
            max_keepalive_connections,
            keepalive_expiry,
        )
    ):
        client_kwargs["limits"] = httpx2.Limits(
            max_connections=max_connections,
            max_keepalive_connections=max_keepalive_connections,
            keepalive_expiry=keepalive_expiry,
        )
    self.session = httpx2.AsyncClient(**client_kwargs)
    self._limiter: AsyncLimiter | None = AsyncLimiter(rate_limit, time_period=1) if rate_limit > 0 else None

close() async

Close session.

Source code in src/solana/rpc/async_http_provider.py 
140
141
142
async def close(self) -> None:
    """Close session."""
    await self.session.aclose()

make_request(body, parser) async

Make an async HTTP request to an http rpc endpoint.

Source code in src/solana/rpc/async_http_provider.py 
95
96
97
98
99
@handle_async_exceptions(SolanaRpcException, httpx2.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)

make_request_unparsed(body) async

Make an async HTTP request to an http rpc endpoint.

Source code in src/solana/rpc/async_http_provider.py 
101
102
103
104
105
106
107
108
@handle_async_exceptions(SolanaRpcException, httpx2.HTTPError)
async def make_request_unparsed(self, body: JsonRpcRequestSerializer) -> str:
    """Make an async HTTP request to an http rpc endpoint."""
    headers = {"Content-Type": "application/json"}
    if self.extra_headers:
        headers.update(self.extra_headers)
    raw_response = await self._post_request(body.to_json(), headers)
    return _after_request_unparsed(raw_response)

get_default_endpoint()

Get the default http rpc endpoint.

Source code in src/solana/rpc/async_http_provider.py 
25
26
27
def get_default_endpoint() -> URI:
    """Get the default http rpc endpoint."""
    return URI(os.environ.get("SOLANARPC_HTTP_URI", "http://localhost:8899"))