Skip to content

jmcore.mempool_api

jmcore.mempool_api

Mempool.space API client for Bitcoin blockchain queries.

Classes

AddressInfo

Bases: BaseModel

Source code in jmcore/src/jmcore/mempool_api.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
class AddressInfo(BaseModel):
    address: str
    chain_stats: AddressStats
    mempool_stats: AddressStats

    def total_received(self) -> int:
        return self.chain_stats.funded_txo_sum + self.mempool_stats.funded_txo_sum

    def total_sent(self) -> int:
        return self.chain_stats.spent_txo_sum + self.mempool_stats.spent_txo_sum

    def balance(self) -> int:
        return self.total_received() - self.total_sent()
Attributes
address: str instance-attribute
chain_stats: AddressStats instance-attribute
mempool_stats: AddressStats instance-attribute
Functions
balance() -> int
Source code in jmcore/src/jmcore/mempool_api.py 
33
34
def balance(self) -> int:
    return self.total_received() - self.total_sent()
total_received() -> int
Source code in jmcore/src/jmcore/mempool_api.py 
27
28
def total_received(self) -> int:
    return self.chain_stats.funded_txo_sum + self.mempool_stats.funded_txo_sum
total_sent() -> int
Source code in jmcore/src/jmcore/mempool_api.py 
30
31
def total_sent(self) -> int:
    return self.chain_stats.spent_txo_sum + self.mempool_stats.spent_txo_sum

AddressStats

Bases: BaseModel

Source code in jmcore/src/jmcore/mempool_api.py 
14
15
16
17
18
19
class AddressStats(BaseModel):
    funded_txo_count: int
    funded_txo_sum: int
    spent_txo_count: int
    spent_txo_sum: int
    tx_count: int
Attributes
funded_txo_count: int instance-attribute
funded_txo_sum: int instance-attribute
spent_txo_count: int instance-attribute
spent_txo_sum: int instance-attribute
tx_count: int instance-attribute

MempoolAPI

Source code in jmcore/src/jmcore/mempool_api.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class MempoolAPI:
    def __init__(
        self,
        base_url: str = "https://mempool.space/api",
        timeout: float = 30.0,
        socks_proxy: str | None = None,
    ):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.socks_proxy = socks_proxy

        client_kwargs: dict[str, Any] = {}
        if socks_proxy:
            try:
                logger.info(f"Attempting to configure SOCKS proxy: {socks_proxy}")
                from httpx_socks import AsyncProxyTransport

                from jmcore.tor_isolation import normalize_proxy_url

                # python-socks does not support the socks5h:// scheme directly.
                # normalize_proxy_url converts socks5h:// -> socks5:// + rdns=True
                # so that .onion addresses are resolved by Tor.
                normalized = normalize_proxy_url(socks_proxy)

                logger.debug("httpx_socks imported successfully")
                transport = AsyncProxyTransport.from_url(normalized.url, rdns=normalized.rdns)
                logger.debug(f"Created SOCKS transport: {transport} (rdns={normalized.rdns})")
                client_kwargs["transport"] = transport
                logger.info(f"MempoolAPI configured to use SOCKS proxy: {socks_proxy}")
            except ImportError as e:
                logger.error(f"httpx-socks not available: {e}")
                logger.warning("httpx-socks is required for SOCKS proxy support")
                logger.warning("Install with: pip install httpx-socks")
            except Exception as e:
                logger.error(f"Failed to configure SOCKS proxy {socks_proxy}: {e}")
                import traceback

                logger.debug(f"SOCKS proxy configuration traceback: {traceback.format_exc()}")

        self.client = httpx.AsyncClient(timeout=timeout, follow_redirects=True, **client_kwargs)

    async def __aenter__(self) -> MempoolAPI:
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.close()

    async def close(self) -> None:
        await self.client.aclose()

    async def test_connection(self) -> bool:
        """Test if the API connection works by making a simple request."""
        try:
            # Test with a lightweight endpoint - get current block tip height
            url = f"{self.base_url}/blocks/tip/height"
            logger.debug(f"Testing connection to: {url}")
            response = await self.client.get(url)
            response.raise_for_status()
            height = int(response.text)
            logger.info(f"Connection test successful - current block height: {height}")
            return True
        except Exception as e:
            logger.error(f"MempoolAPI connection test failed: {e}")
            return False

    async def _get(self, endpoint: str) -> dict[str, Any]:
        url = f"{self.base_url}/{endpoint}"
        try:
            logger.debug(f"MempoolAPI request: GET {url}")
            logger.debug(f"MempoolAPI using SOCKS proxy: {self.socks_proxy}")
            response = await self.client.get(url)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPError as e:
            logger.error(f"MempoolAPI error: {e}")
            logger.debug(
                f"MempoolAPI client transport: {getattr(self.client, '_transport', 'None')}"
            )
            raise MempoolAPIError(f"API request failed: {e}") from e

    async def get_address_info(self, address: str) -> AddressInfo:
        data = await self._get(f"address/{address}")
        return AddressInfo(**data)

    async def get_transaction(self, txid: str) -> Transaction:
        data = await self._get(f"tx/{txid}")
        return Transaction(**data)

    async def get_block_height(self) -> int:
        response = await self.client.get(f"{self.base_url}/blocks/tip/height")
        response.raise_for_status()
        return int(response.text)

    async def get_block_hash(self, height: int) -> str:
        response = await self.client.get(f"{self.base_url}/block-height/{height}")
        response.raise_for_status()
        return response.text

    async def get_utxo_confirmations(self, txid: str, vout: int) -> int | None:
        try:
            tx = await self.get_transaction(txid)
            if not tx.status.confirmed or tx.status.block_height is None:
                return None

            current_height = await self.get_block_height()
            confirmations = current_height - tx.status.block_height + 1
            return max(0, confirmations)
        except MempoolAPIError:
            return None

    async def get_utxo_value(self, txid: str, vout: int) -> int | None:
        try:
            tx = await self.get_transaction(txid)
            if vout >= len(tx.vout):
                return None
            return tx.vout[vout].value
        except MempoolAPIError:
            return None
