### Setup DankWeb3 from Sync Web3 Instance (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Converts a synchronous web3.py instance into an asynchronous DankWeb3 instance, enabling RPC batching. This is the primary method for integrating Dank Mids with existing synchronous web3.py setups. It requires the web3 library and dank_mids.helpers.

```python
from web3 import Web3
from dank_mids.helpers import setup_dank_w3_from_sync

# Create a standard sync Web3 instance
sync_w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"))

# Convert to async DankWeb3 with batching middleware
dank_web3 = setup_dank_w3_from_sync(sync_w3)

# Now use async methods - calls are automatically batched
async def main():
    # These calls will be batched together in a single HTTP request
    block = await dank_web3.eth.get_block("latest")
    balance = await dank_web3.eth.get_balance("0x742d35Cc6634C0532925a3b844Bc9e7595f0Ab1b")
    chain_id = await dank_web3.eth.chain_id

    print(f"Block: {block.number}, Balance: {balance}, Chain: {chain_id}")

import asyncio
asyncio.run(main())
```

--------------------------------

### Setup Dank Mids with web3.py

Source: https://github.com/bobthebuidler/dank_mids/blob/master/README.md

Demonstrates how to integrate Dank Mids with web3.py. It shows two ways to obtain a Dank Mids-wrapped web3 instance: by using `setup_dank_w3_from_sync` with an existing sync web3 instance, or by directly importing the `dank_web3` object if it's pre-configured.

```python
from dank_mids.helpers import setup_dank_w3_from_sync
dank_web3 = setup_dank_w3_from_sync(w3)
# OR
from dank_mids import dank_web3

# Then:
random_block = await dank_web3.eth.get_block(123)
```

--------------------------------

### Install Dank Mids with Pip

Source: https://github.com/bobthebuidler/dank_mids/blob/master/README.md

This command installs the Dank Mids library using pip, making it available for use in your Python projects. Ensure you have pip installed and configured correctly.

```bash
pip install dank-mids
```

--------------------------------

### Compile Vendored Libraries with mypyc

Source: https://github.com/bobthebuidler/dank_mids/blob/master/dank_mids/_vendor/README.md

This Makefile example demonstrates how to include vendored library files in the mypyc compilation process. Specific `.py` files from the vendored directories are listed as arguments to the mypyc command.

```makefile
mypyc:
	mypyc ... dank_mids/_vendor/aiolimiter/__init__.py dank_mids/_vendor/aiolimiter/aiolimiter.py ...
```

--------------------------------

### Setup DankWeb3 from Async Web3 Instance (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Initializes a DankWeb3 instance from an existing asynchronous web3.py instance. This function is useful when you already have an AsyncWeb3 object and want to add Dank Mids' batching capabilities. It requires the web3 library and dank_mids.helpers.

```python
from web3 import AsyncWeb3
from web3.providers.async_rpc import AsyncHTTPProvider
from dank_mids.helpers import setup_dank_w3

# Create an async Web3 instance
async_w3 = AsyncWeb3(AsyncHTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"))

# Inject dank middleware
dank_web3 = setup_dank_w3(async_w3)

async def fetch_data():
    # All calls through dank_web3 are batched
    block_number = await dank_web3.eth.block_number
    latest_block = await dank_web3.eth.get_block("latest")
    return block_number, latest_block
```

--------------------------------

### Get Event Logs with Dank Mids

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Fetches event logs for a given contract address within a specified block range and automatically decodes them into typed structures. Requires the 'dank_mids' and 'evmspec.structs.log.Log' libraries.

```python
from dank_mids import dank_web3
from evmspec.structs.log import Log

async def fetch_logs():
    logs = await dank_web3.eth.get_logs({
        "fromBlock": 15000000,
        "toBlock": 15000010,
        "address": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2",  # WETH
    })

    print(f"Found {len(logs)} logs")
    for log in logs[:5]:
        print(f"  Block {log.blockNumber}: {log.topics[0].hex()[:16]}...")
```

--------------------------------

### Complete Blockchain Data Analysis Example (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

