<commit_before>import os
import json

from collections import defaultdict
from contextlib import contextmanager

from voluptuous import Schema, Invalid

from .exceptions import DvcException
from .lock import LockError
from .utils import relpath

INFO_SCHEMA = {"pid": int, "cmd": str}

SCHEMA = Schema({"write": {str: INFO_SCHEMA}, "read": {str: [INFO_SCHEMA]}})


class RWLockFileCorruptedError(DvcException):
    def __init__(self, path):
        super().__init__(
            "Unable to read RWLock-file '{}'. JSON structure is "
            "corrupted".format(relpath(path))
        )


class RWLockFileFormatError(DvcException):
    def __init__(self, path):
        super().__init__(
            "RWLock-file '{}' format error.".format(relpath(path))
        )


@contextmanager
def _edit_rwlock(lock_dir):
    path = os.path.join(lock_dir, "rwlock")
    try:
        with open(path, "r") as fobj:
            lock = json.load(fobj)
        lock = SCHEMA(lock)
    except FileNotFoundError:
        lock = SCHEMA({})
    except json.JSONDecodeError as exc:
        raise RWLockFileCorruptedError(path) from exc
    except Invalid as exc:
        raise RWLockFileFormatError(path) from exc
    lock = defaultdict(dict, lock)
    lock["read"] = defaultdict(list, lock["read"])
    lock["write"] = defaultdict(dict, lock["write"])
    yield lock
    with open(path, "w+") as fobj:
        json.dump(lock, fobj)
        # NOTE: flush and fsync to ensure that rwlock contents are saved
        fobj.flush()
        os.fsync(fobj.fileno())


def _infos_to_str(infos):
    return "\n".join(
        "  (PID {}): {}".format(info["pid"], info["cmd"]) for info in infos
    )


def _check_blockers(lock, info, *, mode, waiters):
    for path_info in waiters:
        blockers = [
            info
            for path, infos in lock[mode].items()
            if path_info.overlaps(path)
            if info not in (infos if type(infos) is list else [infos])
        ]

        if not blockers:
            continue

        raise LockError(
            "'{path}' is busy, it is being blocked by:\n"
            "{blockers}\n"<commit_msg>rwlock: require info keys to be present in schema (#3283)<commit_after>import os
import json

from collections import defaultdict
from contextlib import contextmanager

from voluptuous import Schema, Invalid, Required, Optional

from .exceptions import DvcException
from .lock import LockError
from .utils import relpath

INFO_SCHEMA = {Required("pid"): int, Required("cmd"): str}

SCHEMA = Schema(
    {
        Optional("write", default={}): {str: INFO_SCHEMA},
        Optional("read", default={}): {str: [INFO_SCHEMA]},
    }
)


class RWLockFileCorruptedError(DvcException):
    def __init__(self, path):
        super().__init__(
            "Unable to read RWLock-file '{}'. JSON structure is "
            "corrupted".format(relpath(path))
        )


class RWLockFileFormatError(DvcException):
    def __init__(self, path):
        super().__init__(
            "RWLock-file '{}' format error.".format(relpath(path))
        )


@contextmanager
def _edit_rwlock(lock_dir):
    path = os.path.join(lock_dir, "rwlock")
    try:
        with open(path, "r") 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 open(path, "w+") as fobj:
        json.dump(lock, fobj)
        # NOTE: flush and fsync to ensure that rwlock contents are saved
        fobj.flush()
        os.fsync(fobj.fileno())


def _infos_to_str(infos):
    return "\n".join(
        "  (PID {}): {}".format(info["pid"], info["cmd"]) for info in infos
    )


def _check_blockers(lock, info, *, mode, waiters):
    for path_info in waiters:
        blockers = [
            info
            for path, infos in lock[mode].items()
            if path_info.overlaps(path)
            if info not in (infos if type(infos) is list else [infos])
        ]

        if not blockers:
            continue

        raise LockError(
            "'{path}' is busy, it is being blocked by:\n"
            "{blockers}\n"
{'fraction_non_alphanumeric': 0.11904761904761904, 'fraction_numerical': 0.0009157509157509158, 'mean_word_length': 2.6499582289055974, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3569738', 'n_tokens_mistral': 1372, 'n_tokens_neox': 1359, 'n_words': 380}