codekingpro/portable-devtools

1from __future__ import annotations2 3import collections4from collections.abc import Iterable5import textwrap6from typing import Any7from typing import Callable8from typing import cast9from typing import Dict10from typing import List11from typing import Mapping12from typing import MutableMapping13from typing import NoReturn14from typing import Optional15from typing import overload16from typing import Sequence17from typing import Set18from typing import Tuple19from typing import Type20from typing import TYPE_CHECKING21from typing import TypeVar22from typing import Union23import uuid24import warnings25 26from sqlalchemy.util import asbool as asbool  # noqa: F40127from sqlalchemy.util import immutabledict as immutabledict  # noqa: F40128from sqlalchemy.util import to_list as to_list  # noqa: F40129from sqlalchemy.util import unique_list as unique_list30 31from .compat import inspect_getfullargspec32 33if True:34    # zimports workaround :(35    from sqlalchemy.util import (  # noqa: F40136        memoized_property as memoized_property,37    )38 39 40EMPTY_DICT: Mapping[Any, Any] = immutabledict()41_T = TypeVar("_T", bound=Any)42 43_C = TypeVar("_C", bound=Callable[..., Any])44 45 46class _ModuleClsMeta(type):47    def __setattr__(cls, key: str, value: Callable[..., Any]) -> None:48        super().__setattr__(key, value)49        cls._update_module_proxies(key)  # type: ignore50 51 52class ModuleClsProxy(metaclass=_ModuleClsMeta):53    """Create module level proxy functions for the54    methods on a given class.55 56    The functions will have a compatible signature57    as the methods.58 59    """60 61    _setups: Dict[62        Type[Any],63        Tuple[64            Set[str],65            List[Tuple[MutableMapping[str, Any], MutableMapping[str, Any]]],66        ],67    ] = collections.defaultdict(lambda: (set(), []))68 69    @classmethod70    def _update_module_proxies(cls, name: str) -> None:71        attr_names, modules = cls._setups[cls]72        for globals_, locals_ in modules:73            cls._add_proxied_attribute(name, globals_, locals_, attr_names)74 75    def _install_proxy(self) -> None:76        attr_names, modules = self._setups[self.__class__]77        for globals_, locals_ in modules:78            globals_["_proxy"] = self79            for attr_name in attr_names:80                globals_[attr_name] = getattr(self, attr_name)81 82    def _remove_proxy(self) -> None:83        attr_names, modules = self._setups[self.__class__]84        for globals_, locals_ in modules:85            globals_["_proxy"] = None86            for attr_name in attr_names:87                del globals_[attr_name]88 89    @classmethod90    def create_module_class_proxy(91        cls,92        globals_: MutableMapping[str, Any],93        locals_: MutableMapping[str, Any],94    ) -> None:95        attr_names, modules = cls._setups[cls]96        modules.append((globals_, locals_))97        cls._setup_proxy(globals_, locals_, attr_names)98 99    @classmethod100    def _setup_proxy(101        cls,102        globals_: MutableMapping[str, Any],103        locals_: MutableMapping[str, Any],104        attr_names: Set[str],105    ) -> None:106        for methname in dir(cls):107            cls._add_proxied_attribute(methname, globals_, locals_, attr_names)108 109    @classmethod110    def _add_proxied_attribute(111        cls,112        methname: str,113        globals_: MutableMapping[str, Any],114        locals_: MutableMapping[str, Any],115        attr_names: Set[str],116    ) -> None:117        if not methname.startswith("_"):118            meth = getattr(cls, methname)119            if callable(meth):120                locals_[methname] = cls._create_method_proxy(121                    methname, globals_, locals_122                )123            else:124                attr_names.add(methname)125 126    @classmethod127    def _create_method_proxy(128        cls,129        name: str,130        globals_: MutableMapping[str, Any],131        locals_: MutableMapping[str, Any],132    ) -> Callable[..., Any]:133        fn = getattr(cls, name)134 135        def _name_error(name: str, from_: Exception) -> NoReturn:136            raise NameError(137                "Can't invoke function '%s', as the proxy object has "138                "not yet been "139                "established for the Alembic '%s' class.  "140                "Try placing this code inside a callable."141                % (name, cls.__name__)142            ) from from_143 144        globals_["_name_error"] = _name_error145 146        translations = getattr(fn, "_legacy_translations", [])147        if translations:148            spec = inspect_getfullargspec(fn)149            if spec[0] and spec[0][0] == "self":150                spec[0].pop(0)151 152            outer_args = inner_args = "*args, **kw"153            translate_str = "args, kw = _translate(%r, %r, %r, args, kw)" % (154                fn.__name__,155                tuple(spec),156                translations,157            )158 159            def translate(160                fn_name: str, spec: Any, translations: Any, args: Any, kw: Any161            ) -> Any:162                return_kw = {}163                return_args = []164 165                for oldname, newname in translations:166                    if oldname in kw:167                        warnings.warn(168                            "Argument %r is now named %r "169                            "for method %s()." % (oldname, newname, fn_name)170                        )171                        return_kw[newname] = kw.pop(oldname)172                return_kw.update(kw)173 174                args = list(args)175                if spec[3]:176                    pos_only = spec[0][: -len(spec[3])]177                else:178                    pos_only = spec[0]179                for arg in pos_only:180                    if arg not in return_kw:181                        try:182                            return_args.append(args.pop(0))183                        except IndexError:184                            raise TypeError(185                                "missing required positional argument: %s"186                                % arg187                            )188                return_args.extend(args)189 190                return return_args, return_kw191 192            globals_["_translate"] = translate193        else:194            outer_args = "*args, **kw"195            inner_args = "*args, **kw"196            translate_str = ""197 198        func_text = textwrap.dedent(199            """\200        def %(name)s(%(args)s):201            %(doc)r202            %(translate)s203            try:204                p = _proxy205            except NameError as ne:206                _name_error('%(name)s', ne)207            return _proxy.%(name)s(%(apply_kw)s)208            e209        """210            % {211                "name": name,212                "translate": translate_str,213                "args": outer_args,214                "apply_kw": inner_args,215                "doc": fn.__doc__,216            }217        )218        lcl: MutableMapping[str, Any] = {}219 220        exec(func_text, cast("Dict[str, Any]", globals_), lcl)221        return cast("Callable[..., Any]", lcl[name])222 223 224def _with_legacy_names(translations: Any) -> Any:225    def decorate(fn: _C) -> _C:226        fn._legacy_translations = translations  # type: ignore[attr-defined]227        return fn228 229    return decorate230 231 232def rev_id() -> str:233    return uuid.uuid4().hex[-12:]234 235 236@overload237def to_tuple(x: Any, default: Tuple[Any, ...]) -> Tuple[Any, ...]:238    ...239 240 241@overload242def to_tuple(x: None, default: Optional[_T] = ...) -> _T:243    ...244 245 246@overload247def to_tuple(248    x: Any, default: Optional[Tuple[Any, ...]] = None249) -> Tuple[Any, ...]:250    ...251 252 253def to_tuple(254    x: Any, default: Optional[Tuple[Any, ...]] = None255) -> Optional[Tuple[Any, ...]]:256    if x is None:257        return default258    elif isinstance(x, str):259        return (x,)260    elif isinstance(x, Iterable):261        return tuple(x)262    else:263        return (x,)264 265 266def dedupe_tuple(tup: Tuple[str, ...]) -> Tuple[str, ...]:267    return tuple(unique_list(tup))268 269 270class Dispatcher:271    def __init__(self, uselist: bool = False) -> None:272        self._registry: Dict[Tuple[Any, ...], Any] = {}273        self.uselist = uselist274 275    def dispatch_for(276        self, target: Any, qualifier: str = "default"277    ) -> Callable[[_C], _C]:278        def decorate(fn: _C) -> _C:279            if self.uselist:280                self._registry.setdefault((target, qualifier), []).append(fn)281            else:282                assert (target, qualifier) not in self._registry283                self._registry[(target, qualifier)] = fn284            return fn285 286        return decorate287 288    def dispatch(self, obj: Any, qualifier: str = "default") -> Any:289        if isinstance(obj, str):290            targets: Sequence[Any] = [obj]291        elif isinstance(obj, type):292            targets = obj.__mro__293        else:294            targets = type(obj).__mro__295 296        for spcls in targets:297            if qualifier != "default" and (spcls, qualifier) in self._registry:298                return self._fn_or_list(self._registry[(spcls, qualifier)])299            elif (spcls, "default") in self._registry:300                return self._fn_or_list(self._registry[(spcls, "default")])301        else:302            raise ValueError("no dispatch function for object: %s" % obj)303 304    def _fn_or_list(305        self, fn_or_list: Union[List[Callable[..., Any]], Callable[..., Any]]306    ) -> Callable[..., Any]:307        if self.uselist:308 309            def go(*arg: Any, **kw: Any) -> None:310                if TYPE_CHECKING:311                    assert isinstance(fn_or_list, Sequence)312                for fn in fn_or_list:313                    fn(*arg, **kw)314 315            return go316        else:317            return fn_or_list  # type: ignore318 319    def branch(self) -> Dispatcher:320        """Return a copy of this dispatcher that is independently321        writable."""322 323        d = Dispatcher()324        if self.uselist:325            d._registry.update(326                (k, [fn for fn in self._registry[k]]) for k in self._registry327            )328        else:329            d._registry.update(self._registry)330        return d331 332 333def not_none(value: Optional[_T]) -> _T:334    assert value is not None335    return value336