hexdoc.utils

53class IProperty(Protocol[_T_covariant]):
54    def __get__(
55        self,
56        __instance: Any,
57        __owner: type | None = None,
58        /,
59    ) -> _T_covariant: ...

24class InheritType(Enum):
25    """Type of Inherit, a singleton representing a value that should be inherited."""
26
27    _token = 0
28
29    def __str__(self):
30        return "Inherit"

 8class NoValueType(Enum):
 9    """Type of NoValue, a singleton representing the value of a nonexistent dict key."""
10
11    _token = 0
12
13    def __str__(self):
14        return "NoValue"
15
16    def __bool__(self) -> Literal[False]:
17        return False

 76class PydanticOrderedSet(OrderedSet[_T]):
 77    def __init__(self, initial: OrderedSetInitializer[_T] | None = None):
 78        super().__init__(initial or [])
 79
 80    @classmethod
 81    def collect(cls, *initial: _T):
 82        return cls(initial)
 83
 84    @classmethod
 85    def __get_pydantic_core_schema__(
 86        cls,
 87        source: type[Any],
 88        handler: GetCoreSchemaHandler,
 89    ) -> core_schema.CoreSchema:
 90        type_arg: Any
 91        match len(type_args := get_args(source)):
 92            case 0:
 93                type_arg = Any
 94            case 1:
 95                type_arg = type_args[0]
 96            case n:
 97                raise ValueError(f"Expected 0 or 1 type args, got {n}: {type_args}")
 98
 99        return core_schema.union_schema(
100            [
101                core_schema.is_instance_schema(cls),
102                cls._get_non_instance_schema(type_arg, handler),
103            ],
104            serialization=cls._get_ser_schema(type_arg, handler),
105        )
106
107    @classmethod
108    def _get_non_instance_schema(
109        cls,
110        type_arg: type[Any],
111        handler: GetCoreSchemaHandler,
112    ) -> core_schema.CoreSchema:
113        # validate from OrderedSetInitializer
114        return core_schema.no_info_after_validator_function(
115            function=PydanticOrderedSet,
116            schema=handler.generate_schema(OrderedSetInitializer[type_arg]),
117        )
118
119    @classmethod
120    def _get_ser_schema(
121        cls,
122        type_arg: type[Any],
123        handler: GetCoreSchemaHandler,
124    ) -> core_schema.SerSchema:
125        # serialize to list
126        return core_schema.plain_serializer_function_ser_schema(
127            function=cls._get_items,
128        )
129
130    def _get_items(self):
131        return self.items

31class Sortable(ABC):
32    """ABC for classes which can be sorted."""
33
34    @property
35    @abstractmethod
36    def _cmp_key(self) -> Any: ...
37
38    def __lt__(self, other: Any) -> bool:
39        if isinstance(other, Sortable):
40            return self._cmp_key < other._cmp_key
41        return NotImplemented

65@unique
66class TryGetEnum(Enum):
67    @classmethod
68    def get(cls, value: Any):
69        try:
70            return cls(value)
71        except ValueError:
72            return None

14class ValidationContext:
15    @classproperty
16    @classmethod
17    def context_key(cls) -> str:
18        return str(cls)
19
20    @overload
21    @classmethod
22    def of(cls, info: ValidationInfo, /) -> Self: ...
23
24    @overload
25    @classmethod
26    def of(cls, context: dict[str, Any] | Context, /) -> Self: ...
27
28    @overload
29    @classmethod
30    def of(cls, source: ContextSource, /) -> Self: ...
31
32    @classmethod
33    def of(cls, source: ContextSource, /) -> Self:
34        match source:
35            case dict() | Context():
36                pass
37            case _:
38                source = cast_or_raise(source.context, dict)
39        return cast_or_raise(source[cls.context_key], cls)
40
41    def add_to_context(self, context: dict[str, Any], overwrite: bool = False):
42        return add_to_context(context, self.context_key, self, overwrite)