A comprehensive Python script using 'dank_mids' and 'asyncio' to concurrently fetch and analyze blockchain data, including Uniswap pool tokens, reserves over time, and block timestamps. It demonstrates efficient batching of multiple asynchronous calls.

```python
import asyncio
from typing import Any
from web3.types import Timestamp

import dank_mids

# Define target pools and blocks
BLOCKS = [15_000_000, 15_100_000, 15_200_000, 15_300_000, 15_400_000, 15_500_000]
UNISWAP_POOLS = [
    "0xBb2b8038a1640196FbE3e38816F3e67Cba72D940",  # WBTC-WETH
    "0x0d4a11d5EEaaC28EC3F61d100daF4d40471f1852",  # USDT-WETH
    "0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc",  # USDC-WETH
    "0xA478c2975Ab1Ea89e8196811F51A7B7Ade33eB11",  # DAI-WETH
]

async def get_tokens_for_pool(pool: dank_mids.Contract) -> tuple[str, str]:
    """Fetch token0 and token1 addresses for a pool."""
    return await asyncio.gather(pool.token0, pool.token1)

async def get_reserves_for_blocks(
    pool: dank_mids.Contract, blocks: list[int]
) -> list[tuple[int, int, int]]:
    """Fetch reserves at multiple blocks for a pool."""
    return await asyncio.gather(
        *(pool.getReserves.coroutine(block_identifier=block) for block in blocks)
    )

async def get_timestamp_at_block(block: int) -> Timestamp:
    """Fetch timestamp for a specific block using dank_mids.eth."""
    data = await dank_mids.eth.get_block(block)
    return data.timestamp

async def main():
    # Initialize pool contracts
    pools = tuple(map(dank_mids.Contract, UNISWAP_POOLS))

    # Fetch all data concurrently - dank_mids batches everything!
    tokens, timestamps, all_reserves = await asyncio.gather(
        asyncio.gather(*(get_tokens_for_pool(pool) for pool in pools)),
        asyncio.gather(*(get_timestamp_at_block(block) for block in BLOCKS)),
        asyncio.gather(*(get_reserves_for_blocks(pool, BLOCKS) for pool in pools)),
    )

    # Process results
    print("=== Pool Tokens ===")
    for pool, (token0, token1) in zip(pools, tokens):
        print(f"{pool.address[:10]}...: token0={token0}, token1={token1}")

    print("\n=== Block Timestamps ===")
    for block, ts in zip(BLOCKS, timestamps):
        print(f"Block {block}: {ts}")

    print("\n=== Reserves Over Time ===")
    for pool, reserves in zip(pools, all_reserves):
        print(f"\n{pool.address[:10]}...:")
        for block, (r0, r1, _) in zip(BLOCKS, reserves):
            print(f"  Block {block}: reserve0={r0:,}, reserve1={r1:,}")

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Get Block Timestamps Efficiently (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Retrieves only the timestamp for multiple blocks concurrently, optimizing performance by avoiding full block decoding. This method is part of the DankEth API and requires the dank_web3 instance.

```python
from dank_mids import dank_web3

async def get_timestamps():
    # Fetch timestamps for multiple blocks concurrently
    import asyncio

    blocks = [15000000, 15100000, 15200000, 15300000]
    timestamps = await asyncio.gather(
        *(dank_web3.eth.get_block_timestamp(block) for block in blocks)
    )

    for block, ts in zip(blocks, timestamps):
        print(f"Block {block}: {ts}")

    # Output:
    # Block 15000000: 1656016034
    # Block 15100000: 1657254623
    # Block 15200000: 1658507831
    # Block 15300000: 1659761428
```

--------------------------------

### Get Transaction Status with Dank Mids

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Quickly retrieves the status of one or more transactions using Dank Mids without decoding the full receipt. This is useful for efficiently checking if transactions succeeded or failed. It takes a list of transaction hashes and returns their statuses. Requires the 'dank_mids' library.

```python
from dank_mids import dank_web3

