Module: utils

Utility functions for the Arweave integration.

edge_unix_ts(edge)

Helper function to extract the unix time stamp from an Arweave transaction edge. See https://arweave.net/graphql for the Arweave graphql schema.

Parameters:

Name	Type	Description	Default
edge	dict[str, Any]	a transaction edge object	required

Returns:

Name	Type	Description
float	float	unix timestamp in seconds

Source code in src/ritual_arweave/utils.py

def edge_unix_ts(edge: dict[str, Any]) -> float:
    """
    Helper function to extract the unix time stamp from an
    Arweave transaction edge. See https://arweave.net/graphql for the
    Arweave graphql schema.

    Args:
        edge (dict[str, Any]): a transaction edge object

    Returns:
        float: unix timestamp in seconds
    """
    # sort matching manifests by time, get latest
    tag_dicts: list[dict[str, str]] = edge["node"]["tags"]
    return float(get_tags_dict(tag_dicts)["Unix-Time"])

get_sha256_digest(file_path)

Helper function that computes the digest of a file in binary mode to handle potentially large files.

Parameters:

Name	Type	Description	Default
file_path	str	path to a file	required

Returns:

Name	Type	Description
str	str	hex string representing the sha256

Source code in src/ritual_arweave/utils.py

def get_sha256_digest(file_path: str) -> str:
    """Helper function that computes the digest
    of a file in binary mode to handle potentially
    large files.

    Args:
        file_path (str): path to a file

    Returns:
        str: hex string representing the sha256
    """
    h = hashlib.sha256()

    with open(file_path, "rb") as file:
        while True:
            # Reading is buffered, so we can read smaller chunks.
            chunk = file.read(h.block_size)
            if not chunk:
                break
            h.update(chunk)

    return h.hexdigest()

get_tags_dict(tag_dicts)

Helper function to merge a list of tag dicts into a single dictionary.

Parameters:

Name	Type	Description	Default
tag_dicts(list[dict[str,	str]	a list of tag dicts with	required

Returns:

Type	Description
dict[str, str]	dict[str, str]: a key value dict mapping tag name to tag value

Source code in src/ritual_arweave/utils.py

def get_tags_dict(tag_dicts: list[dict[str, str]]) -> dict[str, str]:
    """
    Helper function to merge a list of tag dicts into
    a single dictionary.

    Args:
        tag_dicts(list[dict[str, str]): a list of tag dicts with
        keys 'name' and 'value' corresponding to the name and
        value of the tag respectively.

    Returns:
        dict[str, str]: a key value dict mapping tag name to tag value
    """
    tags: dict[str, str] = {item["name"]: item["value"] for item in tag_dicts}
    return tags

load_wallet(file_path=None, api_url=DEFAULT_API_URL)

Helper function to load a wallet from a file path. If a file path is provided as an argument, it is used. Otherwise, the file path is read from the ARWEAVE_WALLET_FILE_PATH environment variable.

Returns:

Name	Type	Description
Wallet	Wallet	an Arweave wallet object

Source code in src/ritual_arweave/utils.py

def load_wallet(
    file_path: Optional[str] = None, api_url: str = DEFAULT_API_URL
) -> Wallet:
    """
    Helper function to load a wallet from a file path. If a file path is provided
    as an argument, it is used. Otherwise, the file path is read from the
    ARWEAVE_WALLET_FILE_PATH environment variable.

    Returns:
        Wallet: an Arweave wallet object
    """
    if not (wallet_file_path := file_path or os.getenv("ARWEAVE_WALLET_FILE_PATH")):
        raise ValueError(
            "Could not find wallet file path. Please either set "
            "ARWEAVE_WALLET_FILE_PATH or explicitly pass a file path."
        )

    log.debug(f"checking wallet file path: {wallet_file_path}")

    if not os.path.exists(wallet_file_path):
        raise ValueError(f"Wallet file {wallet_file_path} does not exist.")

    # wallet used to pay for file upload
    return Wallet(wallet_file_path, api_url=api_url)