my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""Helius RPC and WebSocket Client for Solana"""
#2
#3	import httpx
#4	import json
#5	import asyncio
#6	import websockets
#7	from typing import Optional, Any, Callable, Awaitable
#8	from dataclasses import dataclass
#9
#10
#11	@dataclass
#12	class TokenBalance:
#13	"""Token balance info"""
#14	mint: str
#15	amount: int
#16	decimals: int
#17	ui_amount: float
#18
#19
#20	@dataclass
#21	class AccountInfo:
#22	"""Account info response"""
#23	lamports: int
#24	owner: str
#25	data: Any
#26	executable: bool
#27	rent_epoch: int
#28
#29
#30	class HeliusClient:
#31	"""Client for Helius RPC and DAS API"""
#32
#33	def __init__(
#34	self,
#35	api_key: str,
#36	rpc_url: Optional[str] = None,
#37	wss_url: Optional[str] = None,
#38	):
#39	"""
#40	Initialize Helius client.
#41
#42	Args:
#43	api_key: Helius API key
#44	rpc_url: Helius RPC URL (constructed from api_key if not provided)
#45	wss_url: Helius WebSocket URL (constructed from api_key if not provided)
#46	"""
#47	self.api_key = api_key
#48	self.rpc_url = rpc_url or f"https://mainnet.helius-rpc.com/?api-key={api_key}"
#49	self.wss_url = wss_url or f"wss://mainnet.helius-rpc.com/?api-key={api_key}"
#50	self.das_url = f"https://mainnet.helius-rpc.com/?api-key={api_key}"
#51
#52	self._client = httpx.AsyncClient(timeout=30.0)
#53	self._ws = None
#54	self._ws_callbacks: dict[int, Callable] = {}
#55	self._subscription_id = 0
#56
#57	async def _rpc_call(self, method: str, params: list = None) -> Any:
#58	"""Make an RPC call to Helius"""
#59	payload = {
#60	"jsonrpc": "2.0",
#61	"id": 1,
#62	"method": method,
#63	"params": params or [],
#64	}
#65
#66	response = await self._client.post(self.rpc_url, json=payload)
#67	response.raise_for_status()
#68	data = response.json()
#69
#70	if "error" in data:
#71	raise Exception(f"RPC error: {data['error']}")
#72
#73	return data.get("result")
#74
#75	# =====================
#76	# Account Methods
#77	# =====================
#78
#79	async def get_balance(self, pubkey: str) -> int:
#80	"""
#81	Get SOL balance for an account.
#82
#83	Args:
#84	pubkey: Account public key
#85
#86	Returns:
#87	Balance in lamports
#88	"""
#89	result = await self._rpc_call("getBalance", [pubkey])
#90	return result.get("value", 0)
#91
#92	async def get_sol_balance(self, pubkey: str) -> float:
#93	"""
#94	Get SOL balance in SOL (not lamports).
#95
#96	Args:
#97	pubkey: Account public key
#98
#99	Returns:
#100	Balance in SOL
#101	"""
#102	lamports = await self.get_balance(pubkey)
#103	return lamports / 1_000_000_000
#104
#105	async def get_account_info(self, pubkey: str, encoding: str = "base64") -> Optional[AccountInfo]:
#106	"""
#107	Get account info.
#108
#109	Args:
#110	pubkey: Account public key
#111	encoding: Data encoding (base64, base58, jsonParsed)
#112
#113	Returns:
#114	AccountInfo or None if account doesn't exist
#115	"""
#116	result = await self._rpc_call("getAccountInfo", [pubkey, {"encoding": encoding}])
#117
#118	if result.get("value") is None:
#119	return None
#120
#121	value = result["value"]
#122	return AccountInfo(
#123	lamports=value["lamports"],
#124	owner=value["owner"],
#125	data=value["data"],
#126	executable=value["executable"],
#127	rent_epoch=value["rentEpoch"],
#128	)
#129
#130	async def get_token_accounts_by_owner(
#131	self,
#132	owner: str,
#133	program_id: str = "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
#134	) -> list[TokenBalance]:
#135	"""
#136	Get all token accounts for an owner.
#137
#138	Args:
#139	owner: Owner public key
#140	program_id: Token program ID (SPL Token or Token-2022)
#141
#142	Returns:
#143	List of token balances
#144	"""
#145	result = await self._rpc_call(
#146	"getTokenAccountsByOwner",
#147	[
#148	owner,
#149	{"programId": program_id},
#150	{"encoding": "jsonParsed"}
#151	]
#152	)
#153
#154	balances = []
#155	for account in result.get("value", []):
#156	info = account["account"]["data"]["parsed"]["info"]
#157	token_amount = info["tokenAmount"]
#158	balances.append(TokenBalance(
#159	mint=info["mint"],
#160	amount=int(token_amount["amount"]),
#161	decimals=token_amount["decimals"],
#162	ui_amount=float(token_amount["uiAmount"] or 0),
#163	))
#164
#165	return balances
#166
#167	async def get_token_balance(self, token_account: str) -> TokenBalance:
#168	"""
#169	Get balance of a specific token account.
#170
#171	Args:
#172	token_account: Token account address
#173
#174	Returns:
#175	TokenBalance
#176	"""
#177	result = await self._rpc_call("getTokenAccountBalance", [token_account])
#178	value = result["value"]
#179
#180	return TokenBalance(
#181	mint="", # Not returned by this endpoint
#182	amount=int(value["amount"]),
#183	decimals=value["decimals"],
#184	ui_amount=float(value["uiAmount"] or 0),
#185	)
#186
#187	# =====================
#188	# Transaction Methods
#189	# =====================
#190
#191	async def get_transaction(self, signature: str, encoding: str = "jsonParsed") -> Optional[dict]:
#192	"""
#193	Get transaction details.
#194
#195	Args:
#196	signature: Transaction signature
#197	encoding: Response encoding
#198
#199	Returns:
#200	Transaction details or None
#201	"""
#202	result = await self._rpc_call(
#203	"getTransaction",
#204	[signature, {"encoding": encoding, "maxSupportedTransactionVersion": 0}]
#205	)
#206	return result
#207
#208	async def get_signatures_for_address(
#209	self,
#210	address: str,
#211	limit: int = 10,
#212	before: Optional[str] = None,
#213	) -> list[dict]:
#214	"""
#215	Get recent transaction signatures for an address.
#216
#217	Args:
#218	address: Account address
#219	limit: Max signatures to return
#220	before: Get signatures before this signature
#221
#222	Returns:
#223	List of signature info
#224	"""
#225	params = {"limit": limit}
#226	if before:
#227	params["before"] = before
#228
#229	result = await self._rpc_call("getSignaturesForAddress", [address, params])
#230	return result or []
#231
#232	async def send_raw_transaction(
#233	self,
#234	tx_base64: str,
#235	skip_preflight: bool = False,
#236	) -> str:
#237	"""
#238	Send a raw transaction.
#239
#240	Args:
#241	tx_base64: Base64 encoded transaction
#242	skip_preflight: Skip preflight simulation
#243
#244	Returns:
#245	Transaction signature
#246	"""
#247	result = await self._rpc_call(
#248	"sendTransaction",
#249	[tx_base64, {"skipPreflight": skip_preflight, "encoding": "base64"}]
#250	)
#251	return result
#252
#253	async def get_latest_blockhash(self) -> dict:
#254	"""
#255	Get latest blockhash.
#256
#257	Returns:
#258	Dict with blockhash and lastValidBlockHeight
#259	"""
#260	result = await self._rpc_call("getLatestBlockhash")
#261	return result["value"]
#262
#263	async def simulate_transaction(
#264	self,
#265	tx_base64: str,
#266	sig_verify: bool = False,
#267	) -> dict:
#268	"""
#269	Simulate a transaction.
#270
#271	Args:
#272	tx_base64: Base64 encoded transaction
#273	sig_verify: Verify signatures
#274
#275	Returns:
#276	Simulation result
#277	"""
#278	result = await self._rpc_call(
#279	"simulateTransaction",
#280	[tx_base64, {"sigVerify": sig_verify, "encoding": "base64"}]
#281	)
#282	return result["value"]
#283
#284	# =====================
#285	# DAS API Methods
#286	# =====================
#287
#288	async def get_asset(self, asset_id: str) -> dict:
#289	"""
#290	Get asset info using DAS API.
#291
#292	Args:
#293	asset_id: Asset/token mint address
#294
#295	Returns:
#296	Asset metadata
#297	"""
#298	payload = {
#299	"jsonrpc": "2.0",
#300	"id": 1,
#301	"method": "getAsset",
#302	"params": {"id": asset_id},
#303	}
#304
#305	response = await self._client.post(self.das_url, json=payload)
#306	response.raise_for_status()
#307	data = response.json()
#308
#309	if "error" in data:
#310	raise Exception(f"DAS error: {data['error']}")
#311
#312	return data.get("result")
#313
#314	async def get_assets_by_owner(
#315	self,
#316	owner: str,
#317	page: int = 1,
#318	limit: int = 100,
#319	) -> dict:
#320	"""
#321	Get all assets owned by an address using DAS API.
#322
#323	Args:
#324	owner: Owner address
#325	page: Page number
#326	limit: Items per page
#327
#328	Returns:
#329	Assets list with metadata
#330	"""
#331	payload = {
#332	"jsonrpc": "2.0",
#333	"id": 1,
#334	"method": "getAssetsByOwner",
#335	"params": {
#336	"ownerAddress": owner,
#337	"page": page,
#338	"limit": limit,
#339	},
#340	}
#341
#342	response = await self._client.post(self.das_url, json=payload)
#343	response.raise_for_status()
#344	data = response.json()
#345
#346	if "error" in data:
#347	raise Exception(f"DAS error: {data['error']}")
#348
#349	return data.get("result")
#350
#351	async def search_assets(
#352	self,
#353	owner: Optional[str] = None,
#354	creator: Optional[str] = None,
#355	collection: Optional[str] = None,
#356	page: int = 1,
#357	limit: int = 100,
#358	) -> dict:
#359	"""
#360	Search assets using DAS API.
#361
#362	Args:
#363	owner: Filter by owner
#364	creator: Filter by creator
#365	collection: Filter by collection
#366	page: Page number
#367	limit: Items per page
#368
#369	Returns:
#370	Search results
#371	"""
#372	params = {"page": page, "limit": limit}
#373	if owner:
#374	params["ownerAddress"] = owner
#375	if creator:
#376	params["creatorAddress"] = creator
#377	if collection:
#378	params["grouping"] = ["collection", collection]
#379
#380	payload = {
#381	"jsonrpc": "2.0",
#382	"id": 1,
#383	"method": "searchAssets",
#384	"params": params,
#385	}
#386
#387	response = await self._client.post(self.das_url, json=payload)
#388	response.raise_for_status()
#389	data = response.json()
#390
#391	if "error" in data:
#392	raise Exception(f"DAS error: {data['error']}")
#393
#394	return data.get("result")
#395
#396	# =====================
#397	# WebSocket Methods
#398	# =====================
#399
#400	async def connect_websocket(self):
#401	"""Connect to Helius WebSocket"""
#402	if self._ws is None or self._ws.closed:
#403	self._ws = await websockets.connect(self.wss_url)
#404
#405	async def disconnect_websocket(self):
#406	"""Disconnect from WebSocket"""
#407	if self._ws and not self._ws.closed:
#408	await self._ws.close()
#409	self._ws = None
#410
#411	async def subscribe_account(
#412	self,
#413	pubkey: str,
#414	callback: Callable[[dict], Awaitable[None]],
#415	) -> int:
#416	"""
#417	Subscribe to account changes.
#418
#419	Args:
#420	pubkey: Account public key
#421	callback: Async callback for updates
#422
#423	Returns:
#424	Subscription ID
#425	"""
#426	await self.connect_websocket()
#427
#428	self._subscription_id += 1
#429	sub_id = self._subscription_id
#430
#431	payload = {
#432	"jsonrpc": "2.0",
#433	"id": sub_id,
#434	"method": "accountSubscribe",
#435	"params": [pubkey, {"encoding": "jsonParsed"}],
#436	}
#437
#438	await self._ws.send(json.dumps(payload))
#439	self._ws_callbacks[sub_id] = callback
#440
#441	return sub_id
#442
#443	async def subscribe_logs(
#444	self,
#445	filter_type: str, # "all" or "mentions"
#446	callback: Callable[[dict], Awaitable[None]],
#447	mentions: Optional[list[str]] = None,
#448	) -> int:
#449	"""
#450	Subscribe to transaction logs.
#451
#452	Args:
#453	filter_type: "all" or "mentions"
#454	callback: Async callback for log updates
#455	mentions: List of pubkeys to filter (if filter_type is "mentions")
#456
#457	Returns:
#458	Subscription ID
#459	"""
#460	await self.connect_websocket()
#461
#462	self._subscription_id += 1
#463	sub_id = self._subscription_id
#464
#465	if filter_type == "mentions" and mentions:
#466	filter_param = {"mentions": mentions}
#467	else:
#468	filter_param = filter_type
#469
#470	payload = {
#471	"jsonrpc": "2.0",
#472	"id": sub_id,
#473	"method": "logsSubscribe",
#474	"params": [filter_param],
#475	}
#476
#477	await self._ws.send(json.dumps(payload))
#478	self._ws_callbacks[sub_id] = callback
#479
#480	return sub_id
#481
#482	async def unsubscribe(self, subscription_id: int, method: str = "accountUnsubscribe"):
#483	"""
#484	Unsubscribe from a WebSocket subscription.
#485
#486	Args:
#487	subscription_id: Subscription ID to unsubscribe
#488	method: Unsubscribe method name
#489	"""
#490	if self._ws and not self._ws.closed:
#491	payload = {
#492	"jsonrpc": "2.0",
#493	"id": subscription_id,
#494	"method": method,
#495	"params": [subscription_id],
#496	}
#497	await self._ws.send(json.dumps(payload))
#498	self._ws_callbacks.pop(subscription_id, None)
#499
#500	async def listen_websocket(self):
#501	"""Listen for WebSocket messages and dispatch to callbacks"""
#502	if self._ws is None:
#503	raise Exception("WebSocket not connected")
#504
#505	async for message in self._ws:
#506	data = json.loads(message)
#507
#508	# Handle subscription confirmations
#509	if "result" in data and isinstance(data["result"], int):
#510	continue
#511
#512	# Handle subscription updates
#513	if "params" in data:
#514	sub_id = data["params"].get("subscription")
#515	if sub_id in self._ws_callbacks:
#516	await self._ws_callbacks[sub_id](data["params"]["result"])
#517
#518	# =====================
#519	# Utility Methods
#520	# =====================
#521
#522	async def get_slot(self) -> int:
#523	"""Get current slot"""
#524	return await self._rpc_call("getSlot")
#525
#526	async def get_block_height(self) -> int:
#527	"""Get current block height"""
#528	return await self._rpc_call("getBlockHeight")
#529
#530	async def get_health(self) -> str:
#531	"""Get node health status"""
#532	return await self._rpc_call("getHealth")
#533
#534	async def get_minimum_balance_for_rent_exemption(self, data_length: int) -> int:
#535	"""Get minimum balance for rent exemption"""
#536	return await self._rpc_call("getMinimumBalanceForRentExemption", [data_length])
#537
#538	async def close(self):
#539	"""Close all connections"""
#540	await self.disconnect_websocket()
#541	await self._client.aclose()
#542
#543	async def __aenter__(self):
#544	return self
#545
#546	async def __aexit__(self, exc_type, exc_val, exc_tb):
#547	await self.close()
#548

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public