refactor: move bitcoin config the bitcoin repo

app/repositories/bitcoin.py

@@ -17,7 +17,8 @@ from app.models.bitcoind import (
     FeeEstimationMode,
     NetworkInfo,
 )
-from app.utils import SSE, bitcoin_config, bitcoin_rpc_async, send_sse_message
+from app.repositories.bitcoin_utils import bitcoin_config, bitcoin_rpc_async
+from app.utils import SSE, send_sse_message
 
 _initialized = False
 

app/repositories/bitcoin_utils.py

@@ -0,0 +1,89 @@
+import json
+from types import coroutine
+
+import aiohttp
+import requests
+from decouple import config
+from starlette import status
+
+from app.models.bitcoind import BlockRpcFunc
+
+
+class _BitcoinConfig:
+    def __init__(self) -> None:
+        self.network = config("network")
+        self.zmq_block_rpc = BlockRpcFunc.from_string(config("bitcoind_zmq_block_rpc"))
+
+        if self.network == "testnet":
+            self.ip = config("bitcoind_ip_testnet")
+            self.rpc_port = config("bitcoind_port_rpc_testnet")
+            self.zmq_port = config("bitcoind_zmq_block_port_testnet")
+        else:
+            self.ip = config("bitcoind_ip_mainnet")
+            self.rpc_port = config("bitcoind_port_rpc_mainnet")
+            self.zmq_port = config("bitcoind_zmq_block_port_mainnet")
+
+        self.rpc_url = f"http://{self.ip}:{self.rpc_port}"
+        self.zmq_url = f"tcp://{self.ip}:{self.zmq_port}"
+
+        self.username = config("bitcoind_user")
+        self.pw = config("bitcoind_pw")
+
+
+bitcoin_config = _BitcoinConfig()
+
+
+def bitcoin_rpc(method: str, params: list = []) -> requests.Response:
+    """Make an RPC request to the Bitcoin daemon
+
+    Connection parameters are read from the .env file.
+
+    Parameters
+    ----------
+    method : str
+        The method to call.
+    params : list, optional
+        Any parameters to include with the call
+    """
+    auth = (bitcoin_config.username, bitcoin_config.pw)
+    headers = {"Content-type": "text/plain"}
+    data = (
+        '{"jsonrpc": "2.0", "method": "'
+        + method
+        + '", "id":"0", "params":'
+        + json.dumps(params)
+        + "}"
+    )
+    return requests.post(bitcoin_config.rpc_url, auth=auth, headers=headers, data=data)
+
+
+async def bitcoin_rpc_async(method: str, params: list = []) -> coroutine:
+    auth = aiohttp.BasicAuth(bitcoin_config.username, bitcoin_config.pw)
+    headers = {"Content-type": "text/plain"}
+    data = (
+        '{"jsonrpc": "2.0", "method": "'
+        + method
+        + '", "id":"0", "params":'
+        + json.dumps(params)
+        + "}"
+    )
+
+    async with aiohttp.ClientSession(auth=auth, headers=headers) as session:
+        async with session.post(bitcoin_config.rpc_url, data=data) as resp:
+            if resp.status == status.HTTP_200_OK:
+                return await resp.json()
+            elif resp.status == status.HTTP_401_UNAUTHORIZED:
+                return {
+                    "error": "Access denied to Bitcoin Core RPC. Check if username and password is correct",
+                    "status": status.HTTP_403_FORBIDDEN,
+                }
+            elif resp.status == status.HTTP_403_FORBIDDEN:
+                return {
+                    "error": "Access denied to Bitcoin Core RPC. If this is a remote node, check if 'network.rpcallowip=0.0.0.0/0' is set.",
+                    "status": status.HTTP_403_FORBIDDEN,
+                }
+            else:
+                return {
+                    "error": f"Unknown answer from Bitcoin Core. Reason: {resp.reason}",
+                    "status": resp.status,
+                }

app/repositories/ln_impl/cln_grpc.py

@@ -32,7 +32,7 @@ from app.models.lightning import (
     TxStatus,
     WalletBalance,
 )
-from app.utils import bitcoin_rpc_async
+from app.repositories.bitcoin_utils import bitcoin_rpc_async
 from app.utils import lightning_config as lncfg
 from app.utils import next_push_id
 

app/routers/bitcoin.py

@@ -11,8 +11,8 @@ from app.repositories.bitcoin import (
     get_network_info,
     handle_block_sub,
 )
