container

Re-exports for users of the IH5 variants.

IH5Record

Bases: IH5Group

Class representing a record, which consists of a collection of immutable files.

One file is a base container (with no linked predecessor state), the remaining files are a linear sequence of patch containers.

Runtime invariants to be upheld before/after each method call (after init):

• all files of an instance are open for reading (until close() is called)
• all files in __files__ are in patch index order
• at most one file is open in writable mode (if any, it is the last one)
• modifications are possible only after create_patch was called and until commit_patch or discard_patch was called, and at no other time

Source code in src/metador_core/ih5/record.py

class IH5Record(IH5Group):
    """Class representing a record, which consists of a collection of immutable files.

    One file is a base container (with no linked predecessor state),
    the remaining files are a linear sequence of patch containers.

    Runtime invariants to be upheld before/after each method call (after __init__):

    * all files of an instance are open for reading (until `close()` is called)
    * all files in `__files__` are in patch index order
    * at most one file is open in writable mode (if any, it is the last one)
    * modifications are possible only after `create_patch` was called
        and until `commit_patch` or `discard_patch` was called, and at no other time
    """

    # Characters that may appear in a record name.
    # (to be put into regex [..] symbol braces)
    _ALLOWED_NAME_CHARS = r"A-Za-z0-9\-"

    # filenames for a record named NAME are of the shape:
    # NAME[<PATCH_INFIX>.*]?<FILE_EXT>
    # NOTE: the first symbol of these must be one NOT in ALLOWED_NAME_CHARS!
    # This constraint is needed for correctly filtering filenames
    _PATCH_INFIX = ".p"
    _FILE_EXT = ".ih5"

    # core "wrapped" objects
    __files__: List[h5py.File]

    # attributes
    _closed: bool  # True after close()
    _allow_patching: bool  # false iff opened with "r"
    _ublocks: Dict[Path, IH5UserBlock]  # in-memory copy of HDF5 user blocks

    def __new__(cls, *args, **kwargs):
        ret = super().__new__(cls)
        ret._allow_patching = True
        ret.__files__ = []
        return ret

    def __eq__(self, o) -> bool:
        if not isinstance(o, IH5Group):
            return False
        return self._record._files == o._record._files

    @property
    def _has_writable(self):
        """Return True iff an uncommitted patch exists."""
        if not self.__files__:
            return False
        f = self.__files__[-1]
        return bool(f) and f.mode == "r+"

    @classmethod
    def _is_valid_record_name(cls, name: str) -> bool:
        """Return whether a record name is valid."""
        return re.match(f"^[{cls._ALLOWED_NAME_CHARS}]+$", name) is not None

    @classmethod
    def _base_filename(cls, record_path: Path) -> Path:
        """Given a record path, return path to canonical base container name."""
        return Path(f"{record_path}{cls._FILE_EXT}")

    @classmethod
    def _infer_name(cls, record_path: Path) -> str:
        return record_path.name.split(cls._FILE_EXT)[0].split(cls._PATCH_INFIX)[0]

    def _next_patch_filepath(self) -> Path:
        """Compute filepath for the next patch based on the previous one."""
        path = Path(self.__files__[0].filename)
        parent = path.parent
        patch_index = self._ublock(-1).patch_index + 1
        res = f"{parent}/{self._infer_name(path)}{self._PATCH_INFIX}{patch_index}{self._FILE_EXT}"
        return Path(res)

    def _ublock(self, obj: Union[h5py.File, int]) -> IH5UserBlock:
        """Return the parsed user block of a container file."""
        f: h5py.File = obj if isinstance(obj, h5py.File) else self.__files__[obj]
        return self._ublocks[Path(f.filename)]

    def _set_ublock(self, obj: Union[h5py.File, int], ub: IH5UserBlock):
        f: h5py.File = obj if isinstance(obj, h5py.File) else self.__files__[obj]
        self._ublocks[Path(f.filename)] = ub

    @classmethod
    def _new_container(cls, path: Path, ub: IH5UserBlock) -> h5py.File:
        """Initialize a fresh container file with reserved user block."""
        # create if does not exist, fail if it does
        f = h5py.File(path, mode="x", userblock_size=USER_BLOCK_SIZE)
        # close to pre-fill userblock
        f.close()
        ub.save(path)
        # reopen the container file
        return h5py.File(path, "r+")

    def _check_ublock(
        self,
        filename: Union[str, Path],
        ub: IH5UserBlock,
        prev: Optional[IH5UserBlock] = None,
        check_hashsum: bool = True,
    ):
        """Check given container file.

        If `prev` block is given, assumes that `ub` is from a patch container,
        otherwise from base container.
        """
        filename = Path(filename)
        # check presence+validity of record uuid (should be the same for all)
        if ub.record_uuid != self.ih5_uuid:
            msg = "'record_uuid' inconsistent! Mixed up records?"
            raise ValueError(f"{filename}: {msg}")

        # hash must match with HDF5 content (i.e. integrity check)
        if check_hashsum and ub.hdf5_hashsum is None:
            msg = "hdf5_checksum is missing!"
            raise ValueError(f"{filename}: {msg}")
        if ub.hdf5_hashsum is not None:
            chksum = hashsum_file(filename, skip_bytes=USER_BLOCK_SIZE)
            if ub.hdf5_hashsum != chksum:
                msg = "file has been modified, stored and computed checksum are different!"
                raise ValueError(f"{filename}: {msg}")

        # check patch chain structure
        if prev is not None:
            if ub.patch_index <= prev.patch_index:
                msg = "patch container must have greater index than predecessor!"
                raise ValueError(f"{filename}: {msg}")
            if ub.prev_patch is None:
                msg = "patch must have an attribute 'prev_patch'!"
                raise ValueError(f"{filename}: {msg}")
            # claimed predecessor uuid must match with the predecessor by index
            # (can compare as strings directly, as we checked those already)
            if ub.prev_patch != prev.patch_uuid:
                msg = f"patch for {ub.prev_patch}, but predecessor is {prev.patch_uuid}"
                raise ValueError(f"{filename}: {msg}")

    def _expect_open(self):
        if self._closed:
            raise ValueError("Record is not open!")

    def _clear(self):
        """Clear all contents of the record."""
        for k in self.attrs.keys():
            del self.attrs[k]
        for k in self.keys():
            del self[k]

    def _is_empty(self) -> bool:
        """Return whether this record currently contains any data."""
        return not self.attrs.keys() and not self.keys()

    @classmethod
    def _create(cls: Type[T], record: Union[Path, str], truncate: bool = False) -> T:
        """Create a new record consisting of a base container.

        The base container is exposed as the `writable` container.
        """
        record = Path(record)  # in case it was a str
        if not cls._is_valid_record_name(record.name):
            raise ValueError(f"Invalid record name: '{record.name}'")
        path = cls._base_filename(record)

        # if overwrite flag is set, check and remove old record if present
        if truncate and path.is_file():
            cls.delete_files(record)

        # create new container
        ret = cls.__new__(cls)
        super().__init__(ret, ret)
        ret._closed = False

        ub = IH5UserBlock.create(prev=None)
        ret._ublocks = {path: ub}

        ret.__files__ = [cls._new_container(path, ub)]
        return ret

    @classmethod
    def _open(cls: Type[T], paths: List[Path], **kwargs) -> T:
        """Open a record consisting of a base container + possible set of patches.

        Expects a set of full file paths forming a valid record.
        Will throw an exception in case of a detected inconsistency.

        Will open latest patch in writable mode if it lacks a hdf5 checksum.
        """
        if not paths:
            raise ValueError("Cannot open empty list of containers!")
        allow_baseless: bool = kwargs.pop("allow_baseless", False)

        ret = cls.__new__(cls)
        super().__init__(ret, ret)
        ret._closed = False

        ret._ublocks = {Path(path): IH5UserBlock.load(path) for path in paths}
        # files, sorted  by patch index order (important!)
        # if something is wrong with the indices, this will throw an exception.
        ret.__files__ = [h5py.File(path, "r") for path in paths]
        ret.__files__.sort(key=lambda f: ret._ublock(f).patch_index)
        # ----
        has_patches: bool = len(ret.__files__) > 1

        # check containers and relationship to each other:

        # check first container (it could be a base container and it has no predecessor)
        if not allow_baseless and ret._ublock(0).prev_patch is not None:
            msg = "base container must not have attribute 'prev_patch'!"
            raise ValueError(f"{ret.__files__[0].filename}: {msg}")
        ret._check_ublock(ret.__files__[0].filename, ret._ublock(0), None, has_patches)

        # check patches except last one (with checking the hashsum)
        for i in range(1, len(ret.__files__) - 1):
            filename = ret.__files__[i].filename
            ret._check_ublock(filename, ret._ublock(i), ret._ublock(i - 1), True)
        if has_patches:  # check latest patch (without checking hashsum)
            ret._check_ublock(
                ret.__files__[-1].filename, ret._ublock(-1), ret._ublock(-2), False
            )

        # now check whether the last container (patch or base or whatever) has a checksum
        if ret._ublock(-1).hdf5_hashsum is None:
            if kwargs.pop("reopen_incomplete_patch", False):
                # if opening in writable mode, allow to complete the patch
                f = ret.__files__[-1]
                path = ret.__files__[-1].filename
                f.close()
                ret.__files__[-1] = h5py.File(Path(path), "r+")

        # additional sanity check: container uuids must be all distinct
        cn_uuids = {ret._ublock(f).patch_uuid for f in ret.__files__}
        if len(cn_uuids) != len(ret.__files__):
            raise ValueError("Some patch_uuid is not unique, invalid file set!")
        # all looks good
        return ret

    # ---- public attributes and interface ----

    @property
    def ih5_uuid(self) -> UUID:
        """Return the common record UUID of the set of containers."""
        return self._ublock(0).record_uuid

    @property
    def ih5_files(self) -> List[Path]:
        """List of container filenames this record consists of."""
        return [Path(f.filename) for f in self.__files__]

    @property
    def ih5_meta(self) -> List[IH5UserBlock]:
        """Return user block metadata, in container patch order."""
        return [self._ublock(i).copy() for i in range(len(self.__files__))]

    @classmethod
    def find_files(cls, record: Path) -> List[Path]:
        """Return file names that look like they belong to the same record.

        This operation is based on purely syntactic pattern matching on file names
        that follow the default naming convention.

        Given a path `/foo/bar`, will find all containers in directory
        `/foo` whose name starts with `bar` followed by the correct file extension(s),
        such as `/foo/bar.ih5` and `/foo/bar.p1.ih5`.
        """
        record = Path(record)  # in case it was a str
        if not cls._is_valid_record_name(record.name):
            raise ValueError(f"Invalid record name: '{record.name}'")

        globstr = f"{record.name}*{cls._FILE_EXT}"  # rough wildcard pattern
        # filter out possible false positives (i.e. foobar* matching foo* as well)
        return [
            p
            for p in record.parent.glob(globstr)
            if re.match(f"^{record.name}[^{cls._ALLOWED_NAME_CHARS}]", p.name)
        ]

    @classmethod
    def list_records(cls, dir: Path) -> List[Path]:
        """Return paths of records found in the given directory.

        Will NOT recurse into subdirectories.

        This operation is based on purely syntactic pattern matching on file names
        that follow the default naming convention (i.e. just as `find_files`).

        Returned paths can be used as-is for opening the (supposed) record.
        """
        dir = Path(dir)  # in case it was a str
        if not dir.is_dir():
            raise ValueError(f"'{dir}' is not a directory")

        ret = []
        namepat = f"[{cls._ALLOWED_NAME_CHARS}]+(?=[^{cls._ALLOWED_NAME_CHARS}])"
        for p in dir.glob(f"*{cls._FILE_EXT}"):
            if m := re.match(namepat, p.name):
                ret.append(m[0])
        return list(map(lambda name: dir / name, set(ret)))

    def __init__(
        self, record: Union[str, Path, List[Path]], mode: OpenMode = "r", **kwargs
    ):
        """Open or create a record.

        This method uses `find_files` to infer the correct set of files syntactically.

        The open mode semantics are the same as for h5py.File.

        If the mode is 'r', then creating, committing or discarding patches is disabled.

        If the mode is 'a' or 'r+', then a new patch will be created in case the latest
        patch has already been committed.
        """
        super().__init__(self)

        if isinstance(record, list):
            if mode[0] == "w" or mode == "x":
                raise ValueError("Pass a prefix path for creating or overwriting!")
            paths = record
        else:
            paths = None
            path: Path = Path(record)

        if mode not in _OPEN_MODES:
            raise ValueError(f"Unknown file open mode: {mode}")

        if mode[0] == "w" or mode == "x":
            # create new or overwrite to get new
            ret = self._create(path, truncate=(mode == "w"))
            self.__dict__.update(ret.__dict__)
            return

        if mode == "a" or mode[0] == "r":
            if not paths:  # user passed a path prefix -> find files
                paths = self.find_files(path)  # type: ignore

            if not paths:  # no files were found
                if mode != "a":  # r/r+ need existing containers
                    raise FileNotFoundError(f"No files found for record: {path}")
                else:  # 'a' means create new if not existing (will be writable)
                    ret = self._create(path, truncate=False)
                    self.__dict__.update(ret.__dict__)
                    return

            # open existing (will be ro if everything is fine, writable if latest patch was uncommitted)
            want_rw = mode != "r"
            ret = self._open(paths, reopen_incomplete_patch=want_rw, **kwargs)
            self.__dict__.update(ret.__dict__)
            self._allow_patching = want_rw

            if want_rw and not self._has_writable:
                # latest patch was completed correctly -> make writable by creating new patch
                self.create_patch()

    @property
    def mode(self) -> Literal["r", "r+"]:
        return "r+" if self._allow_patching else "r"

    def close(self, commit: bool = True) -> None:
        """Close all files that belong to this record.

        If there exists an uncommited patch, it will be committed
        (unless `commit` is set to false).

        After this, the object may not be used anymore.
        """
        if self._closed:
            return  # nothing to do

        if self._has_writable and commit:
            self.commit_patch()
        for f in self.__files__:
            f.close()
        self.__files__ = []
        self._closed = True

    def _expect_not_ro(self):
        if self.mode == "r":
            raise ValueError("The container is opened as read-only!")

    def create_patch(self) -> None:
        """Create a new patch in order to update the record."""
        self._expect_open()
        self._expect_not_ro()
        if self._has_writable:
            raise ValueError("There already exists a writable container, commit first!")

        path = self._next_patch_filepath()
        ub = IH5UserBlock.create(prev=self._ublock(-1))
        self.__files__.append(self._new_container(path, ub))
        self._ublocks[path] = ub

    def _delete_latest_container(self) -> None:
        """Discard the current writable container (patch or base)."""
        cfile = self.__files__.pop()
        fn = cfile.filename
        del self._ublocks[Path(fn)]
        cfile.close()
        Path(fn).unlink()

    def discard_patch(self) -> None:
        """Discard the current incomplete patch container."""
        self._expect_open()
        self._expect_not_ro()
        if not self._has_writable:
            raise ValueError("No patch to discard!")
        if len(self.__files__) == 1:
            raise ValueError("Cannot discard base container! Just delete the file!")
            # reason: the base container provides record_uuid,
            # destroying it makes this object inconsistent / breaks invariants
            # so if this is done, it should not be used anymore.
        return self._delete_latest_container()

    def commit_patch(self, **kwargs) -> None:
        """Complete the current writable container (base or patch) for the record.

        Will perform checks on the new container and throw an exception on failure.

        After committing the patch is completed and cannot be edited anymore, so
        any further modifications must go into a new patch.
        """
        if kwargs:
            raise ValueError(f"Unknown keyword arguments: {kwargs}")

        self._expect_open()
        self._expect_not_ro()
        if not self._has_writable:
            raise ValueError("No patch to commit!")
        cfile = self.__files__[-1]
        filepath = Path(cfile.filename)
        cfile.close()  # must close it now, as we will write outside of HDF5 next

        # compute checksum, write user block
        chksum = hashsum_file(filepath, skip_bytes=USER_BLOCK_SIZE)
        self._ublocks[filepath].hdf5_hashsum = QualHashsumStr(chksum)
        self._ublocks[filepath].save(filepath)

        # reopen the container file now as read-only
        self.__files__[-1] = h5py.File(filepath, "r")

    def _fixes_after_merge(self, merged_file, ub):
        """Run hook for subclasses into merge process.

        The method is called after creating the merged container, but before
        updating its user block on disk.

        The passed userblock is a prepared userblock with updated HDF5 hashsum for the
        merged container and adapted prev_patch field, as will it be written to the file.
        Additional changes done to it in-place will be included.

        The passed filename can be used to perform additional necessary actions.
        """

    def merge_files(self, target: Path) -> Path:
        """Given a path with a record name, merge current record into new container.

        Returns new resulting container.
        """
        self._expect_open()
        if self._has_writable:
            raise ValueError("Cannot merge, please commit or discard your changes!")

        with type(self)(target, "x") as ds:
            source_node = self["/"]
            target_node = ds["/"]
            for k, v in source_node.attrs.items():  # copy root attributes
                target_node.attrs[k] = v
            for name in source_node.keys():  # copy each entity (will recurse)
                h5_copy_from_to(source_node[name], target_node, name)

            cfile = ds.ih5_files[0]  # store filename to override userblock afterwards

        # compute new merged userblock
        ub = self._ublock(-1).copy(update={"prev_patch": self._ublock(0).prev_patch})
        # update hashsum with saved new merged hdf5 payload
        chksum = hashsum_file(cfile, skip_bytes=USER_BLOCK_SIZE)
        ub.hdf5_hashsum = QualHashsumStr(chksum)

        self._fixes_after_merge(cfile, ub)  # for subclass hooks

        self._set_ublock(-1, ub)
        ub.save(cfile)
        return cfile

    @classmethod
    def delete_files(cls, record: Path):
        """Irreversibly(!) delete all containers matching the record path.

        This object is invalid after this operation.
        """
        for file in cls.find_files(record):
            file.unlink()

    def __repr__(self):
        return f"<IH5 record (mode {self.mode}) {self.__files__}>"

    # ---- context manager support (i.e. to use `with`) ----

    def __enter__(self):
        return self

    def __exit__(self, ex_type, ex_value, ex_traceback):
        # this will ensure that commit_patch() is called and the files are closed
        self.close()