45def add_to_context(
46    context: dict[str, Any],
47    key: str,
48    value: Any,
49    overwrite: bool = False,
50):
51    if not overwrite and key in context:
52        raise KeyError(f"Key {key} for {value} already exists in context")
53    context[key] = value

56def cast_context(source: ContextSource) -> dict[str, Any]:
57    """Wrapper for `typing.cast` to simplify passing `ContextSource` to validation
58    methods. This is a lie to keep the type checker happy."""
59    return cast(dict[str, Any], source)

52def cast_or_raise(
53    val: Any,
54    class_or_tuple: type[_T] | tuple[type[_T], ...],
55    message: str | None = None,
56) -> _T:
57    assert isinstance_or_raise(val, class_or_tuple, message)
58    return val

153def clamping_validator(lower: float | None, upper: float | None):
154    def validator(value: float):
155        lower_ = lower if lower is not None else value
156        upper_ = upper if upper is not None else value
157        return max(lower_, min(upper_, value))
158
159    return AfterValidator(validator)

24def classproperty(
25    func: Callable[[type[_T_cv]], _R_co],
26) -> ClassPropertyDescriptor[_T_cv, _R_co]:
27    if isinstance(func, classmethod):
28        return ClassPropertyDescriptor(func)
29    return ClassPropertyDescriptor(classmethod(func))

27def decode_and_flatten_json_dict(data: str) -> dict[str, str]:
28    # replace `\<LF>       foobar` with `\<LF>foobar`
29    data = re.sub(r"\\\n\s*", "\\\n", data)
30
31    # decode and flatten
32    decoded = decode_json_dict(data)
33    return _flatten_inner(decoded, "")

15def decode_json_dict(data: str | bytes) -> JSONDict:
16    match data:
17        case str():
18            decoded = pyjson5.decode(data)
19        case _:
20            decoded = pyjson5.decode_utf8(data)
21    assert isinstance_or_raise(decoded, dict)
22    return cast(JSONDict, decoded)

 7def git_root(cwd: str | Path) -> Path:
 8    return commands.run(
 9        ["git", "rev-parse", "--show-toplevel"],
10        cwd=cwd,
11        type=Path,
12    )

13def isinstance_or_raise(
14    val: Any,
15    class_or_tuple: type[_T] | tuple[type[_T], ...],
16    message: str | None = None,
17) -> TypeGuard[_T]:
18    """Usage: `assert isinstance_or_raise(val, str)`
19
20    message placeholders: `{expected}`, `{actual}`, `{value}`
21    """
22
23    # convert generic types into the origin type
24    if not isinstance(class_or_tuple, tuple):
25        class_or_tuple = (class_or_tuple,)
26    ungenericed_classes = tuple(get_origin(t) or t for t in class_or_tuple)
27
28    if not isinstance(val, ungenericed_classes):
29        # just in case the caller messed up the message formatting
30        subs = {
31            "expected": list(class_or_tuple),
32            "actual": type(val),
33            "value": val,
34        }
35
36        if logger.getEffectiveLevel() >= logging.WARNING:
37            default_message = _DEFAULT_MESSAGE_SHORT
38        else:
39            default_message = _DEFAULT_MESSAGE_LONG
40
41        if message is None:
42            raise TypeError(default_message.format(**subs))
43
44        try:
45            raise TypeError(message.format(**subs))
46        except KeyError:
47            raise TypeError(default_message.format(**subs))
48
49    return True

21def listify(f: Callable[_P, Iterator[_T]]) -> Callable[_P, list[_T]]:
22    @functools.wraps(f)
23    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> list[_T]:
24        return list(f(*args, **kwargs))
25
26    return wrapper

134def load_toml_with_placeholders(path: Path) -> TOMLDict:
135    data = tomllib.loads(path.read_text("utf-8"))
136    fill_placeholders(data)
137    return data

 9def must_yield_something(f: Callable[_P, Iterator[_T]]) -> Callable[_P, Iterator[_T]]:
10    """Raises StopIteration if the wrapped iterator doesn't yield anything."""
11
12    @functools.wraps(f)
13    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Iterator[_T]:
14        iterator = f(*args, **kwargs)
15        yield next(iterator)
16        yield from iterator
17
18    return wrapper

