degenbot.fork

Anvil fork management utilities for local test chains.

A thin companion shell (ADR-005 three-layer Python layer) over the rust core degenbot._ffi.AnvilFork PyO3 seam (degenbot_fork::AnvilFork). The rust core owns the spawned anvil subprocess (via alloy::node_bindings::Anvil) + a connected alloy DynProvider (over IPC) + the 12 anvil dev-RPC methods. This Python shell:

• carries the anvil launch configuration in ForkLaunchConfig (the fork target stays a direct fork_url / fork_block argument);

• re-uses rust-driven dev-method invocations (mine / reset / set_snapshot / return_to_snapshot / set_balance / …);

• re-exposes general RPC against the forked node as self.provider — a degenbot._ffi.AlloyProvider constructed over the fork’s HTTP endpoint (replaces the legacy self.w3 Web3 handle; the rust-owned core’s dev-RPC runs over its own IPC DynProvider, while the companion general-RPC provider is HTTP so borrowed consumers never hold an alloy pubsub).

The breaking change vs the pre-0.7 Python AnvilFork is self.w3 → self.provider (and the dropped capture-file / middlewares / free-port management).

_FFIAnvilFork exposes the IPC path the rust core resolved for the spawned anvil process. Former callers using fork.w3.eth.X are migrated to fork.provider.X — the legacy attribute is gone.

Package Contents

exception degenbot.fork.DegenbotValueError(*, message: str | None = None)

Bases: DegenbotError

DegenbotValueError error.

exception degenbot.fork.AnvilError(method: str, error: str)

Bases: degenbot.exceptions.base.DegenbotError

Raised on errors resulting from failed calls to Anvil via JSON-RPC.

This exception is specifically for errors that occur when making RPC calls to an Anvil instance, such as invalid method calls, parameter errors, or other Anvil-specific failures.

method

error

degenbot.fork.logger

class degenbot.fork.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)

type degenbot.fork.BlockNumber = int

type degenbot.fork.ValidatedUint256 = Annotated[int, Field(ge=MIN_UINT256, le=MAX_UINT256)]

class degenbot.fork.ForkLaunchConfig

Non-target launch knobs for the rust-owned anvil subprocess.

The fork target (fork_url / fork_block) stays a direct AnvilFork argument because nearly every caller names it; this carrier groups the anvil-CLI, mining, and post-spawn state-override families that most callers leave at their defaults. Values forward to the rust core unchanged (localhost becomes its host key; ipc_path is stringified).

localhost: str = '127.0.0.1'

fork_transaction_hash: str | None = None

mining_mode: Literal['auto', 'interval', 'none'] = 'auto'

mining_interval: int | None = None

storage_caching: bool = True

base_fee: int | None = None

ipc_path: pathlib.Path | None = None

mnemonic: str = 'patient rude simple dog close planet oval animal hunt sketch suspect slim'

chain_id: int | None = None

balance_overrides: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress, int]] | None = None

bytecode_overrides: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress, bytes]] | None = None

nonce_overrides: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress, int]] | None = None

storage_overrides: collections.abc.Iterable[tuple[degenbot.types.chain.HexAddress | bytes, int, degenbot.types.chain.HexStr | bytes | int]] | None = None

coinbase: degenbot.types.chain.HexAddress | None = None

anvil_opts: list[str] | None = None

class degenbot.fork.AnvilFork(*, fork_url: str | None = None, fork_block: degenbot.types.aliases.BlockNumber | None = None, launch: ForkLaunchConfig | None = None)

A rust-owned Anvil fork subprocess + IPC DynProvider + dev-RPC surface.

Companion shell (ADR-005 Python layer) over degenbot._ffi.AnvilFork (the PyO3 seam on degenbot_fork::AnvilFork). The rust core owns the subprocess lifecycle (drop = kill) + an alloy DynProvider (IPC transport) + the 12 anvil dev-RPC methods. The shell carries the legacy constructor knobs in ForkLaunchConfig, preserves the legacy dev-method names, raises the legacy AnvilError exception subclass on rust-side RPC failures, and exposes general RPC through the AlloyProvider pyclass at provider (replacing the legacy self.w3 Web3 handle — that handle is gone; all callers now use fork.provider.get_block_number() etc.).

The rust core (degenbot_fork::AnvilFork) owns:

• the anvil subprocess lifecycle (drop = kill);

• a connected alloy DynProvider over IPC (transport);

• the 12 dev-RPC methods (evm_mine/anvil_reset/evm_snapshot/ evm_revert/anvil_set_balance/…), exposed on this shell as the legacy Python names (mine/reset/set_snapshot/ return_to_snapshot/set_balance/…).

The Python shell keeps:

• the legacy constructor knobs (grouped in ForkLaunchConfig) + the multi-override queueing (balance_overrides / bytecode_overrides / nonce_overrides / storage_overrides / coinbase) — applied post-spawn via the dev-method delegations;

• the legacy Python dev-method names so existing API surface doesn’t reshuffle (the rust core’s name for set_snapshot is snapshot, return_to_snapshot → revert, set_storage → set_storage_at, set_next_base_fee → set_next_base_fee — these translations happen in-shell);

