afa26d0e29
Implement Solana RPC proxy with automatic failover and caching - Add multi-provider support for 5 free Solana RPC endpoints (Alchemy, PublicNode, Helius, QuickNode, Solana Public) - Implement automatic failover with 30-minute backoff for failed providers - Add disk-based response caching with 100GB LRU eviction - Create SQLite error logging with UUID tracking - Support both HTTP JSON-RPC and WebSocket connections - Include provider-specific authentication handling - Add response normalization for consistent output - Write end-to-end tests for core functionality The proxy provides a unified endpoint that automatically routes requests to available providers, caches responses to reduce load, and logs all errors with retrievable UUIDs for debugging.
113 lines
2.9 KiB
Python
113 lines
2.9 KiB
Python
from abc import ABC, abstractmethod
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Any, Optional
|
|
import os
|
|
|
|
|
|
class Provider(ABC):
|
|
def __init__(self, name: str):
|
|
self.name = name
|
|
self.backoff_until: Optional[datetime] = None
|
|
|
|
@property
|
|
@abstractmethod
|
|
def http_url(self) -> str:
|
|
pass
|
|
|
|
@property
|
|
@abstractmethod
|
|
def ws_url(self) -> str:
|
|
pass
|
|
|
|
def transform_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
|
|
return request
|
|
|
|
def transform_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
|
|
return response
|
|
|
|
def is_available(self) -> bool:
|
|
if self.backoff_until is None:
|
|
return True
|
|
return datetime.now() > self.backoff_until
|
|
|
|
def mark_failed(self, backoff_minutes: int = 30) -> None:
|
|
self.backoff_until = datetime.now() + timedelta(minutes=backoff_minutes)
|
|
|
|
|
|
class AlchemyProvider(Provider):
|
|
def __init__(self):
|
|
super().__init__("alchemy")
|
|
self.api_key = os.getenv("ALCHEMY_API_KEY", "")
|
|
|
|
@property
|
|
def http_url(self) -> str:
|
|
return f"https://solana-mainnet.g.alchemy.com/v2/{self.api_key}"
|
|
|
|
@property
|
|
def ws_url(self) -> str:
|
|
return f"wss://solana-mainnet.g.alchemy.com/v2/{self.api_key}"
|
|
|
|
|
|
class PublicNodeProvider(Provider):
|
|
def __init__(self):
|
|
super().__init__("publicnode")
|
|
|
|
@property
|
|
def http_url(self) -> str:
|
|
return "https://solana-rpc.publicnode.com"
|
|
|
|
@property
|
|
def ws_url(self) -> str:
|
|
return "wss://solana-rpc.publicnode.com"
|
|
|
|
|
|
class HeliusProvider(Provider):
|
|
def __init__(self):
|
|
super().__init__("helius")
|
|
self.api_key = os.getenv("HELIUS_API_KEY", "")
|
|
|
|
@property
|
|
def http_url(self) -> str:
|
|
return f"https://mainnet.helius-rpc.com/?api-key={self.api_key}"
|
|
|
|
@property
|
|
def ws_url(self) -> str:
|
|
return f"wss://mainnet.helius-rpc.com/?api-key={self.api_key}"
|
|
|
|
|
|
class QuickNodeProvider(Provider):
|
|
def __init__(self):
|
|
super().__init__("quicknode")
|
|
self.endpoint = os.getenv("QUICKNODE_ENDPOINT", "")
|
|
self.token = os.getenv("QUICKNODE_TOKEN", "")
|
|
|
|
@property
|
|
def http_url(self) -> str:
|
|
return f"https://{self.endpoint}/{self.token}/"
|
|
|
|
@property
|
|
def ws_url(self) -> str:
|
|
return f"wss://{self.endpoint}/{self.token}/"
|
|
|
|
|
|
class SolanaPublicProvider(Provider):
|
|
def __init__(self):
|
|
super().__init__("solana_public")
|
|
|
|
@property
|
|
def http_url(self) -> str:
|
|
return "https://api.mainnet-beta.solana.com"
|
|
|
|
@property
|
|
def ws_url(self) -> str:
|
|
return "wss://api.mainnet-beta.solana.com"
|
|
|
|
|
|
def create_providers() -> list[Provider]:
|
|
return [
|
|
AlchemyProvider(),
|
|
PublicNodeProvider(),
|
|
HeliusProvider(),
|
|
QuickNodeProvider(),
|
|
SolanaPublicProvider()
|
|
] |