Module `jmcore.mempool_api`

Mempool.space API client for Bitcoin blockchain queries.

Classes

class AddressInfo (**data: Any)

Expand source code

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()

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

pydantic.main.BaseModel

Class variables

var address : str: The type of the None singleton.
var chain_stats : AddressStats: The type of the None singleton.
var mempool_stats : AddressStats: The type of the None singleton.
var model_config: The type of the None singleton.

Methods

def balance(self) ‑> int

Expand source code

def balance(self) -> int:
    return self.total_received() - self.total_sent()

def total_received(self) ‑> int

Expand source code

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

def total_sent(self) ‑> int

Expand source code

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

class AddressStats (**data: Any)

Expand source code

class AddressStats(BaseModel):
    funded_txo_count: int
    funded_txo_sum: int
    spent_txo_count: int
    spent_txo_sum: int
    tx_count: int

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

pydantic.main.BaseModel

Class variables

var funded_txo_count : int: The type of the None singleton.
var funded_txo_sum : int: The type of the None singleton.
var model_config: The type of the None singleton.
var spent_txo_count : int: The type of the None singleton.
var spent_txo_sum : int: The type of the None singleton.
var tx_count : int: The type of the None singleton.

class MempoolAPI (base_url: str = 'https://mempool.space/api', timeout: float = 30.0, socks_proxy: str | None = None)

Expand source code

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}")
                # Try httpx-socks
                from httpx_socks import AsyncProxyTransport

                logger.debug("httpx_socks imported successfully")
                transport = AsyncProxyTransport.from_url(socks_proxy)
                logger.debug(f"Created SOCKS transport: {transport}")
                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

Methods

async def close(self) ‑> None

Expand source code

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

async def get_address_info(self, address: str) ‑> AddressInfo

Expand source code

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

async def get_block_hash(self, height: int) ‑> str

Expand source code

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_block_height(self) ‑> int

Expand source code

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_transaction(self, txid: str) ‑> Transaction

Expand source code

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

async def get_utxo_confirmations(self, txid: str, vout: int) ‑> int | None

Expand source code

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

Expand source code

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

async def test_connection(self) ‑> bool

Expand source code

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

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

class MempoolAPIError (*args, **kwargs)

Expand source code

class MempoolAPIError(Exception):
    pass

Common base class for all non-exit exceptions.

Ancestors

builtins.Exception
builtins.BaseException

class Transaction (**data: Any)

Expand source code

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

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

pydantic.main.BaseModel

Class variables

var fee : int: The type of the None singleton.
var locktime : int: The type of the None singleton.
var model_config: The type of the None singleton.
var size : int: The type of the None singleton.
var status : TxStatus: The type of the None singleton.
var txid : str: The type of the None singleton.
var version : int: The type of the None singleton.
var vin : list[dict[str, typing.Any]]: The type of the None singleton.
var vout : list[TxOut]: The type of the None singleton.
var weight : int: The type of the None singleton.

class TxOut (**data: Any)

Expand source code

class TxOut(BaseModel):
    scriptpubkey: str
    scriptpubkey_asm: str
    scriptpubkey_type: str
    scriptpubkey_address: str | None = None
    value: int

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

pydantic.main.BaseModel

Class variables

var model_config: The type of the None singleton.
var scriptpubkey : str: The type of the None singleton.
var scriptpubkey_address : str | None: The type of the None singleton.
var scriptpubkey_asm : str: The type of the None singleton.
var scriptpubkey_type : str: The type of the None singleton.
var value : int: The type of the None singleton.

class TxStatus (**data: Any)

Expand source code

class TxStatus(BaseModel):
    confirmed: bool
    block_height: int | None = None
    block_hash: str | None = None
    block_time: int | None = None

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

pydantic.main.BaseModel

Class variables

var block_hash : str | None: The type of the None singleton.
var block_height : int | None: The type of the None singleton.
var block_time : int | None: The type of the None singleton.
var confirmed : bool: The type of the None singleton.
var model_config: The type of the None singleton.