ih5_uuid property

ih5_uuid: UUID

Return the common record UUID of the set of containers.

ih5_files property

ih5_files: List[Path]

List of container filenames this record consists of.

ih5_meta property

ih5_meta: List[IH5UserBlock]

Return user block metadata, in container patch order.

find_files classmethod

find_files(record: Path) -> List[Path]

Return file names that look like they belong to the same record.

This operation is based on purely syntactic pattern matching on file names that follow the default naming convention.

Given a path /foo/bar, will find all containers in directory /foo whose name starts with bar followed by the correct file extension(s), such as /foo/bar.ih5 and /foo/bar.p1.ih5.

Source code in src/metador_core/ih5/record.py

@classmethod
def find_files(cls, record: Path) -> List[Path]:
    """Return file names that look like they belong to the same record.

    This operation is based on purely syntactic pattern matching on file names
    that follow the default naming convention.

    Given a path `/foo/bar`, will find all containers in directory
    `/foo` whose name starts with `bar` followed by the correct file extension(s),
    such as `/foo/bar.ih5` and `/foo/bar.p1.ih5`.
    """
    record = Path(record)  # in case it was a str
    if not cls._is_valid_record_name(record.name):
        raise ValueError(f"Invalid record name: '{record.name}'")

    globstr = f"{record.name}*{cls._FILE_EXT}"  # rough wildcard pattern
    # filter out possible false positives (i.e. foobar* matching foo* as well)
    return [
        p
        for p in record.parent.glob(globstr)
        if re.match(f"^{record.name}[^{cls._ALLOWED_NAME_CHARS}]", p.name)
    ]

