utils

Documentation for eth_defi.utils Python module.

Bunch of random utilities.

Functions

addr(address)	Convert various address formats to HexAddress.
chunked(iterable, chunk_size)
find_free_port([min_port, max_port, max_attempt])	Find a free localhost port to bind.
from_unix_timestamp(timestamp)	Convert UNIX seconds since epoch to naive Python datetime.
get_url_domain(url)	Redact URL so that only domain is displayed.
is_good_multichain_address(address)	Check if a vault address has a recognised format.
is_localhost_port_listening(port[, host])	Check if a localhost is running a server already.
sanitise_string(s[, max_length])	Remove null characters.
shutdown_hard(process[, log_level, block, ...])	Kill Psutil process.
to_unix_timestamp(dt)	Convert Python UTC datetime to UNIX seconds since epoch.
wait_other_writers(path[, timeout])	Wait other potential writers writing the same file.

is_good_multichain_address(address)

Check if a vault address has a recognised format.

• EVM vaults use 0x-prefixed hex addresses

• Non-EVM protocols like GRVT use platform-specific IDs (e.g. VLT:xxx)

• Lighter pools use synthetic IDs (e.g. lighter-pool-281474976710654)

• Hibachi vaults use synthetic IDs (e.g. hibachi-vault-2)

Parameters

address (str) – The vault address string to validate.

Returns

True if the address starts with a known prefix.

Return type

bool

sanitise_string(s, max_length=None)

Remove null characters.

Parameters

• s (str) –

• max_length (Optional[int]) –

Return type

str

is_localhost_port_listening(port, host='localhost')

Check if a localhost is running a server already.

See https://www.adamsmith.haus/python/answers/how-to-check-if-a-network-port-is-open-in-python

Returns

True if there is a process occupying the port

Parameters

port (int) –

Return type

bool

find_free_port(min_port=20000, max_port=40000, max_attempt=20)

Find a free localhost port to bind.

Does by random.

Note

Subject to race condition, but should be rareish.

Parameters

• min_port (int) – Minimum port range

• max_port (int) – Maximum port range

• max_attempt (int) – Give up and die with an exception if no port found after this many attempts.

Returns

Free port number

Return type

int

shutdown_hard(process, log_level=None, block=True, block_timeout=30, check_port=None)

Kill Psutil process.

• Straight out OS SIGKILL a process

• Log output if necessary

• Use port listening to check that the process goes down and frees its ports

Parameters

• process (psutil.Popen) – Process to kill

• block –

  Block the execution until the process has terminated.

  You must give check_port option to ensure we enforce the shutdown.

• block_timeout – How long we give for process to clean up after itself

• log_level (Optional[int]) – If set, dump anything in Anvil stdout to the Python logging using level INFO.

• check_port (Optional[int]) – Check that TCP/IP localhost port is freed after shutdown

Returns

stdout, stderr as string

Return type

tuple[bytes, bytes]

to_unix_timestamp(dt)

Convert Python UTC datetime to UNIX seconds since epoch.

Example:

import datetime
from eth_defi.utils import to_unix_timestamp

dt = datetime.datetime(1970, 1, 1)
unix_time = to_unix_timestamp(dt)
assert unix_time == 0

Parameters

dt (datetime.datetime) – Python datetime to convert

Returns

Datetime as seconds since 1970-1-1

Return type

float

from_unix_timestamp(timestamp)

Convert UNIX seconds since epoch to naive Python datetime.

Parameters

timestamp (float) – Timestamp in since 1970-1-1 as float or int as seconds

Returns

Naive Python datetime in UTC timezone (tzinfo is None, but the time is in UTC)

Return type

datetime.datetime

get_url_domain(url)

Redact URL so that only domain is displayed.

Some services e.g. infura use path as an API key.

Parameters

url (str) –

Return type

str

addr(address)

Convert various address formats to HexAddress.

Args:

address: Can be a string, HexAddress, or HexStr

Returns:

HexAddress object

Parameters

address (Union[str, eth_typing.evm.HexAddress, eth_typing.encoding.HexStr]) –

Return type

eth_typing.evm.HexAddress

wait_other_writers(path, timeout=120)

Wait other potential writers writing the same file.

• Work around issues when parallel unit tests and such try to write the same file

Example:

import urllib
import tempfile

import pytest
import pandas as pd


@pytest.fixture()
def my_cached_test_data_frame() -> pd.DataFrame:
    # Al tests use a cached dataset stored in the /tmp directory
    path = os.path.join(tempfile.gettempdir(), "my_shared_data.parquet")

    with wait_other_writers(path):
        # Read result from the previous writer
        if not path.exists(path):
            # Download and write to cache
            urllib.request.urlretrieve("https://example.com", path)

        return pd.read_parquet(path)

Parameters

• path (Union[pathlib.Path, str]) – File that is being written

• timeout (int) –

  How many seconds wait to acquire the lock file.

  Default 2 minutes.

Raises

filelock.Timeout – If the file writer is stuck with the lock.