async def check_statuses():
    import asyncio

    tx_hashes = [
        "0x5c504ed432cb51138bcf09aa5e8a410dd4a1e204ef84bfed1be16dfba1b22060",
        "0x6f1b50e50d0b6d8e3c86a7e6c8c5a8f8a3f7c5b8e6d8a3c7b5a8f6d4c2b0a9e8",
    ]

    statuses = await asyncio.gather(
        *(dank_web3.eth.get_transaction_status(h) for h in tx_hashes)
    )

    for tx_hash, status in zip(tx_hashes, statuses):
        print(f"{tx_hash[:16]}...: {'Success' if status else 'Failed'}")
```

--------------------------------

### Get Transaction Receipt with Dank Mids

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Fetches the full transaction receipt for a given transaction hash using Dank Mids. It decodes the receipt into a TransactionReceipt object, providing details like status, gas used, block number, and logs. Requires the 'dank_mids' and 'evmspec' libraries.

```python
from dank_mids import dank_web3
from evmspec import TransactionReceipt

async def check_receipt():
    tx_hash = "0x5c504ed432cb51138bcf09aa5e8a410dd4a1e204ef84bfed1be16dfba1b22060"

    # Get full receipt
    receipt = await dank_web3.eth.get_transaction_receipt(tx_hash)
    print(f"Status: {receipt.status}")
    print(f"Gas used: {receipt.gasUsed}")
    print(f"Block: {receipt.blockNumber}")
    print(f"Logs: {len(receipt.logs)}")
```

--------------------------------

### Configure Dank Mids Behavior with Environment Variables (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Shows how to customize Dank Mids behavior by setting environment variables before importing the library. This allows tuning parameters like multicall size, JSONRPC batch size, request rate, and timeouts. The script imports 'os' to set variables and then 'dank_mids' to access the configuration.

```python
import os

# Set before importing dank_mids
os.environ["DANK_MIDS_MAX_MULTICALL_SIZE"] = "5000"  # Max eth_calls per multicall (default: 10000)
os.environ["DANK_MIDS_MAX_JSONRPC_BATCH_SIZE"] = "250"  # Max RPC calls per batch (default: 500)
os.environ["DANK_MIDS_REQUESTS_PER_SECOND"] = "25"  # Rate limit (default: 50)
os.environ["DANK_MIDS_DEMO_MODE"] = "True"  # Visual batching output (default: False)
os.environ["DANK_MIDS_AIOHTTP_TIMEOUT"] = "600"  # Request timeout in seconds (default: 1200)
os.environ["DANK_MIDS_GANACHE_FORK"] = "True"  # Disable state override for Ganache (default: False)

# Now import dank_mids
from dank_mids import dank_web3

# Environment variables are read at import time
# Check current configuration
from dank_mids import ENVIRONMENT_VARIABLES as ENVS
print(f"Max multicall size: {ENVS.MAX_MULTICALL_SIZE}")
print(f"Max batch size: {ENVS.MAX_JSONRPC_BATCH_SIZE}")
print(f"Demo mode: {ENVS.DEMO_MODE}")
```

--------------------------------

### Dank Mids Contract Integration with Brownie

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Demonstrates how to integrate Dank Mids with Brownie contracts, enabling asynchronous calls to contract methods. It shows how to initialize a contract as a Dank Mids.Contract object and use asyncio.gather for concurrent batched calls. Requires 'asyncio', 'dank_mids'.

```python
import asyncio
import dank_mids

# Initialize contracts as dank_mids.Contract objects
uniswap_pool = dank_mids.Contract("0xBb2b8038a1640196FbE3e38816F3e67Cba72D940")

async def main():
    # Use asyncio.gather for concurrent batched calls
    token0, token1, reserves = await asyncio.gather(
        uniswap_pool.token0,  # Awaitable for no-arg methods at latest block
        uniswap_pool.token1,
        uniswap_pool.getReserves.coroutine(),  # .coroutine() for explicit async call
    )

    print(f"Token0: {token0}")
    print(f"Token1: {token1}")
    print(f"Reserves: {reserves}")