+from app.repositories.bitcoin_utils import bitcoin_rpc
 from app.routers.bitcoin_docs import blocks_sub_doc, estimate_fee_mode_desc
-from app.utils import bitcoin_rpc
 
 _PREFIX = "bitcoin"
 

app/utils.py

@@ -2,20 +2,15 @@ import array
 import asyncio
 import json
 import logging
-import os
 import random
 import re
 import time
-from types import coroutine
 from typing import Dict
 
-import aiohttp
 import grpc
-import requests
 from decouple import config
 from fastapi.encoders import jsonable_encoder
 from fastapi_plugins import redis_plugin
-from starlette import status
 
 node_type = config("ln_node")
 if node_type == "none":
@@ -32,33 +27,6 @@ else:
     raise ValueError(f"Unknown node type: {node_type}")
 
 
-from app.models.bitcoind import BlockRpcFunc
-
-
-class BitcoinConfig:
-    def __init__(self) -> None:
-        self.network = config("network")
-        self.zmq_block_rpc = BlockRpcFunc.from_string(config("bitcoind_zmq_block_rpc"))
-
-        if self.network == "testnet":
-            self.ip = config("bitcoind_ip_testnet")
-            self.rpc_port = config("bitcoind_port_rpc_testnet")
-            self.zmq_port = config("bitcoind_zmq_block_port_testnet")
-        else:
-            self.ip = config("bitcoind_ip_mainnet")
-            self.rpc_port = config("bitcoind_port_rpc_mainnet")
-            self.zmq_port = config("bitcoind_zmq_block_port_mainnet")
-
-        self.rpc_url = f"http://{self.ip}:{self.rpc_port}"
-        self.zmq_url = f"tcp://{self.ip}:{self.zmq_port}"
-
-        self.username = config("bitcoind_user")
-        self.pw = config("bitcoind_pw")
-
-
-bitcoin_config = BitcoinConfig()
-
-
 class LightningConfig:
     def __init__(self) -> None:
         self.network = config("network")
@@ -96,62 +64,6 @@ class LightningConfig:
 lightning_config = LightningConfig()
 
 
-def bitcoin_rpc(method: str, params: list = []) -> requests.Response:
-    """Make an RPC request to the Bitcoin daemon
-
-    Connection parameters are read from the .env file.
-
-    Parameters
-    ----------
-    method : str
-        The method to call.
-    params : list, optional
-        Any parameters to include with the call
-    """
-    auth = (bitcoin_config.username, bitcoin_config.pw)
-    headers = {"Content-type": "text/plain"}
-    data = (
-        '{"jsonrpc": "2.0", "method": "'
-        + method
-        + '", "id":"0", "params":'
-        + json.dumps(params)
-        + "}"
-    )
-    return requests.post(bitcoin_config.rpc_url, auth=auth, headers=headers, data=data)
-
-
-async def bitcoin_rpc_async(method: str, params: list = []) -> coroutine:
-    auth = aiohttp.BasicAuth(bitcoin_config.username, bitcoin_config.pw)
-    headers = {"Content-type": "text/plain"}
-    data = (
-        '{"jsonrpc": "2.0", "method": "'
-        + method
-        + '", "id":"0", "params":'
-        + json.dumps(params)
-        + "}"
-    )
-
-    async with aiohttp.ClientSession(auth=auth, headers=headers) as session:
-        async with session.post(bitcoin_config.rpc_url, data=data) as resp:
-            if resp.status == status.HTTP_200_OK:
-                return await resp.json()
-            elif resp.status == status.HTTP_401_UNAUTHORIZED:
-                return {
-                    "error": "Access denied to Bitcoin Core RPC. Check if username and password is correct",
-                    "status": status.HTTP_403_FORBIDDEN,
-                }
-            elif resp.status == status.HTTP_403_FORBIDDEN:
-                return {
-                    "error": "Access denied to Bitcoin Core RPC. If this is a remote node, check if 'network.rpcallowip=0.0.0.0/0' is set.",
-                    "status": status.HTTP_403_FORBIDDEN,
-                }
-            else:
-                return {
-                    "error": f"Unknown answer from Bitcoin Core. Reason: {resp.reason}",
-                    "status": resp.status,
-                }
-
-
 async def send_sse_message(id: str, json_data: Dict):
     """Send a message to any SSE connections