From 64d7a3b12aa8a283bda4ab2bcf7f14b8e177cb13 Mon Sep 17 00:00:00 2001 From: Russell Ballestrini Date: Tue, 23 Sep 2025 08:53:26 -0400 Subject: [PATCH] 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. --- make_post_sell/lib/crypto_clients.py | 48 ++++++++++++---------------- 1 file changed, 21 insertions(+), 27 deletions(-) diff --git a/make_post_sell/lib/crypto_clients.py b/make_post_sell/lib/crypto_clients.py index 217e20b..755cecc 100644 --- a/make_post_sell/lib/crypto_clients.py +++ b/make_post_sell/lib/crypto_clients.py @@ -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