degenbot.provider¶

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 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 AlloyProvider live in degenbot.provider.sync, the async twins in degenbot.provider.async_provider.

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¶

Package Contents¶

degenbot.provider.RustAlloyProvider¶
degenbot.provider.RustAsyncAlloyProvider¶
class degenbot.provider.AsyncAlloyProvider(rust_provider: degenbot.provider._rust.RustAsyncAlloyProvider)¶

Bases: _AsyncAlloyEndpointMixin, _AsyncAlloyQueryMixin, _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 as_async_alloy() for Rust-side call seams. The public surface is assembled from the async query mixins.

Parameters:

rust_provider – The underlying Rust AsyncAlloyProvider pyclass.

Prefer create() to construct an instance from an RPC URL.

static 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¶
Async:

Create an AsyncAlloyProvider asynchronously.

Parameters:

  • rpc_url – HTTP/HTTPS/WS/IPC endpoint URL.

  • max_retries – Maximum retry attempts.

  • max_blocks_per_request – Maximum blocks per log request.

  • requests_per_second – Optional rate limit.

  • burst – Optional burst size for rate limiting.

  • chain_id – Chain to bind the endpoint to; the Rust core raises ValueError when the endpoint serves another chain.

Returns:

An AsyncAlloyProvider instance.

exception degenbot.provider.ChainIdentityMismatchError¶

Bases: degenbot.exceptions.base.DegenbotValueError, 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 ValueError as well as a DegenbotValueError, so a caller that already handled a misconfigured endpoint keeps catching it exactly as it caught the core’s own refusal.

async degenbot.provider.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¶

Build a chain-bound AsyncAlloyProvider for the session chain.

Async counterpart of 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.

Parameters:

  • chain_id – The explicit chain override; resolved from the config layers when absent.

  • node – The explicit endpoint override, classified by its own value.

  • resolve_uri – The endpoint resolver called as resolve_uri(session_chain_id, node=node).

  • provider_factory – The awaited provider constructor called as provider_factory(endpoint, chain_id=session_chain_id).

  • 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 DegenbotValueError and a ValueError.

degenbot.provider.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 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 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.

Parameters:

  • chain_id – The explicit chain override; resolved from the config layers when absent.

  • node – The explicit endpoint override, classified by its own value.

  • resolve_uri – The endpoint resolver called as resolve_uri(session_chain_id, node=node).

  • provider_factory – The provider constructor called as provider_factory(endpoint, chain_id=session_chain_id).

  • 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 DegenbotValueError and a ValueError.

class degenbot.provider.OfflineProvider(chain_id: int, blocks: dict[str, dict[str, Any]])¶

Bases: _OfflineDataMixin, _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 AlloyProvider built over an in-memory offline transport, so it is indistinguishable from a live provider to any consumer (including BotIo). Recorded-JSON metadata (chain_id, block_numbers) is parsed in Python for the convenience accessors; the public surface is assembled from the mixins.

chain_id¶

The chain ID this provider serves data for

blocks¶

Dictionary of recorded block data keyed by block number string

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)
classmethod from_json_file(path: pathlib.Path) → OfflineProvider¶

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

Parameters:

path – Path to the JSON file containing recorded data

Returns:

An OfflineProvider instance loaded from the file.

class degenbot.provider.AlloyProvider(rpc_url: str, max_retries: int = 10, max_blocks_per_request: int = 5000, chain_id: int | None = None)¶

Bases: _AlloyEndpointMixin, _AlloyQueryMixin, _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.

Parameters:

  • rpc_url – HTTP/HTTPS endpoint URL

  • max_retries – Maximum retry attempts (default: 10)

  • max_blocks_per_request – Maximum logs per request (default: 5000)

  • chain_id – Chain to bind the endpoint to. The Rust core reads eth_chainId once at construction and raises ValueError when the endpoint serves another chain; None constructs the provider with no binding.

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)
class degenbot.provider.LogFilter¶

Filter criteria for log fetching.

Parameters:

  • from_block – Starting block number (inclusive)

  • to_block – Ending block number (inclusive)

  • addresses – Contract addresses to filter (optional)

  • topics – Event topic signatures, nested by position (optional)

Example

>>> filter = LogFilter(
...     from_block=18_000_000,
...     to_block=18_010_000,
...     addresses=["0xContractAddress..."],
...     topics=[["0xTransfer..."]],  # Match first topic
... )
from_block: degenbot.types.aliases.BlockNumber¶
to_block: degenbot.types.aliases.BlockNumber¶
addresses: list[str] = []¶
topics: list[list[str]] = []¶