codekingpro/portable-devtools

1from __future__ import annotations2 3import sys4from collections.abc import Callable5from functools import wraps6from typing import Any, TypeVar, overload7 8if sys.version_info >= (3, 10):9    from typing import Concatenate, ParamSpec10else:11    from typing_extensions import Concatenate, ParamSpec12 13 14__author__ = "Timothy Heys"15__email__ = "theys@kayak.com"16 17 18class QueryException(Exception):19    pass20 21 22class GroupingException(Exception):23    pass24 25 26class CaseException(Exception):27    pass28 29 30class JoinException(Exception):31    pass32 33 34class SetOperationException(Exception):35    pass36 37 38class RollupException(Exception):39    pass40 41 42class DialectNotSupported(Exception):43    pass44 45 46class FunctionException(Exception):47    pass48 49 50_Self = TypeVar("_Self")51P = ParamSpec("P")52R = TypeVar("R")53 54 55@overload56def builder(func: Callable[Concatenate[_Self, P], None]) -> Callable[Concatenate[_Self, P], _Self]: ...57 58 59@overload60def builder(func: Callable[Concatenate[_Self, P], R]) -> Callable[Concatenate[_Self, P], R]: ...61 62 63def builder(func: Callable[Concatenate[_Self, P], R | None]) -> Callable[Concatenate[_Self, P], _Self | R]:64    """65    Decorator for wrapper "builder" functions.  These are functions on the Query class or other classes used for66    building queries which mutate the query and return self.  To make the build functions immutable, this decorator is67    used which will deepcopy the current instance.  This decorator will return the return value of the inner function68    or the new copy of the instance.  The inner function does not need to return self.69    """70    import copy71 72    @wraps(func)73    def _copy(self: _Self, *args: P.args, **kwargs: P.kwargs) -> _Self | R:74        self_copy = copy.copy(self) if getattr(self, "immutable", True) else self75        result = func(self_copy, *args, **kwargs)76 77        # Return self if the inner function returns None.  This way the inner function can return something78        # different (for example when creating joins, a different builder is returned).79        if result is None:80            return self_copy81 82        return result83 84    return _copy85 86 87def ignore_copy(func: Callable[[_Self, str], R]) -> Callable[[_Self, str], R]:88    """89    Decorator for wrapping the __getattr__ function for classes that are copied via deepcopy.  This prevents infinite90    recursion caused by deepcopy looking for magic functions in the class. Any class implementing __getattr__ that is91    meant to be deepcopy'd should use this decorator.92 93    deepcopy is used by pypika in builder functions (decorated by @builder) to make the results immutable.  Any data94    model type class (stored in the Query instance) is copied.95    """96 97    @wraps(func)98    def _getattr(self, name: str) -> R:99        if name in [100            "__copy__",101            "__deepcopy__",102            "__getstate__",103            "__setstate__",104            "__getnewargs__",105        ]:106            raise AttributeError("'%s' object has no attribute '%s'" % (self.__class__.__name__, name))107 108        return func(self, name)109 110    return _getattr111 112 113def resolve_is_aggregate(values: list[bool | None]) -> bool | None:114    """115    Resolves the is_aggregate flag for an expression that contains multiple terms.  This works like a voter system,116    each term votes True or False or abstains with None.117 118    :param values: A list of booleans (or None) for each term in the expression119    :return: If all values are True or None, True is returned.  If all values are None, None is returned. Otherwise,120        False is returned.121    """122    result = [x for x in values if x is not None]123    if result:124        return all(result)125    return None126 127 128def format_quotes(value: Any, quote_char: str | None) -> str:129    if quote_char:130        value = str(value).replace(quote_char, quote_char * 2)131 132    return "{quote}{value}{quote}".format(value=value, quote=quote_char or "")133 134 135def format_alias_sql(136    sql: str,137    alias: str | None,138    quote_char: str | None = None,139    alias_quote_char: str | None = None,140    as_keyword: bool = False,141    **kwargs: Any,142) -> str:143    if alias is None:144        return sql145    return "{sql}{_as}{alias}".format(146        sql=sql, _as=' AS ' if as_keyword else ' ', alias=format_quotes(alias, alias_quote_char or quote_char)147    )148 149 150def validate(*args: Any, exc: Exception | None = None, type: type | None = None) -> None:151    if type is not None:152        for arg in args:153            if not isinstance(arg, type):154                raise exc155