Function _edit_rwlock

dvc/rwlock.py:45–67

(lock_dir, fs, hardlink)

@contextmanager
def _edit_rwlock(lock_dir, fs, hardlink):
    path = fs.join(lock_dir, RWLOCK_FILE)

    rwlock_guard = make_lock(
        fs.join(lock_dir, RWLOCK_LOCK),
        tmp_dir=lock_dir,
        hardlink_lock=hardlink,
    )
    with rwlock_guard:
        try:
            with fs.open(path, encoding="utf-8") as fobj:
                lock = SCHEMA(json.load(fobj))
        except FileNotFoundError:
            lock = SCHEMA({})
        except json.JSONDecodeError as exc:
            raise RWLockFileCorruptedError(path) from exc
        except Invalid as exc:
            raise RWLockFileFormatError(path) from exc
        lock["read"] = defaultdict(list, lock["read"])
        lock["write"] = defaultdict(dict, lock["write"])
        yield lock
        with fs.open(path, "w", encoding="utf-8") as fobj:
            json.dump(lock, fobj)


def _infos_to_str(infos):