0
# Providers
1
2
Connection providers for different Ethereum client interfaces supporting HTTP, WebSocket, IPC, and testing environments with automatic detection and persistent connection management.
3
4
## Capabilities
5
6
### HTTP Providers
7
8
JSON-RPC over HTTP connection providers for standard Ethereum node communication.
9
10
```python { .api }
11
class HTTPProvider(JSONBaseProvider):
12
def __init__(
13
self,
14
endpoint_uri: Optional[Union[URI, str]] = None,
15
request_kwargs: Optional[Dict[str, Any]] = None
16
):
17
"""
18
HTTP JSON-RPC provider.
19
20
Parameters:
21
- endpoint_uri: HTTP endpoint URL
22
- request_kwargs: Additional kwargs for requests
23
"""
24
25
class AsyncHTTPProvider(AsyncJSONBaseProvider):
26
def __init__(
27
self,
28
endpoint_uri: Optional[Union[URI, str]] = None,
29
request_kwargs: Optional[Dict[str, Any]] = None
30
):
31
"""
32
Async HTTP JSON-RPC provider.
33
34
Parameters:
35
- endpoint_uri: HTTP endpoint URL
36
- request_kwargs: Additional kwargs for aiohttp requests
37
"""
38
```
39
40
### WebSocket Providers
41
42
WebSocket connection providers supporting both legacy and persistent connection patterns.
43
44
```python { .api }
45
class WebSocketProvider(PersistentConnectionProvider):
46
def __init__(
47
self,
48
endpoint_uri: Optional[Union[URI, str]] = None,
49
websockets_kwargs: Optional[Dict[str, Any]] = None
50
):
51
"""
52
WebSocket provider with persistent connection.
53
54
Parameters:
55
- endpoint_uri: WebSocket endpoint URL
56
- websockets_kwargs: Additional kwargs for websockets connection
57
"""
58
59
class LegacyWebSocketProvider(JSONBaseProvider):
60
def __init__(
61
self,
62
endpoint_uri: Optional[Union[URI, str]] = None,
63
websockets_kwargs: Optional[Dict[str, Any]] = None
64
):
65
"""
66
Legacy WebSocket provider.
67
68
Parameters:
69
- endpoint_uri: WebSocket endpoint URL
70
- websockets_kwargs: Additional kwargs for websockets connection
71
"""
72
```
73
74
### IPC Providers
75
76
Inter-Process Communication providers for local Ethereum node connections.
77
78
```python { .api }
79
class IPCProvider(JSONBaseProvider):
80
def __init__(self, ipc_path: Optional[str] = None):
81
"""
82
IPC provider for Unix socket connections.
83
84
Parameters:
85
- ipc_path: Path to IPC socket file
86
"""
87
88
class AsyncIPCProvider(PersistentConnectionProvider):
89
def __init__(self, ipc_path: Optional[str] = None):
90
"""
91
Async IPC provider with persistent connection.
92
93
Parameters:
94
- ipc_path: Path to IPC socket file
95
"""
96
```
97
98
### Testing Providers
99
100
Providers for testing environments with in-memory blockchain simulation.
101
102
```python { .api }
103
class EthereumTesterProvider(BaseProvider):
104
def __init__(self, eth_tester: Optional[EthereumTester] = None):
105
"""
106
Provider for eth-tester backend.
107
108
Parameters:
109
- eth_tester: EthereumTester instance
110
"""
111
112
class AsyncEthereumTesterProvider(AsyncBaseProvider):
113
def __init__(self, eth_tester: Optional[EthereumTester] = None):
114
"""
115
Async provider for eth-tester backend.
116
117
Parameters:
118
- eth_tester: EthereumTester instance
119
"""
120
```
121
122
### Auto Provider
123
124
Automatic provider detection and configuration.
125
126
```python { .api }
127
class AutoProvider:
128
@staticmethod
129
def from_auto() -> BaseProvider:
130
"""
131
Automatically detect and configure provider.
132
133
Returns:
134
Configured provider based on environment
135
"""
136
```
137
138
### Base Provider Classes
139
140
Abstract base classes defining provider interfaces.
141
142
```python { .api }
143
class BaseProvider:
144
def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
145
"""Make JSON-RPC request."""
146
147
def is_connected(self) -> bool:
148
"""Check if provider is connected."""
149
150
class AsyncBaseProvider:
151
async def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
152
"""Make async JSON-RPC request."""
153
154
async def is_connected(self) -> bool:
155
"""Check if provider is connected."""
156
157
class JSONBaseProvider(BaseProvider):
158
def decode_rpc_response(self, response: bytes) -> RPCResponse:
159
"""Decode JSON-RPC response."""
160
161
def encode_rpc_request(self, method: RPCEndpoint, params: Any) -> bytes:
162
"""Encode JSON-RPC request."""
163
164
class PersistentConnectionProvider(AsyncBaseProvider):
165
def __init__(self, endpoint_uri: Optional[str] = None):
166
"""Base class for persistent connection providers."""
167
168
async def connect(self) -> None:
169
"""Establish persistent connection."""
170
171
async def disconnect(self) -> None:
172
"""Close persistent connection."""
173
```
174
175
### Connection Management
176
177
```python { .api }
178
class PersistentConnection:
179
def __init__(self, provider: PersistentConnectionProvider):
180
"""Manage persistent connection lifecycle."""
181
182
async def __aenter__(self) -> PersistentConnection:
183
"""Async context manager entry."""
184
185
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
186
"""Async context manager exit."""
187
```
188
189
## Usage Examples
190
191
### HTTP Provider
192
193
```python
194
from web3 import Web3
195
196
# Mainnet via Infura
197
w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/YOUR-PROJECT-ID'))
198
199
# Local node
200
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
201
202
# Custom request configuration
203
provider = Web3.HTTPProvider(
204
'https://mainnet.infura.io/v3/YOUR-PROJECT-ID',
205
request_kwargs={'timeout': 60}
206
)
207
w3 = Web3(provider)
208
```
209
210
### WebSocket Provider
211
212
```python
213
from web3 import Web3
214
215
# WebSocket connection with subscription support
216
w3 = Web3(Web3.WebSocketProvider('wss://mainnet.infura.io/ws/v3/YOUR-PROJECT-ID'))
217
218
# Local WebSocket
219
w3 = Web3(Web3.WebSocketProvider('ws://localhost:8546'))
220
```
221
222
### IPC Provider
223
224
```python
225
from web3 import Web3
226
227
# Default IPC path (varies by OS)
228
w3 = Web3(Web3.IPCProvider())
229
230
# Custom IPC path
231
w3 = Web3(Web3.IPCProvider('/path/to/geth.ipc'))
232
```
233
234
### Testing Provider
235
236
```python
237
from web3 import Web3
238
from eth_tester import EthereumTester
239
240
# Use eth-tester for testing
241
w3 = Web3(Web3.EthereumTesterProvider())
242
243
# Custom eth-tester configuration
244
eth_tester = EthereumTester()
245
w3 = Web3(Web3.EthereumTesterProvider(eth_tester))
246
```
247
248
### Auto Provider Detection
249
250
```python
251
from web3 import Web3
252
253
# Automatically detect provider
254
w3 = Web3(Web3.AutoProvider.from_auto())
255
```
256
257
### Async Providers
258
259
```python
260
import asyncio
261
from web3 import AsyncWeb3
262
263
async def main():
264
# Async HTTP
265
w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider('https://mainnet.infura.io/v3/YOUR-PROJECT-ID'))
266
267
# Async WebSocket with persistent connection
268
w3 = AsyncWeb3(AsyncWeb3.WebSocketProvider('wss://mainnet.infura.io/ws/v3/YOUR-PROJECT-ID'))
269
270
# Check connection
271
connected = await w3.is_connected()
272
print(f"Connected: {connected}")
273
274
asyncio.run(main())
275
```
276
277
### Persistent Connection Management
278
279
```python
280
import asyncio
281
from web3 import AsyncWeb3
282
283
async def main():
284
provider = AsyncWeb3.WebSocketProvider('wss://mainnet.infura.io/ws/v3/YOUR-PROJECT-ID')
285
w3 = AsyncWeb3(provider)
286
287
async with provider:
288
# Connection automatically managed
289
block = await w3.eth.get_block('latest')
290
print(f"Block number: {block.number}")
291
292
asyncio.run(main())
293
```