wait_other_writers

Documentation for eth_defi.utils.wait_other_writers function.

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