codekingpro/portable-devtools

1# Copyright 2016 Julien Danjou2# Copyright 2016 Joshua Harlow3# Copyright 2013-2014 Ray Holder4#5# Licensed under the Apache License, Version 2.0 (the "License");6# you may not use this file except in compliance with the License.7# You may obtain a copy of the License at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS,13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14# See the License for the specific language governing permissions and15# limitations under the License.16import functools17import inspect18import sys19import typing20from datetime import timedelta21 22 23# sys.maxsize:24# An integer giving the maximum value a variable of type Py_ssize_t can take.25MAX_WAIT = sys.maxsize / 226 27 28class LoggerProtocol(typing.Protocol):29    """30    Protocol used by utils expecting a logger (eg: before_log).31 32    Compatible with logging, structlog, loguru, etc...33    """34 35    def log(36        self, level: int, msg: str, /, *args: typing.Any, **kwargs: typing.Any37    ) -> typing.Any: ...38 39 40def find_ordinal(pos_num: int) -> str:41    # See: https://en.wikipedia.org/wiki/English_numerals#Ordinal_numbers42    if pos_num == 0:43        return "th"44    elif pos_num == 1:45        return "st"46    elif pos_num == 2:47        return "nd"48    elif pos_num == 3:49        return "rd"50    elif 4 <= pos_num <= 20:51        return "th"52    else:53        return find_ordinal(pos_num % 10)54 55 56def to_ordinal(pos_num: int) -> str:57    return f"{pos_num}{find_ordinal(pos_num)}"58 59 60def get_callback_name(cb: typing.Callable[..., typing.Any]) -> str:61    """Get a callback fully-qualified name.62 63    If no name can be produced ``repr(cb)`` is called and returned.64    """65    segments = []66    try:67        segments.append(cb.__qualname__)68    except AttributeError:69        try:70            segments.append(cb.__name__)71        except AttributeError:72            pass73    if not segments:74        return repr(cb)75    else:76        try:77            # When running under sphinx it appears this can be none?78            if cb.__module__:79                segments.insert(0, cb.__module__)80        except AttributeError:81            pass82        return ".".join(segments)83 84 85time_unit_type = typing.Union[int, float, timedelta]86 87 88def to_seconds(time_unit: time_unit_type) -> float:89    return float(90        time_unit.total_seconds() if isinstance(time_unit, timedelta) else time_unit91    )92 93 94def is_coroutine_callable(call: typing.Callable[..., typing.Any]) -> bool:95    if inspect.isclass(call):96        return False97    if inspect.iscoroutinefunction(call):98        return True99    partial_call = isinstance(call, functools.partial) and call.func100    dunder_call = partial_call or getattr(call, "__call__", None)101    return inspect.iscoroutinefunction(dunder_call)102 103 104def wrap_to_async_func(105    call: typing.Callable[..., typing.Any],106) -> typing.Callable[..., typing.Awaitable[typing.Any]]:107    if is_coroutine_callable(call):108        return call109 110    async def inner(*args: typing.Any, **kwargs: typing.Any) -> typing.Any:111        return call(*args, **kwargs)112 113    return inner114