asyncio.run(main())
```

--------------------------------

### Async Calls with Contract.coroutine() in Dank Mids

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Illustrates the use of the `.coroutine()` method on Dank Mids contracts to enable asynchronous calls with specific block identifiers. This allows fetching historical contract data efficiently through batching. Requires 'asyncio', 'dank_mids'.

```python
import asyncio
import dank_mids

pool = dank_mids.Contract("0xBb2b8038a1640196FbE3e38816F3e67Cba72D940")

async def fetch_historical_data():
    blocks = [15000000, 15100000, 15200000, 15300000, 15400000, 15500000]

    # Fetch reserves at multiple historical blocks - all batched!
    reserves = await asyncio.gather(
        *(pool.getReserves.coroutine(block_identifier=block) for block in blocks)
    )

    for block, reserve in zip(blocks, reserves):
        reserve0, reserve1, timestamp = reserve
        print(f"Block {block}: Reserve0={reserve0}, Reserve1={reserve1}")

asyncio.run(fetch_historical_data())
```

--------------------------------

### Inject Dank Middleware into Web3 Stack (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Manually injects the `dank_middleware` into a web3.py middleware stack for fine-grained control over JSON-RPC batching. Injecting at layer 0 ensures it's the first middleware to process requests, maximizing batching opportunities. Requires web3 and dank_mids.middleware.

```python
from web3 import AsyncWeb3
from web3.providers.async_rpc import AsyncHTTPProvider
from dank_mids.middleware import dank_middleware