• the legacy AnvilError exception on rust RPC failures (translated from rust’s PyRuntimeError at the boundary, so the legacy AnvilError(method=..., error=...) shape + the exception-subclass identity stay intact);

• the @validate_call + ValidatedUint256 annotations on set_balance / set_next_base_fee / set_next_block_timestamp so input-range validation raises pydantic ValidationError before the rust round-trip (matches the legacy pre-0.7 surface).

Dropped (no longer needed now that subprocess lifecycle is rust-owned): the –auto-impersonate CLI arg list construction, the _setup_process / _setup_w3 / _close_ipc_socket block, the socket-based free-port allocation, the tenacity spawn retry, the subprocess.Popen capture-file management (capture_path / preserve_capture / *_capture_filename properties), the web3.IPCProvider client + middleware_onion.inject(middleware) plumbing (alloy’s DynProvider is the only transport layer under the rust core), the ipc_provider_kwargs shim, and http_url / ws_url / port / ipc_filename (anvil-CLI control knobs that the rust core controls directly now — the IPC path is the only socket surfaced back, via ipc_path).

The one breaking change for callers: self.w3 is gone. Use provider (a degenbot._ffi.AlloyProvider) for general RPC.

property provider: degenbot.provider.AlloyProvider

The general-RPC AlloyProvider bound to the fork over HTTP.

Use for retry-aware eth_call / get_logs against the in-memory fork (mirrors the legacy self.w3 handle). HTTP (not the rust core’s IPC DynProvider) so that consumers which borrow it never hold an alloy IPC/WS pubsub that would reconnect into a dead socket when the fork closes. Raising once the fork is closed guarantees a provider over a dead subprocess is never silently handed out.

Raises:

AnvilError – If the fork was closed.

property fork_url: str | None

Fork URL the spawned anvil process forks from, or None for in-memory.

property ipc_path: str

Resolved IPC socket path the spawned anvil process listens on.

Use to construct a second AlloyProvider (or other transport) against the same forked node. Replaces the legacy AnvilFork.ipc_filename (a pathlib.Path derived from a user-supplied ipc_path parent + the allocated HTTP port — that bookkeeping is gone; the rust core owns the IPC socket directly + reports the resolved path here).

Raises:

AnvilError – If the fork was closed.

property http_url: str

HTTP endpoint of the spawned anvil node (e.g. http://127.0.0.1:PORT).

Use to construct a second AlloyProvider over the rust-owned anvil subprocess when the IPC-bound provider is not enough (e.g. a plain HTTP transport from another Python process).

property ws_url: str

WebSocket endpoint of the spawned anvil node (e.g. ws://127.0.0.1:PORT).

property port: int

TCP port the spawned anvil node listens on.

close() → None

Drop the rust-owned subprocess + IPC handle.

The rust core (degenbot_fork::AnvilFork) owns the subprocess via alloy’s AnvilInstance; dropping the _FFIAnvilFork pyclass handle terminates the anvil process + closes the IPC transport. This method just clears the Python-side reference so the GC finalizes the rust handle — there is no separate IPC-socket cleanup needed (alloy’s AnvilInstance Drop impl handles it).

mine() → None

Mine a single block (evm_mine).

Raises:

AnvilError – If the RPC call fails.

reset(fork_url: str | None = None, block_number: degenbot.types.aliases.BlockNumber | None = None, transaction_hash: str | None = None) → None

Fork from a new endpoint, block number, or transaction hash.

Resetting to a new block number only is done in-place (no subprocess relaunch). Resetting to a new endpoint or from a transaction hash tears down the existing rust-owned subprocess + spawns a new one with the new fork parameters. Slower than in-place.

Raises:

• AnvilError – If the in-place RPC call fails.

• DegenbotValueError – If no reset options are provided.

set_balance(address: str, balance: degenbot.validation.evm_values.ValidatedUint256) → None

Set balance (anvil_setBalance).

Raises:

AnvilError – If the RPC call fails.

set_code(address: str, bytecode: bytes) → None

Set code (anvil_setCode).

Raises:

AnvilError – If the RPC call fails.

set_coinbase(address: str) → None

Set coinbase (anvil_setCoinbase).

Raises:

AnvilError – If the RPC call fails.

set_block_timestamp_interval(interval: int) → None

Set block timestamp interval (anvil_setBlockTimestampInterval).

Raises:

AnvilError – If the RPC call fails.

set_next_base_fee(fee: degenbot.validation.evm_values.ValidatedUint256) → None

Set the next block base fee (anvil_setNextBlockBaseFeePerGas).

Raises:

AnvilError – If the RPC call fails.

set_next_block_timestamp(timestamp: degenbot.validation.evm_values.ValidatedUint256) → None

Set the next block timestamp (evm_setNextBlockTimestamp).

Raises:

AnvilError – If the RPC call fails.

set_nonce(address: str, nonce: int) → None

Set nonce (anvil_setNonce).

Raises:

AnvilError – If the RPC call fails.

set_storage(address: degenbot.types.chain.HexAddress | bytes, position: int, value: degenbot.types.chain.HexStr | bytes | int) → None

Set storage (anvil_setStorageAt).

Raises:

AnvilError – If the RPC call fails.