list_records classmethod

list_records(dir: Path) -> List[Path]

Return paths of records found in the given directory.

Will NOT recurse into subdirectories.

This operation is based on purely syntactic pattern matching on file names that follow the default naming convention (i.e. just as find_files).

Returned paths can be used as-is for opening the (supposed) record.

Source code in src/metador_core/ih5/record.py

@classmethod
def list_records(cls, dir: Path) -> List[Path]:
    """Return paths of records found in the given directory.

    Will NOT recurse into subdirectories.

    This operation is based on purely syntactic pattern matching on file names
    that follow the default naming convention (i.e. just as `find_files`).

    Returned paths can be used as-is for opening the (supposed) record.
    """
    dir = Path(dir)  # in case it was a str
    if not dir.is_dir():
        raise ValueError(f"'{dir}' is not a directory")

    ret = []
    namepat = f"[{cls._ALLOWED_NAME_CHARS}]+(?=[^{cls._ALLOWED_NAME_CHARS}])"
    for p in dir.glob(f"*{cls._FILE_EXT}"):
        if m := re.match(namepat, p.name):
            ret.append(m[0])
    return list(map(lambda name: dir / name, set(ret)))

__init__

__init__(
    record: Union[str, Path, List[Path]],
    mode: OpenMode = "r",
    **kwargs: OpenMode
)

