degenbot.provider ================= .. py:module:: degenbot.provider .. autoapi-nested-parse:: High-performance Ethereum RPC provider using Alloy. This module provides a Rust-based provider for fast log fetching and RPC calls. RPC methods return the ``LogData`` / ``BlockData`` mappings defined in :mod:`degenbot.types.rpc_types`: plain ``bytes`` for hash and data fields, EIP-55 checksummed strings for addresses, and Python ``int`` for numeric fields. Log dicts use camelCase keys; block/transaction dicts use snake_case. The provider surface is split by concurrency: the sync mixins and :class:`AlloyProvider` live in :mod:`degenbot.provider.sync`, the async twins in :mod:`degenbot.provider.async_provider`. .. rubric:: Example >>> from degenbot.provider import AlloyProvider, LogFilter >>> provider = AlloyProvider("https://eth-mainnet.example.com") >>> >>> # Direct property access >>> chain_id = provider.chain_id >>> block_number = provider.block_number >>> >>> # Log fetching with LogFilter >>> logs = provider.get_logs( ... LogFilter( ... from_block=18_000_000, ... to_block=18_010_000, ... addresses=["0x..."], ... ) ... ) >>> >>> # Or using keyword arguments >>> logs = provider.get_logs( ... from_block=18_000_000, ... to_block=18_010_000, ... addresses=["0x..."], ... ) Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/degenbot/provider/alloy_errors/index /autoapi/degenbot/provider/async_provider/index /autoapi/degenbot/provider/block_helpers/index /autoapi/degenbot/provider/call_helpers/index /autoapi/degenbot/provider/factory/index /autoapi/degenbot/provider/offline_provider/index /autoapi/degenbot/provider/sync/index Package Contents ---------------- .. py:data:: RustAlloyProvider .. py:data:: RustAsyncAlloyProvider .. py:class:: AsyncAlloyProvider(rust_provider: degenbot.provider._rust.RustAsyncAlloyProvider) Bases: :py:obj:`_AsyncAlloyEndpointMixin`, :py:obj:`_AsyncAlloyQueryMixin`, :py:obj:`_AsyncAlloyIntrospectionMixin` High-performance async Ethereum RPC provider using Alloy. A thin Python wrapper around the Rust ``AsyncAlloyProvider`` pyclass. Adds string block-identifier resolution (``'latest'``, ``'earliest'``, ``'pending'``) and exposes the inner Rust pyclass via :meth:`as_async_alloy` for Rust-side call seams. The public surface is assembled from the async query mixins. :param rust_provider: The underlying Rust ``AsyncAlloyProvider`` pyclass. Prefer :meth:`create` to construct an instance from an RPC URL. .. py:method:: create(rpc_url: str, max_retries: int = 10, max_blocks_per_request: int = 5000, *, requests_per_second: int | None = None, burst: int | None = None, chain_id: int | None = None) -> AsyncAlloyProvider :staticmethod: :async: Create an ``AsyncAlloyProvider`` asynchronously. :param rpc_url: HTTP/HTTPS/WS/IPC endpoint URL. :param max_retries: Maximum retry attempts. :param max_blocks_per_request: Maximum blocks per log request. :param requests_per_second: Optional rate limit. :param burst: Optional burst size for rate limiting. :param chain_id: Chain to bind the endpoint to; the Rust core raises :class:`ValueError` when the endpoint serves another chain. :returns: An ``AsyncAlloyProvider`` instance. .. py:exception:: ChainIdentityMismatchError Bases: :py:obj:`degenbot.exceptions.base.DegenbotValueError`, :py:obj:`ValueError` The core's chain-binding refusal, re-homed into this package's hierarchy. The check is the core's invariant; only the error class is re-homed here. It stays a :class:`ValueError` as well as a :class:`~degenbot.exceptions.base.DegenbotValueError`, so a caller that already handled a misconfigured endpoint keeps catching it exactly as it caught the core's own refusal. .. py:function:: get_async_provider_from_config(*, chain_id: int | str | None = None, node: str | None = None, resolve_uri: collections.abc.Callable[..., str] = resolve_http_rpc_uri, provider_factory: collections.abc.Callable[..., Any] | None = None, chain_mismatch_error: type[degenbot.provider.ChainMismatchError] = ChainMismatchError) -> degenbot.provider.async_provider.AsyncAlloyProvider :async: Build a chain-bound :class:`AsyncAlloyProvider` for the session chain. Async counterpart of :func:`get_provider_from_config`: the same resolution, the same core check, awaited on the caller's event loop. The DI seams are the sync factory's; an omitted ``provider_factory`` resolves to ``AsyncAlloyProvider.create``. :param chain_id: The explicit chain override; resolved from the config layers when absent. :param node: The explicit endpoint override, classified by its own value. :param resolve_uri: The endpoint resolver called as ``resolve_uri(session_chain_id, node=node)``. :param provider_factory: The awaited provider constructor called as ``provider_factory(endpoint, chain_id=session_chain_id)``. :param chain_mismatch_error: The exception class the core's refusal is caught as before re-homing. :returns: A chain-bound AsyncAlloyProvider over the resolved RPC endpoint. :raises ChainIdentityMismatchError: When the endpoint serves another chain than the session targets. It is both a :class:`~degenbot.exceptions.base.DegenbotValueError` and a :class:`ValueError`. .. py:function:: get_provider_from_config(*, chain_id: int | str | None = None, node: str | None = None, resolve_uri: collections.abc.Callable[..., str] = resolve_http_rpc_uri, provider_factory: collections.abc.Callable[..., degenbot.provider.sync.AlloyProvider] | None = None, chain_mismatch_error: type[degenbot.provider.ChainMismatchError] = ChainMismatchError) -> degenbot.provider.sync.AlloyProvider Build a chain-bound :class:`AlloyProvider` for the session chain. The endpoint resolves through the request scope of the four-layer cascade and the chain through the chain-id cascade, both from the installed typed config; ``chain_id``/``node`` are the explicit override layer when given. The constructed provider is BOUND to that chain: the Rust core verifies the endpoint's ``eth_chainId`` once, and its refusal is translated into a :class:`DegenbotValueError` here. ``resolve_uri``/``provider_factory``/``chain_mismatch_error`` are the DI seams (tests inject a recording resolver, a stand-in constructor, and a constructible stand-in for the PyO3 refusal class); omitted kwargs keep the production bindings. :param chain_id: The explicit chain override; resolved from the config layers when absent. :param node: The explicit endpoint override, classified by its own value. :param resolve_uri: The endpoint resolver called as ``resolve_uri(session_chain_id, node=node)``. :param provider_factory: The provider constructor called as ``provider_factory(endpoint, chain_id=session_chain_id)``. :param chain_mismatch_error: The exception class the core's refusal is caught as before re-homing. :returns: A chain-bound AlloyProvider over the resolved RPC endpoint. :raises ChainIdentityMismatchError: When the endpoint serves another chain than the session targets. It is both a :class:`~degenbot.exceptions.base.DegenbotValueError` and a :class:`ValueError`. .. py:class:: OfflineProvider(chain_id: int, blocks: dict[str, dict[str, Any]]) Bases: :py:obj:`_OfflineDataMixin`, :py:obj:`_OfflineLifecycleMixin` Ethereum provider that serves pre-recorded chain data. Delegates every data-serving RPC (``call``, ``get_code``, ``get_block``, ``get_block_number``, ``get_block_timestamp``) to a real Rust :class:`AlloyProvider` built over an in-memory offline transport, so it is indistinguishable from a live provider to any consumer (including :class:`BotIo`). Recorded-JSON metadata (``chain_id``, ``block_numbers``) is parsed in Python for the convenience accessors; the public surface is assembled from the mixins. .. attribute:: chain_id The chain ID this provider serves data for .. attribute:: blocks Dictionary of recorded block data keyed by block number string .. rubric:: Example >>> offline = OfflineProvider( ... chain_id=1, ... blocks={ ... "24945700": { ... "timestamp": 1776984059, ... "calls": {"0x...:0x...": "0x..."}, ... "code": {"0x...": "0x..."}, ... } ... }, ... ) >>> result = offline.call(to="0x...", data=b"...", block=24945700) .. py:method:: from_json_file(path: pathlib.Path) -> OfflineProvider :classmethod: Load recorded data from a JSON file. Supports both old multi-block format (with "blocks" key) and new single-block format (with "block_number" key). :param path: Path to the JSON file containing recorded data :returns: An OfflineProvider instance loaded from the file. .. py:class:: AlloyProvider(rpc_url: str, max_retries: int = 10, max_blocks_per_request: int = 5000, chain_id: int | None = None) Bases: :py:obj:`_AlloyEndpointMixin`, :py:obj:`_AlloyQueryMixin`, :py:obj:`_AlloyIntrospectionMixin` High-performance Ethereum RPC provider using Alloy. Backs log fetching and basic RPC calls with a Rust HTTP client using connection pooling for optimal performance. The public surface is assembled from the query mixins; this class owns construction and teardown of the wrapped Rust provider. :param rpc_url: HTTP/HTTPS endpoint URL :param max_retries: Maximum retry attempts (default: 10) :param max_blocks_per_request: Maximum logs per request (default: 5000) :param chain_id: Chain to bind the endpoint to. The Rust core reads ``eth_chainId`` once at construction and raises :class:`ValueError` when the endpoint serves another chain; ``None`` constructs the provider with no binding. .. rubric:: Example >>> provider = AlloyProvider("https://eth-mainnet.example.com") >>> >>> # Properties >>> chain_id = provider.chain_id >>> block_number = provider.block_number >>> >>> # Methods >>> block = provider.get_block(18_000_000) >>> logs = provider.get_logs(from_block=18_000_000, to_block=18_010_000) >>> code = provider.get_code("0x...") >>> result = provider.call("0x...", calldata) .. py:class:: LogFilter Filter criteria for log fetching. :param from_block: Starting block number (inclusive) :param to_block: Ending block number (inclusive) :param addresses: Contract addresses to filter (optional) :param topics: Event topic signatures, nested by position (optional) .. rubric:: Example >>> filter = LogFilter( ... from_block=18_000_000, ... to_block=18_010_000, ... addresses=["0xContractAddress..."], ... topics=[["0xTransfer..."]], # Match first topic ... ) .. py:attribute:: from_block :type: degenbot.types.aliases.BlockNumber .. py:attribute:: to_block :type: degenbot.types.aliases.BlockNumber .. py:attribute:: addresses :type: list[str] :value: [] .. py:attribute:: topics :type: list[list[str]] :value: []