Attributes
base_url = base_url.rstrip('/') instance-attribute
client = httpx.AsyncClient(timeout=timeout, follow_redirects=True, **client_kwargs) instance-attribute
socks_proxy = socks_proxy instance-attribute
timeout = timeout instance-attribute
Functions
__aenter__() -> MempoolAPI async
Source code in jmcore/src/jmcore/mempool_api.py 
109
110
async def __aenter__(self) -> MempoolAPI:
    return self
__aexit__(*args: Any) -> None async
Source code in jmcore/src/jmcore/mempool_api.py 
112
113
async def __aexit__(self, *args: Any) -> None:
    await self.close()
__init__(base_url: str = 'https://mempool.space/api', timeout: float = 30.0, socks_proxy: str | None = None)
Source code in jmcore/src/jmcore/mempool_api.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    base_url: str = "https://mempool.space/api",
    timeout: float = 30.0,
    socks_proxy: str | None = None,
):
    self.base_url = base_url.rstrip("/")
    self.timeout = timeout
    self.socks_proxy = socks_proxy

    client_kwargs: dict[str, Any] = {}
    if socks_proxy:
        try:
            logger.info(f"Attempting to configure SOCKS proxy: {socks_proxy}")
            from httpx_socks import AsyncProxyTransport

            from jmcore.tor_isolation import normalize_proxy_url

            # python-socks does not support the socks5h:// scheme directly.
            # normalize_proxy_url converts socks5h:// -> socks5:// + rdns=True
            # so that .onion addresses are resolved by Tor.
            normalized = normalize_proxy_url(socks_proxy)

            logger.debug("httpx_socks imported successfully")
            transport = AsyncProxyTransport.from_url(normalized.url, rdns=normalized.rdns)
            logger.debug(f"Created SOCKS transport: {transport} (rdns={normalized.rdns})")
            client_kwargs["transport"] = transport
            logger.info(f"MempoolAPI configured to use SOCKS proxy: {socks_proxy}")
        except ImportError as e:
            logger.error(f"httpx-socks not available: {e}")
            logger.warning("httpx-socks is required for SOCKS proxy support")
            logger.warning("Install with: pip install httpx-socks")
        except Exception as e:
            logger.error(f"Failed to configure SOCKS proxy {socks_proxy}: {e}")
            import traceback

            logger.debug(f"SOCKS proxy configuration traceback: {traceback.format_exc()}")

    self.client = httpx.AsyncClient(timeout=timeout, follow_redirects=True, **client_kwargs)
close() -> None async
Source code in jmcore/src/jmcore/mempool_api.py 
115
116
async def close(self) -> None:
    await self.client.aclose()