Open or create a record.

This method uses find_files to infer the correct set of files syntactically.

The open mode semantics are the same as for h5py.File.

If the mode is 'r', then creating, committing or discarding patches is disabled.

If the mode is 'a' or 'r+', then a new patch will be created in case the latest patch has already been committed.

Source code in src/metador_core/ih5/record.py

def __init__(
    self, record: Union[str, Path, List[Path]], mode: OpenMode = "r", **kwargs
):
    """Open or create a record.

    This method uses `find_files` to infer the correct set of files syntactically.

    The open mode semantics are the same as for h5py.File.

    If the mode is 'r', then creating, committing or discarding patches is disabled.

    If the mode is 'a' or 'r+', then a new patch will be created in case the latest
    patch has already been committed.
    """
    super().__init__(self)

    if isinstance(record, list):
        if mode[0] == "w" or mode == "x":
            raise ValueError("Pass a prefix path for creating or overwriting!")
        paths = record
    else:
        paths = None
        path: Path = Path(record)

    if mode not in _OPEN_MODES:
        raise ValueError(f"Unknown file open mode: {mode}")

    if mode[0] == "w" or mode == "x":
        # create new or overwrite to get new
        ret = self._create(path, truncate=(mode == "w"))
        self.__dict__.update(ret.__dict__)
        return

    if mode == "a" or mode[0] == "r":
        if not paths:  # user passed a path prefix -> find files
            paths = self.find_files(path)  # type: ignore

        if not paths:  # no files were found
            if mode != "a":  # r/r+ need existing containers
                raise FileNotFoundError(f"No files found for record: {path}")
            else:  # 'a' means create new if not existing (will be writable)
                ret = self._create(path, truncate=False)
                self.__dict__.update(ret.__dict__)
                return

        # open existing (will be ro if everything is fine, writable if latest patch was uncommitted)
        want_rw = mode != "r"
        ret = self._open(paths, reopen_incomplete_patch=want_rw, **kwargs)
        self.__dict__.update(ret.__dict__)
        self._allow_patching = want_rw

        if want_rw and not self._has_writable:
            # latest patch was completed correctly -> make writable by creating new patch
            self.create_patch()

