gistfile1.txt

import h5py
from filelock import FileLock, Timeout
from contracting.storage.encoder import encode, decode
from contracting import constants

# Constants
ATTR_VALUE = "value"
ATTR_BLOCK = "block"
LOCK_SUFFIX = ".lock"


def get_file_lock(file_path):
    """Retrieve a file-based lock."""
    return FileLock(f"{file_path}{LOCK_SUFFIX}")


def get_attr(file_path, group_name, attr_name):
    try:
        with h5py.File(file_path, 'r') as f:
            try:
                value = f[group_name].attrs[attr_name]
                return value.decode() if isinstance(value, bytes) else value
            except KeyError:
                return None
    except OSError:
        return None


def get_value(file_path, group_name):
    return get_attr(file_path, group_name, ATTR_VALUE)


def get_block(file_path, group_name):
    return get_attr(file_path, group_name, ATTR_BLOCK)


def get_groups(file_path):
    try:
        with h5py.File(file_path, 'r') as f:
            return list(f.keys())
    except OSError:
        return []


def set(file_path, group_name, value, blocknum, timeout=20):
    lock = get_file_lock(file_path)
    try:
        with lock.acquire(timeout=timeout):
            with h5py.File(file_path, 'a') as f:
                write_attr(f, group_name, ATTR_VALUE, value)
                write_attr(f, group_name, ATTR_BLOCK, blocknum)
    except Timeout:
        raise TimeoutError("Lock acquisition timed out")


def write_attr(file, group_name, attr_name, value):
    grp = file.require_group(group_name)
    if attr_name in grp.attrs:
        del grp.attrs[attr_name]
    if value is not None:
        grp.attrs[attr_name] = value


def delete(file_path, group_name, timeout=20):
    lock = get_file_lock(file_path)
    try:
        with lock.acquire(timeout=timeout):
            with h5py.File(file_path, 'a') as f:
                grp = f.get(group_name)
                if grp:
                    grp.attrs.pop(ATTR_VALUE, None)
                    grp.attrs.pop(ATTR_BLOCK, None)
    except Timeout:
        raise TimeoutError("Lock acquisition timed out")


def set_value_to_disk(file_path, group_name, value, block_num=None, timeout=20):
    encoded_value = encode(value) if value is not None else None
    set(file_path, group_name, encoded_value, block_num if block_num is not None else -1, timeout)


def delete_key_from_disk(file_path, group_name, timeout=20):
    delete(file_path, group_name, timeout)


def get_value_from_disk(file_path, group_name):
    raw_value = get_value(file_path, group_name)
    return decode(raw_value) if raw_value else None


def get_all_keys_from_file(file_path):
    keys = []

    def visit_func(name, node):
        keys.append(name.replace(constants.HDF5_GROUP_SEPARATOR, constants.DELIMITER))

    with h5py.File(file_path, 'r') as f:
        f.visititems(visit_func)

    return keys