Method create_helper

src/borg/archive.py:1562–1614

(self, tarinfo, status=None, type=None)

    @contextmanager
    def create_helper(self, tarinfo, status=None, type=None):
        ph = tarinfo.pax_headers
        if ph and "BORG.item.version" in ph:
            assert ph["BORG.item.version"] == "1"
            meta_bin = base64.b64decode(ph["BORG.item.meta"])
            meta_dict = msgpack.unpackb(meta_bin, object_hook=StableDict)
            item = Item(internal_dict=meta_dict)
        else:

            def s_to_ns(s):
                return safe_ns(int(float(s) * 1e9))

            # if the tar has names starting with "./", normalize them like borg create also does.
            # ./dir/file must become dir/file in the borg archive.
            normalized_path = posixpath.normpath(tarinfo.name)
            item = Item(
                path=make_path_safe(normalized_path),
                mode=tarinfo.mode | type,
                uid=tarinfo.uid,
                gid=tarinfo.gid,
                mtime=s_to_ns(tarinfo.mtime),
            )
            if tarinfo.uname:
                item.user = tarinfo.uname
            if tarinfo.gname:
                item.group = tarinfo.gname
            if ph:
                # note: for mtime this is a bit redundant as it is already done by tarfile module,
                #       but we just do it in our way to be consistent for sure.
                for name in "atime", "ctime", "mtime":
                    if name in ph:
                        ns = s_to_ns(ph[name])
                        setattr(item, name, ns)
                xattrs = StableDict()
                for key, value in ph.items():
                    if key.startswith(SCHILY_XATTR):
                        key = key.removeprefix(SCHILY_XATTR)
                        # the tarfile code gives us str keys and str values,
                        # but we need bytes keys and bytes values.
                        bkey = key.encode("utf-8", errors="surrogateescape")
                        bvalue = value.encode("utf-8", errors="surrogateescape")
                        xattrs[bkey] = bvalue
                    elif key == SCHILY_ACL_ACCESS:
                        # Process POSIX access ACL
                        item.acl_access = value.encode("utf-8", errors="surrogateescape")
                    elif key == SCHILY_ACL_DEFAULT:
                        # Process POSIX default ACL
                        item.acl_default = value.encode("utf-8", errors="surrogateescape")
                if xattrs:
                    item.xattrs = xattrs
        yield item, status
        # if we get here, "with"-block worked ok without error/exception, the item was processed ok...
        self.add_item(item, stats=self.stats)

    def process_dir(self, *, tarinfo, status, type):
        with self.create_helper(tarinfo, status, type) as (item, status):