close

close(commit: bool = True) -> None

Close all files that belong to this record.

If there exists an uncommited patch, it will be committed (unless commit is set to false).

After this, the object may not be used anymore.

Source code in src/metador_core/ih5/record.py

def close(self, commit: bool = True) -> None:
    """Close all files that belong to this record.

    If there exists an uncommited patch, it will be committed
    (unless `commit` is set to false).

    After this, the object may not be used anymore.
    """
    if self._closed:
        return  # nothing to do

    if self._has_writable and commit:
        self.commit_patch()
    for f in self.__files__:
        f.close()
    self.__files__ = []
    self._closed = True

create_patch

create_patch() -> None

Create a new patch in order to update the record.

Source code in src/metador_core/ih5/record.py

def create_patch(self) -> None:
    """Create a new patch in order to update the record."""
    self._expect_open()
    self._expect_not_ro()
    if self._has_writable:
        raise ValueError("There already exists a writable container, commit first!")

    path = self._next_patch_filepath()
    ub = IH5UserBlock.create(prev=self._ublock(-1))
    self.__files__.append(self._new_container(path, ub))
    self._ublocks[path] = ub

discard_patch

discard_patch() -> None

Discard the current incomplete patch container.

Source code in src/metador_core/ih5/record.py

def discard_patch(self) -> None:
    """Discard the current incomplete patch container."""
    self._expect_open()
    self._expect_not_ro()
    if not self._has_writable:
        raise ValueError("No patch to discard!")
    if len(self.__files__) == 1:
        raise ValueError("Cannot discard base container! Just delete the file!")
        # reason: the base container provides record_uuid,
        # destroying it makes this object inconsistent / breaks invariants
        # so if this is done, it should not be used anymore.
    return self._delete_latest_container()

