2025-10-14

[MtkN1]

Today I Learned ...

[MtkN1]

Multicall3 Python Wrapper (Web3.py)

from __future__ import annotations

__all__ = [
    "AsyncMulticall3",
    "MULTICALL3_ABI",
    "MULTICALL3_ADDRESS",
    "Multicall3",
    "Result",
]

import json
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING

import eth_abi
from eth_utils import to_checksum_address
from eth_utils.abi import get_abi_input_types, get_abi_output_types
from eth_utils.conversions import to_hex
from hexbytes import HexBytes

if TYPE_CHECKING:
    from typing import Any

    from web3.contract import AsyncContract, Contract
    from web3.contract.async_contract import AsyncContractFunction
    from web3.contract.contract import ContractFunction

MULTICALL3_ADDRESS = to_checksum_address(
    "0xcA11bde05977b3631167028862bE2a173976CA11",
)
MULTICALL3_ABI = json.loads(
    Path(__file__).parent.joinpath("IMulticall3.json").read_bytes(),
)


@dataclass
class Result:
    success: bool
    return_data: Any


def create_call(
    contract_function: ContractFunction,
    *,
    allow_failure: bool = True,
) -> dict[str, Any]:
    encoded_arguments = eth_abi.encode(
        get_abi_input_types(contract_function.abi),
        contract_function.arguments,
    )

    call_data = to_hex(HexBytes(contract_function.selector) + encoded_arguments)

    return {
        "target": contract_function.address,
        "allowFailure": allow_failure,
        "callData": call_data,
    }


def parse_result(
    contract_function: ContractFunction,
    result: list[tuple[bool, bytes]],
) -> Result:
    success, raw_data = result

    if success:
        return_data = eth_abi.decode(
            get_abi_output_types(contract_function.abi),
            raw_data,
        )
    else:
        return_data = raw_data

    return Result(success, return_data)


class Multicall3:
    def __init__(self, contract: Contract, /, *, batch_size: int | None = None) -> None:
        self._contract = contract
        self._batch_size = batch_size

    def aggregate3(
        self,
        *calls: ContractFunction,
        allow_failure: bool = True,
    ) -> list[Result]:
        call_dicts = [create_call(call, allow_failure=allow_failure) for call in calls]

        total_raw_results: list[tuple[bool, bytes]] = []
        total_size = len(call_dicts)
        batch_size = self._batch_size or total_size
        for i in range(0, total_size, batch_size):
            batch = call_dicts[i : i + batch_size]
            raw_results = self._contract.functions["aggregate3"](batch).call()
            total_raw_results.extend(raw_results)

        return [
            parse_result(call, result) for call, result in zip(calls, total_raw_results)
        ]


class AsyncMulticall3:
    def __init__(
        self, contract: AsyncContract, /, *, batch_size: int | None = None
    ) -> None:
        self._contract = contract
        self._batch_size = batch_size

    async def aggregate3(
        self,
        *calls: AsyncContractFunction,
        allow_failure: bool = True,
    ) -> list[Result]:
        call_dicts = [create_call(call, allow_failure=allow_failure) for call in calls]

        total_raw_results: list[tuple[bool, bytes]] = []
        total_size = len(call_dicts)
        batch_size = self._batch_size or total_size
        for i in range(0, total_size, batch_size):
            batch = call_dicts[i : i + batch_size]
            raw_results = await self._contract.functions["aggregate3"](batch).call()
            total_raw_results.extend(raw_results)

        return [
            parse_result(call, result) for call, result in zip(calls, total_raw_results)
        ]

[MtkN1]

動的な並行処理の静的型付け

from __future__ import annotations

from typing import TYPE_CHECKING

import trio

if TYPE_CHECKING:
    from collections.abc import Awaitable, Callable


async def fetch_decimals(decimals: int, /) -> int:
    return decimals


async def fetch_name(name: str, /) -> str:
    return name


class _MISSING_TYPE:
    pass


_MISSING = _MISSING_TYPE()


class Task[*P, R]:
    def __init__(self, async_fn: Callable[[*P], Awaitable[R]], /, *args: *P) -> None:
        self._async_fn = async_fn
        self._args = args
        self._result: R | _MISSING_TYPE = _MISSING

    async def _await_impl(self) -> R:
        self._result = await self._async_fn(*self._args)
        return self._result

    def __call__(self) -> Awaitable[R]:
        return self._await_impl()

    def result(self) -> R:
        if isinstance(self._result, _MISSING_TYPE):
            raise RuntimeError

        return self._result


async def main() -> None:
    tasks_decimals = tuple(Task(fetch_decimals, decimal) for decimal in (18, 6))
    tasks_name = tuple(Task(fetch_name, name) for name in ("Wrapped Ether", "USD Coin"))

    async with trio.open_nursery() as nursery:
        for task_decimals in tasks_decimals:
            nursery.start_soon(task_decimals)
        for task_name in tasks_name:
            nursery.start_soon(task_name)

    decimals_list = [task_decimals.result() for task_decimals in tasks_decimals]
    name_list = [task_name.result() for task_name in tasks_name]

    print(*zip(name_list, decimals_list, strict=True))


if __name__ == "__main__":
    trio.run(main)

[MtkN1]

センチネル (sentinel) オブジェクト・ベストプラクティス

Refs: configparser.py#L389-L392

_UNSET = object()

Refs: dataclasses.py#L181-L185

class _MissingType:
    pass


_MISSING = _MissingType()