get_address_info(address: str) -> AddressInfo async
Source code in jmcore/src/jmcore/mempool_api.py 
148
149
150
async def get_address_info(self, address: str) -> AddressInfo:
    data = await self._get(f"address/{address}")
    return AddressInfo(**data)
get_block_hash(height: int) -> str async
Source code in jmcore/src/jmcore/mempool_api.py 
161
162
163
164
async def get_block_hash(self, height: int) -> str:
    response = await self.client.get(f"{self.base_url}/block-height/{height}")
    response.raise_for_status()
    return response.text
get_block_height() -> int async
Source code in jmcore/src/jmcore/mempool_api.py 
156
157
158
159
async def get_block_height(self) -> int:
    response = await self.client.get(f"{self.base_url}/blocks/tip/height")
    response.raise_for_status()
    return int(response.text)
get_transaction(txid: str) -> Transaction async
Source code in jmcore/src/jmcore/mempool_api.py 
152
153
154
async def get_transaction(self, txid: str) -> Transaction:
    data = await self._get(f"tx/{txid}")
    return Transaction(**data)
get_utxo_confirmations(txid: str, vout: int) -> int | None async
Source code in jmcore/src/jmcore/mempool_api.py 
166
167
168
169
170
171
172
173
174
175
176
async def get_utxo_confirmations(self, txid: str, vout: int) -> int | None:
    try:
        tx = await self.get_transaction(txid)
        if not tx.status.confirmed or tx.status.block_height is None:
            return None

        current_height = await self.get_block_height()
        confirmations = current_height - tx.status.block_height + 1
        return max(0, confirmations)
    except MempoolAPIError:
        return None
get_utxo_value(txid: str, vout: int) -> int | None async
Source code in jmcore/src/jmcore/mempool_api.py 
178
179
180
181
182
183
184
185
async def get_utxo_value(self, txid: str, vout: int) -> int | None:
    try:
        tx = await self.get_transaction(txid)
        if vout >= len(tx.vout):
            return None
        return tx.vout[vout].value
    except MempoolAPIError:
        return None
test_connection() -> bool async

Test if the API connection works by making a simple request.

Source code in jmcore/src/jmcore/mempool_api.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def test_connection(self) -> bool:
    """Test if the API connection works by making a simple request."""
    try:
        # Test with a lightweight endpoint - get current block tip height
        url = f"{self.base_url}/blocks/tip/height"
        logger.debug(f"Testing connection to: {url}")
        response = await self.client.get(url)
        response.raise_for_status()
        height = int(response.text)
        logger.info(f"Connection test successful - current block height: {height}")
        return True
    except Exception as e:
        logger.error(f"MempoolAPI connection test failed: {e}")
        return False

MempoolAPIError

Bases: Exception

Source code in jmcore/src/jmcore/mempool_api.py 
64
65
class MempoolAPIError(Exception):
    pass

Transaction

Bases: BaseModel

Source code in jmcore/src/jmcore/mempool_api.py 
52
53
54
55
56
57
58
59
60
61
class Transaction(BaseModel):
    txid: str
    version: int
    locktime: int
    size: int
    weight: int
    fee: int
    vin: list[dict[str, Any]]
    vout: list[TxOut]
    status: TxStatus
Attributes
fee: int instance-attribute
locktime: int instance-attribute
size: int instance-attribute
status: TxStatus instance-attribute
txid: str instance-attribute
version: int instance-attribute
vin: list[dict[str, Any]] instance-attribute
vout: list[TxOut] instance-attribute
weight: int instance-attribute

TxOut

Bases: BaseModel

Source code in jmcore/src/jmcore/mempool_api.py 
37
38
39
40
41
42
class TxOut(BaseModel):
    scriptpubkey: str
    scriptpubkey_asm: str
    scriptpubkey_type: str
    scriptpubkey_address: str | None = None
    value: int
Attributes
scriptpubkey: str instance-attribute
scriptpubkey_address: str | None = None class-attribute instance-attribute
scriptpubkey_asm: str instance-attribute
scriptpubkey_type: str instance-attribute
value: int instance-attribute

TxStatus

Bases: BaseModel

Source code in jmcore/src/jmcore/mempool_api.py 
45
46
47
48
49
class TxStatus(BaseModel):
    confirmed: bool
    block_height: int | None = None
    block_hash: str | None = None
    block_time: int | None = None
Attributes
block_hash: str | None = None class-attribute instance-attribute
block_height: int | None = None class-attribute instance-attribute
block_time: int | None = None class-attribute instance-attribute
confirmed: bool instance-attribute