commit_patch

commit_patch(**kwargs) -> None

Complete the current writable container (base or patch) for the record.

Will perform checks on the new container and throw an exception on failure.

After committing the patch is completed and cannot be edited anymore, so any further modifications must go into a new patch.

Source code in src/metador_core/ih5/record.py

def commit_patch(self, **kwargs) -> None:
    """Complete the current writable container (base or patch) for the record.

    Will perform checks on the new container and throw an exception on failure.

    After committing the patch is completed and cannot be edited anymore, so
    any further modifications must go into a new patch.
    """
    if kwargs:
        raise ValueError(f"Unknown keyword arguments: {kwargs}")

    self._expect_open()
    self._expect_not_ro()
    if not self._has_writable:
        raise ValueError("No patch to commit!")
    cfile = self.__files__[-1]
    filepath = Path(cfile.filename)
    cfile.close()  # must close it now, as we will write outside of HDF5 next

    # compute checksum, write user block
    chksum = hashsum_file(filepath, skip_bytes=USER_BLOCK_SIZE)
    self._ublocks[filepath].hdf5_hashsum = QualHashsumStr(chksum)
    self._ublocks[filepath].save(filepath)

    # reopen the container file now as read-only
    self.__files__[-1] = h5py.File(filepath, "r")

merge_files

merge_files(target: Path) -> Path

Given a path with a record name, merge current record into new container.

Returns new resulting container.

Source code in src/metador_core/ih5/record.py

def merge_files(self, target: Path) -> Path:
    """Given a path with a record name, merge current record into new container.

    Returns new resulting container.
    """
    self._expect_open()
    if self._has_writable:
        raise ValueError("Cannot merge, please commit or discard your changes!")

    with type(self)(target, "x") as ds:
        source_node = self["/"]
        target_node = ds["/"]
        for k, v in source_node.attrs.items():  # copy root attributes
            target_node.attrs[k] = v
        for name in source_node.keys():  # copy each entity (will recurse)
            h5_copy_from_to(source_node[name], target_node, name)

        cfile = ds.ih5_files[0]  # store filename to override userblock afterwards

    # compute new merged userblock
    ub = self._ublock(-1).copy(update={"prev_patch": self._ublock(0).prev_patch})
    # update hashsum with saved new merged hdf5 payload
    chksum = hashsum_file(cfile, skip_bytes=USER_BLOCK_SIZE)
    ub.hdf5_hashsum = QualHashsumStr(chksum)

    self._fixes_after_merge(cfile, ub)  # for subclass hooks

    self._set_ublock(-1, ub)
    ub.save(cfile)
    return cfile

delete_files classmethod

delete_files(record: Path)

Irreversibly(!) delete all containers matching the record path.

This object is invalid after this operation.

Source code in src/metador_core/ih5/record.py

@classmethod
def delete_files(cls, record: Path):
    """Irreversibly(!) delete all containers matching the record path.

    This object is invalid after this operation.
    """
    for file in cls.find_files(record):
        file.unlink()

IH5Dataset

Bases: IH5Node

IH5Node representing a h5py.Dataset, i.e. a leaf of the tree.

Source code in src/metador_core/ih5/overlay.py

