Switch DogecoinClient to use requests library

Replace urllib with requests library for DogecoinClient HTTP calls:
- Consistent with MoneroClient implementation
- Cleaner code using requests.post() with Basic auth
- Removed unused _headers method and imports
- Both crypto clients now use the same HTTP library

This provides better reliability and consistency across all RPC clients.
This commit is contained in:
Russell Ballestrini 2025-09-23 08:53:26 -04:00
parent 4693fa495f
commit 64d7a3b12a

View file

@ -2,9 +2,6 @@ from typing import Any, Dict, List, Optional, Tuple
import json
import time
import urllib.request
import urllib.error
import base64
# Try to import requests for better Digest auth support
try:
@ -33,10 +30,6 @@ class MoneroClient:
self.rpc_pass = rpc_pass
self.timeout = timeout
def _headers(self) -> Dict[str, str]:
# MoneroClient now uses Digest auth in _call method
# This method is only used for no-auth requests
return {"Content-Type": "application/json"}
def _call(self, method: str, params: Optional[Dict[str, Any]] = None) -> Any:
payload = {
@ -172,13 +165,6 @@ class DogecoinClient:
self.rpc_pass = rpc_pass
self.timeout = timeout
def _headers(self) -> Dict[str, str]:
headers = {"Content-Type": "application/json"}
if self.rpc_user and self.rpc_pass:
auth = f"{self.rpc_user}:{self.rpc_pass}".encode()
headers["Authorization"] = "Basic " + base64.b64encode(auth).decode()
return headers
def _call(self, method: str, params: Optional[List[Any]] = None) -> Any:
"""Note: Dogecoin RPC expects params as array, not dict"""
payload = {
@ -189,20 +175,28 @@ class DogecoinClient:
if params is not None:
payload["params"] = params
data = json.dumps(payload).encode()
req = urllib.request.Request(self.rpc_url, data=data, headers=self._headers())
# Use requests library for all HTTP calls - consistent with MoneroClient
if not HAS_REQUESTS:
raise RuntimeError("requests library required for Dogecoin RPC. Install with: pip install requests")
auth = (self.rpc_user, self.rpc_pass) if self.rpc_user and self.rpc_pass else None
try:
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
body = resp.read()
obj = json.loads(body)
if "error" in obj and obj["error"]:
raise RuntimeError(f"RPC error: {obj['error']}")
return obj.get("result")
except urllib.error.HTTPError as e:
raise RuntimeError(f"HTTP error {e.code}: {e.reason}")
except Exception as e:
raise RuntimeError(f"RPC call failed: {e}")
response = requests.post(
self.rpc_url,
json=payload,
auth=auth, # Basic auth for Dogecoin
timeout=self.timeout
)
response.raise_for_status()
obj = response.json()
if "error" in obj and obj["error"]:
raise RuntimeError(f"RPC error: {obj['error']}")
return obj.get("result")
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Dogecoin RPC connection error: {e}")
# Wallet Management