new `HTTPClient` class

mmgen/http.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+#
+# MMGen Wallet, a terminal-based cryptocurrency wallet
+# Copyright (C)2013-2025 The MMGen Project <mmgen@tuta.io>
+# Licensed under the GNU General Public License, Version 3:
+#   https://www.gnu.org/licenses
+# Public project repositories:
+#   https://github.com/mmgen/mmgen-wallet
+#   https://gitlab.com/mmgen/mmgen-wallet
+
+"""
+http: HTTP client base class
+"""
+
+import requests
+
+class HTTPClient:
+
+	proto = 'https'
+	host = None
+	timeout = 60
+	http_hdrs = {
+		'User-Agent': 'curl/8.7.1',
+		'Proxy-Connection': 'Keep-Alive'}
+	extra_http_hdrs = {}
+	verify = True
+	text_mode = True
+
+	def __init__(self, cfg, *, proto=None, host=None):
+		self.cfg = cfg
+		if proto:
+			self.proto = proto
+		if host:
+			self.host = host
+		self.session = requests.Session()
+		self.session.trust_env = False # ignore *_PROXY environment vars
+		self.session.headers = (self.http_hdrs | self.extra_http_hdrs)
+		if cfg.proxy == 'env':
+			self.session.trust_env = True
+		elif cfg.proxy:
+			self.session.proxies.update({
+				'http':  f'socks5h://{cfg.proxy}',
+				'https': f'socks5h://{cfg.proxy}'
+			})
+
+	def call(self, name, path, err_fs, timeout, *, data=None):
+		url = self.proto + '://' + self.host + path
+		kwargs = {
+			'url': url,
+			'timeout': timeout or self.timeout,
+			'verify': self.verify}
+		if data:
+			kwargs['data'] = data
+		res = getattr(self.session, name)(**kwargs)
+		if res.status_code != 200:
+			from .util import die
+			die(2, '\n' + err_fs.format(s=res.status_code, u=url, d=data))
+		return res.content.decode() if self.text_mode else res.content
+
+	def get(self, *, path, timeout=None):
+		return self.call(
+			'get',
+			path,
+			'HTTP GET failed with status code {s}\n  URL: {u}',
+			timeout)
+
+	def post(self, *, path, data, timeout=None):
+		return self.call(
+			'post',
+			path,
+			'HTTP POST failed with status code {s}\n  URL: {u}\n  DATA: {d}',
+			timeout,
+			data = data)

mmgen/swap/proto/thorchain/thornode.py

@@ -16,6 +16,7 @@ import time, json
 from collections import namedtuple
 
 from ....amt import UniAmt
+from ....http import HTTPClient
 
 _gd = namedtuple('gas_unit_data', ['code', 'disp'])
 gas_unit_data = {
@@ -23,37 +24,14 @@ gas_unit_data = {
 	'gwei':        _gd('G', 'Gwei'),
 }
 
-class ThornodeRPCClient:
+class ThornodeRPCClient(HTTPClient):
 
 	http_hdrs = {'Content-Type': 'application/json'}
-	proto = 'https'
 	host = 'thornode.ninerealms.com'
-	verify = True
 	timeout = 5
 
 	def __init__(self, tx, *, proto=None, host=None):
-		self.cfg = tx.cfg
-		if proto:
-			self.proto = proto
-		if host:
-			self.host = host
-		import requests
-		self.session = requests.Session()
-		self.session.trust_env = False # ignore *_PROXY environment vars
-		self.session.headers = self.http_hdrs
-		if self.cfg.proxy == 'env':
-			self.session.trust_env = True
-		elif self.cfg.proxy:
-			self.session.proxies.update({
-				'http':  f'socks5h://{self.cfg.proxy}',
-				'https': f'socks5h://{self.cfg.proxy}'
-			})
-
-	def get(self, path, *, timeout=None):
-		return self.session.get(
-			url     = self.proto + '://' + self.host + path,
-			timeout = timeout or self.timeout,
-			verify  = self.verify)
+		super().__init__(tx.cfg, proto=proto, host=host)
 
 class Thornode:
 
@@ -71,7 +49,7 @@ class Thornode:
 				f'to_asset={recv}&'
 				f'amount={amt}&'
 				f'streaming_interval={swap_cfg.stream_interval}')
-			data = json.loads(self.rpc.get(get_str).content)
+			data = json.loads(self.rpc.get(path=get_str))
 			if not 'expiry' in data:
 				from ....util import pp_fmt, die
 				die(2, pp_fmt(data))

mmgen/tx/tx_proxy.py

@@ -14,50 +14,9 @@ tx.tx_proxy: tx proxy classes
 
 from ..color import green, pink, orange
 from ..util import msg, msg_r, die
+from ..http import HTTPClient
 
-class TxProxyClient:
-
-	proto = 'https'
-	verify = True
-	timeout = 60
-	http_hdrs = {
-		'User-Agent': 'curl/8.7.1',
-		'Proxy-Connection': 'Keep-Alive'}
-
-	def __init__(self, cfg):
-		self.cfg = cfg
-		import requests
-		self.session = requests.Session()
-		self.session.trust_env = False # ignore *_PROXY environment vars
-		self.session.headers = self.http_hdrs
-		if self.cfg.proxy == 'env':
-			self.session.trust_env = True
-		elif self.cfg.proxy:
-			self.session.proxies.update({
-				'http':  f'socks5h://{self.cfg.proxy}',
-				'https': f'socks5h://{self.cfg.proxy}'
-			})
-
-	def call(self, name, path, err_fs, timeout, *, data=None):
-		url = self.proto + '://' + self.host + path
-		kwargs = {
-			'url': url,
-			'timeout': timeout or self.timeout,
-			'verify': self.verify}
-		if data:
-			kwargs['data'] = data
-		res = getattr(self.session, name)(**kwargs)
-		if res.status_code != 200:
-			die(2, '\n' + err_fs.format(s=res.status_code, u=url, d=data))
-		return res.content.decode()
-
-	def get(self, *, path, timeout=None):
-		err_fs = 'HTTP Get failed with status code {s}\n  URL: {u}'
-		return self.call('get', path, err_fs, timeout)
-
-	def post(self, *, path, data, timeout=None):
-		err_fs = 'HTTP Post failed with status code {s}\n  URL: {u}\n  DATA: {d}'
-		return self.call('post', path, err_fs, timeout, data=data)
+class TxProxyClient(HTTPClient):
 
 	def get_form(self, timeout=None):
 		return self.get(path=self.form_path, timeout=timeout)