class IH5Dataset(IH5Node):
    """`IH5Node` representing a `h5py.Dataset`, i.e. a leaf of the tree."""

    def __init__(self, files, gpath, creation_idx):
        super().__init__(files, gpath, creation_idx)

    def copy_into_patch(self):
        """Copy the most recent value at this path into the current patch.

        This is useful e.g. for editing inside a complex value, such as an array.
        """
        self._guard_open()
        self._guard_read_only()
        if self._cidx == self._last_idx:
            raise ValueError("Cannot copy, this node is already from latest patch!")
        # copy value from older container to current patch
        self._files[-1][self._gpath] = self[()]

    # h5py-like interface
    @property
    def name(self) -> str:
        return self._gpath

    @property
    def file(self) -> IH5Record:
        return self._record

    @property
    def parent(self) -> IH5Group:
        return self._record[self._parent_path()]

    @property
    def attrs(self) -> IH5AttributeManager:
        self._guard_open()
        return IH5AttributeManager(self._record, self._gpath, self._cidx)

    # this one is also needed to work with H5DatasetLike
    @property
    def ndim(self) -> int:
        return self._files[self._cidx][self._gpath].ndim  # type: ignore

    # for a dataset, instead of paths the numpy data is indexed. at this level
    # the patching mechanism ends, so it's just passing through to h5py

    def __getitem__(self, key):
        # just pass through dataset indexing to underlying dataset
        self._guard_open()
        return self._files[self._cidx][self._gpath][key]  # type: ignore

    def __setitem__(self, key, val):
        self._guard_open()
        self._guard_read_only()
        if self._cidx != self._last_idx:
            raise ValueError(f"Cannot set '{key}', node is not from the latest patch!")
        # if we're in the latest patch, allow writing as usual (pass through)
        self._files[-1][self._gpath][key] = val  # type: ignore

copy_into_patch

copy_into_patch()

Copy the most recent value at this path into the current patch.

This is useful e.g. for editing inside a complex value, such as an array.

Source code in src/metador_core/ih5/overlay.py

def copy_into_patch(self):
    """Copy the most recent value at this path into the current patch.

    This is useful e.g. for editing inside a complex value, such as an array.
    """
    self._guard_open()
    self._guard_read_only()
    if self._cidx == self._last_idx:
        raise ValueError("Cannot copy, this node is already from latest patch!")
    # copy value from older container to current patch
    self._files[-1][self._gpath] = self[()]

IH5AttributeManager

Bases: IH5InnerNode

IH5Node representing an h5py.AttributeManager.

Source code in src/metador_core/ih5/overlay.py

class IH5AttributeManager(IH5InnerNode):
    """`IH5Node` representing an `h5py.AttributeManager`."""

    def __init__(self, files, gpath, creation_idx):
        super().__init__(files, gpath, creation_idx, True)

    def __setitem__(self, key: str, val):
        self._guard_open()
        self._guard_read_only()
        self._guard_key(key)
        self._guard_value(val)

        # if path does not exist in current patch, just create "virtual node"
        if self._gpath not in self._files[-1]:
            self._files[-1].create_group(self._gpath)
        # deletion marker at `key` (if set) is overwritten automatically here
        # so no need to worry about removing it before assigning `val`
        self._files[-1][self._gpath].attrs[key] = val

    def __delitem__(self, key: str):
        self._guard_open()
        self._guard_read_only()
        self._guard_key(key)
        # remove the entity if it is found in newest container,
        # mark the path as deleted if doing a patch and not working on base container
        if self._expect_real_item_idx(key) == self._last_idx:
            del self._files[-1][self._gpath].attrs[key]
        if len(self._files) > 1:  # is a patch?
            if self._gpath not in self._files[-1]:  # no node at path in latest?
                self._files[-1].create_group(self._gpath)  # create "virtual" node
            self._files[-1][self._gpath].attrs[key] = DEL_VALUE  # mark deleted

IH5Group

Bases: IH5InnerNode

IH5Node representing a h5py.Group.

Source code in src/metador_core/ih5/overlay.py