13def relative_path_root(path: Path):
14    return set_contextvar(_relative_path_root, path)

26def replace_suffixes(path: Path, suffix: str) -> Path:
27    """Replaces all suffixes of a path. This is helpful because `path.with_suffix()`
28    only replaces the last suffix.
29
30    For example:
31    ```py
32    path = Path("lang/en_us.flatten.json5")
33    replace_suffixes(path, ".json")  # lang/en_us.json
34    path.with_suffix(".json")        # lang/en_us.flatten.json
35    ```
36    """
37    return strip_suffixes(path).with_suffix(suffix)

 9@contextmanager
10def set_contextvar(contextvar: ContextVar[_T], value: _T):
11    token = contextvar.set(value)
12    try:
13        yield
14    finally:
15        contextvar.reset(token)

 80def setup_logging(
 81    verbosity: int,
 82    ci: bool,
 83    *,
 84    filters: Iterable[Filter] | None = None,
 85    quiet_langs: Iterable[str] | None = None,
 86):
 87    logging.addLevelName(TRACE, "TRACE")
 88
 89    root_logger = logging.getLogger()
 90
 91    if root_logger.handlers:
 92        for handler in root_logger.handlers:
 93            if isinstance(handler.formatter, HexdocLevelFormatter):
 94                logger.debug(f"Removing existing handler from root logger: {handler}")
 95                root_logger.removeHandler(handler)
 96
 97    level = verbosity_log_level(verbosity)
 98    root_logger.setLevel(level)
 99
100    formats = {
101        logging.DEBUG: log_format("relativeCreated", "levelname", "name"),
102    }
103
104    if level >= logging.INFO:
105        # set this here so we don't clobber exceptions in verbose mode
106        # but don't set it if Typer's pretty tracebacks are enabled
107        # see also: https://typer.tiangolo.com/tutorial/exceptions/#disable-pretty-exceptions
108        if os.getenv("_TYPER_STANDARD_TRACEBACK"):
109            sys.excepthook = filtered_excepthook
110
111        formats |= {
112            logging.INFO: log_format("levelname"),
113            logging.WARNING: log_format("levelname", "name"),
114        }
115
116    if ci:
117        formats |= {
118            logging.WARNING: "::warning file={name},line={lineno},title={levelname}::{message}",
119            logging.ERROR: "::error file={name},line={lineno},title={levelname}::{message}",
120        }
121
122    handler = StreamHandler()
123
124    handler.setLevel(level)
125    handler.setFormatter(HexdocLevelFormatter(formats, style="{"))
126
127    if filters:
128        for filter in filters:
129            handler.addFilter(filter)
130
131    if quiet_langs:
132        for lang in quiet_langs:
133            handler.addFilter(RegexFilter(f"^No translation in {lang}"))
134
135    root_logger.addHandler(handler)
136
137    logging.getLogger("PIL").setLevel(logging.INFO)
138
139    logger.debug("Initialized logger.")

49def sorted_dict(d: Mapping[_T, _T_Sortable]) -> dict[_T, _T_Sortable]:
50    return dict(sorted(d.items(), key=lambda item: item[1]))

10def strip_suffixes(path: Path) -> Path:
11    """Removes all suffixes from a path. This is helpful because `path.with_suffix("")`
12    only removes the last suffix.
13
14    For example:
15    ```py
16    path = Path("lang/en_us.flatten.json5")
17    strip_suffixes(path)  # lang/en_us
18    path.with_suffix("")  # lang/en_us.flatten
19    ```
20    """
21    while path.suffix:
22        path = path.with_suffix("")
23    return path

40def write_to_path(path: Path, data: str | bytes, encoding: str = "utf-8"):
41    logger.log(TRACE, f"Writing to {path}")
42    path.parent.mkdir(parents=True, exist_ok=True)
43    match data:
44        case str():
45            path.write_text(data, encoding)
46        case _:
47            path.write_bytes(data)