# Create async Web3 instance
w3 = AsyncWeb3(AsyncHTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"))

# Inject dank middleware at layer 0 (lowest layer) for maximum batching
w3.middleware_onion.inject(dank_middleware, layer=0)

async def example():
    # Multiple concurrent calls are automatically batched
    import asyncio
    results = await asyncio.gather(
        w3.eth.get_block(15000000),
        w3.eth.get_block(15000001),
        w3.eth.get_block(15000002),
    )
    for block in results:
        print(f"Block {block['number']}: {block['timestamp']}")
```

--------------------------------

### Add New Vendored Library with Git Submodule

Source: https://github.com/bobthebuidler/dank_mids/blob/master/dank_mids/_vendor/README.md

This command adds an upstream repository as a Git submodule to the specified local path. It's the first step in vendoring a new library, ensuring it's tracked separately within your repository.

```shell
git submodule add <REPO_URL> dank_mids/_vendor/<package_name>
git submodule update --init --recursive
```

--------------------------------

### Trace Transaction with Dank Mids

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Retrieves all execution traces produced by a specific transaction hash. This provides a detailed breakdown of the transaction's execution flow. Requires the 'dank_mids' library.

```python
from dank_mids import dank_web3

async def trace_tx():
    tx_hash = "0x5c504ed432cb51138bcf09aa5e8a410dd4a1e204ef84bfed1be16dfba1b22060"
    traces = await dank_web3.eth.trace_transaction(tx_hash)

    print(f"Transaction produced {len(traces)} traces:")
    for trace in traces:
        print(f"  {trace.type}: {trace.action}")
```

--------------------------------

### BlockSemaphore for Prioritized Concurrency (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Demonstrates the use of BlockSemaphore to manage concurrent asynchronous operations, prioritizing tasks associated with older block numbers. It requires the 'dank_mids' library and 'asyncio'. The semaphore is initialized with a concurrency limit and used as a context manager within an async function.

```python
from dank_mids.semaphores import BlockSemaphore

# Create a semaphore allowing 10 concurrent operations
sem = BlockSemaphore(10, name="my_semaphore")

async def fetch_with_priority(block_number: int):
    # Lower block numbers get higher priority
    async with sem[block_number]:
        # Perform operation
        result = await some_async_call(block_number)
        return result

async def main():
    import asyncio

    # Blocks with lower numbers are processed first
    blocks = [15500000, 15000000, 15200000, 15100000]
    results = await asyncio.gather(
        *(fetch_with_priority(b) for b in blocks)
    )
```

--------------------------------

### Update Vendored Library with Git Fetch and Pull

Source: https://github.com/bobthebuidler/dank_mids/blob/master/dank_mids/_vendor/README.md

This process updates a vendored library to its latest upstream version by fetching changes and pulling them into the local submodule directory. It requires navigating to the submodule's directory and performing standard Git operations.

```shell
cd dank_mids/_vendor/<package_name>
git fetch
git checkout main
git pull
```

--------------------------------

### Trace Filter with Dank Mids

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Filters and retrieves execution traces within a specified block range for a given contract address. This function automatically handles large block ranges for efficient trace retrieval. Requires the 'dank_mids' library.

```python
from dank_mids import dank_web3

async def get_traces():
    # Get all traces for a specific contract in a block range
    traces = await dank_web3.eth.trace_filter({
        "fromBlock": 15000000,
        "toBlock": 15000100,
        "toAddress": ["0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"],  # WETH
    })

    print(f"Found {len(traces)} traces")
    for trace in traces[:5]:
        print(f"  Type: {trace.type}, From: {trace.action.from_}, Value: {trace.action.value}")
```

--------------------------------

### Import and Use Vendored Python Library

Source: https://github.com/bobthebuidler/dank_mids/blob/master/dank_mids/_vendor/README.md

This Python code snippet shows how to import a class or function from a vendored library. Imports should be made from the `_vendor` path to ensure the vendored version is used, avoiding conflicts with external dependencies.

```python
from dank_mids._vendor.aiolimiter import AsyncLimiter
```

--------------------------------

### Retrieve Block Transactions (Python)

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Fetches transactions from a specified block, with an option to retrieve only transaction hashes for improved performance. This method is available through the DankEth API and supports batching for multiple requests.

```python
from dank_mids import dank_web3

async def fetch_transactions():
    # Get full transaction objects
    transactions = await dank_web3.eth.get_transactions(15000000)
    print(f"Block has {len(transactions)} transactions")
    for tx in transactions[:3]:
        print(f"  TX: {tx.hash.hex()} from {tx.sender} to {tx.to}")

    # Get only transaction hashes (faster)
    tx_hashes = await dank_web3.eth.get_transactions(15000000, hashes_only=True)
    print(f"Transaction hashes: {[h.hex()[:16]}... for h in tx_hashes[:3]]")
```

--------------------------------

### Run Pytest with Disabled Plugin Autoload

Source: https://github.com/bobthebuidler/dank_mids/blob/master/AGENTS.md

This command executes the Pytest test suite while disabling plugin autoloading. It's used to ensure tests are run in a controlled environment, preventing potential interference from automatically loaded plugins. The `PYTEST_DISABLE_PLUGIN_AUTOLOAD=1` environment variable is a prerequisite for this specific test execution.

```shell
PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 pytest <files>
```

--------------------------------

### Patch Brownie Contract for Async with Dank Mids

Source: https://context7.com/bobthebuidler/dank_mids/llms.txt

Shows how to patch an existing Brownie contract object to add asynchronous capabilities using Dank Mids. This avoids the need to create a new Dank Mids.Contract instance and allows direct use of `.coroutine()` on the patched contract. Requires 'brownie', 'dank_mids'.

```python
import brownie
from dank_mids import patch_contract, dank_web3

# Existing brownie contract
brownie_contract = brownie.Contract("0xBb2b8038a1640196FbE3e38816F3e67Cba72D940")

# Patch it to add async methods
patched = patch_contract(brownie_contract, dank_web3)

async def use_patched():
    # Now you can use .coroutine() on the patched contract
    result = await patched.getReserves.coroutine(block_identifier=15000000)
    print(f"Reserves: {result}")

# Note: To run this, you would typically need an async event loop setup,
# e.g., by calling use_patched() within an async function and running it with asyncio.run()
```

--------------------------------

### Makefile Target to Update Vendored Library

Source: https://github.com/bobthebuidler/dank_mids/blob/master/dank_mids/_vendor/README.md

This Makefile target automates the process of updating a specific vendored library. It encapsulates the Git commands needed to fetch and pull the latest changes from the upstream repository.

```makefile
update-aiolimiter:
	cd dank_mids/_vendor/aiolimiter && git fetch && git checkout main && git pull
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.