class IH5Group(IH5InnerNode):
    """`IH5Node` representing a `h5py.Group`."""

    def _require_node(self, name: str, node_type: Type[T]) -> Optional[T]:
        # helper for require_{group|dataset}
        grp = self.get(name)
        if isinstance(grp, node_type):
            return grp
        if grp is not None:
            msg = f"Incompatible object ({type(grp).__name__}) already exists"
            raise TypeError(msg)
        return None

    def __init__(self, record, gpath: str = "/", creation_idx: Optional[int] = None):
        if gpath == "/":
            creation_idx = 0
        if creation_idx is None:
            raise ValueError("Need creation_idx for path != '/'!")
        super().__init__(record, gpath, creation_idx, False)

    def _create_virtual(self, path: str) -> bool:
        nodes = self._node_seq(path)
        path = self._abs_path(path)
        if (
            nodes[-1]._gpath == path
            and nodes[-1]._cidx == self._last_idx
            and not _node_is_del_mark(nodes[-1])
        ):
            return False  # something at that path in most recent container exists

        # most recent entity is a deletion marker or not existing?
        if nodes[-1]._gpath != path or _node_is_del_mark(nodes[-1]):
            suf_segs = nodes[-1]._rel_path(path).split("/")
            # create "overwrite" group in most recent patch...
            self.create_group(f"{nodes[-1]._gpath}/{suf_segs[0]}")
            # ... and create (nested) virtual group node(s), if needed
            if len(suf_segs) > 1:
                self._files[-1].create_group(path)

        return True

    # h5py-like interface

    def __setitem__(self, path: str, value):
        return self.create_dataset(path, data=value)

    def __delitem__(self, key: str):
        self._guard_open()
        self._guard_read_only()
        self._guard_key(key)
        self._expect_real_item_idx(key)
        # remove the entity if it is found in newest container,
        # mark the path as deleted if doing a patch and not working on base container
        path = self._abs_path(key)
        if path in self._files[-1]:
            del self._files[-1][path]
        if len(self._files) > 1:  # has patches? mark deleted (instead of real delete)
            self._files[-1][path] = DEL_VALUE

    @property
    def name(self) -> str:
        return self._gpath

    @property
    def file(self):  # -> IH5Record
        return self._record

    @property
    def parent(self) -> IH5Group:
        return self._record[self._parent_path()]

    @property
    def attrs(self) -> IH5AttributeManager:
        self._guard_open()
        return IH5AttributeManager(self._record, self._gpath, self._cidx)

    def create_group(self, name: str) -> IH5Group:
        self._guard_open()
        self._guard_read_only()

        path = self._abs_path(name)
        nodes = self._node_seq(path)
        if not isinstance(nodes[-1], IH5Group):
            raise ValueError(f"Cannot create group, {nodes[-1]._gpath} is a dataset!")
        if nodes[-1]._gpath == path:
            raise ValueError("Cannot create group, it already exists!")

        # remove "deleted" marker, if set at current path in current patch container
        if path in self._files[-1] and _node_is_del_mark(self._files[-1][path]):
            del self._files[-1][path]
        # create group (or fail if something else exists there already)
        self._files[-1].create_group(path)
        # if this is a patch: mark as non-virtual, i.e. "overwrite" with empty group
        # because the intent here is to "create", not update something.
        if len(self._files) > 1:
            self._files[-1][path].attrs[SUBST_KEY] = h5py.Empty(None)

        return IH5Group(self._record, path, self._last_idx)

    def create_dataset(
        self, path: str, shape=None, dtype=None, data=None, **kwargs
    ) -> IH5Dataset:
        self._guard_open()
        self._guard_read_only()
        self._guard_key(path)
        self._guard_value(data)

        if unknown_kwargs := set(kwargs.keys()) - {"compression", "compression_opts"}:
            raise ValueError(f"Unkown kwargs: {unknown_kwargs}")

        path = self._abs_path(path)
        fidx = self._find(path)
        if fidx is not None:
            prev_val = self._get_child(path, fidx)
            if isinstance(prev_val, (IH5Group, IH5Dataset)):
                raise ValueError("Path exists, in order to replace - delete first!")

        if path in self._files[-1] and _node_is_del_mark(
            self._get_child_raw(path, self._last_idx)
        ):
            # remove deletion marker in latest patch, if set
            del self._files[-1][path]
        elif path not in self._files[-1]:
            # create path and overwrite-group in latest patch
            self._create_virtual(path)
            assert path in self._files[-1]
            del self._files[-1][path]

        self._files[-1].create_dataset(  # actually create it, finally
            path, shape=shape, dtype=dtype, data=data, **kwargs
        )
        return IH5Dataset(self._record, path, self._last_idx)

    def require_group(self, name: str) -> IH5Group:
        if (n := self._require_node(name, IH5Group)) is not None:
            return n  # existing group
        return self.create_group(name)

    def require_dataset(self, name: str, *args, **kwds) -> IH5Dataset:
        if (n := self._require_node(name, IH5Dataset)) is not None:
            # TODO: check dimensions etc, copy into patch if it fits
            return n
        return self.create_dataset(name, *args, **kwds)

    def copy(self, source: CopySource, dest: CopyDest, **kwargs):
        src_node = self[source] if isinstance(source, str) else source
        name: str = kwargs.pop("name", src_node.name.split("/")[-1])
        dst_name: str
        if isinstance(dest, str):
            # if dest is a path, ignore inferred/passed name
            segs = self._abs_path(dest).split("/")
            dst_group = self.require_group("/".join(segs[:-1]) or "/")
            dst_name = segs[-1]
        else:
            # given dest is a group node, use inferred/passed name

            dst_group = dest if dest.name != "/" else dest["/"]  # *
            # * ugly workaround for treating files as groups in the copy method

            dst_name = name
        return h5_copy_from_to(src_node, cast(Any, dst_group), dst_name, **kwargs)

    def move(self, source: str, dest: str):
        self.copy(source, dest)
        del self[source]

    def visititems(self, func: Callable[[str, object], Optional[Any]]) -> Any:
        self._guard_open()
        stack = list(reversed(self._get_children()))
        while stack:
            curr = stack.pop()
            val = func(self._rel_path(curr._gpath), curr)
            if val is not None:
                return val
            if isinstance(curr, IH5Group):
                stack += reversed(curr._get_children())

    def visit(self, func: Callable[[str], Optional[Any]]) -> Any:
        return self.visititems(lambda x, _: func(x))