codekingpro/portable-devtools

1from __future__ import annotations2 3import abc4 5from pypika.terms import (6    Field,7    Function,8    Term,9)10from pypika.utils import format_alias_sql11 12 13class Array(Term):14    def __init__(15        self, values: list, converter_cls=None, converter_options: dict | None = None, alias: str | None = None16    ):17        super().__init__(alias)18        self._values = values19        self._converter_cls = converter_cls20        self._converter_options = converter_options or dict()21 22    def get_sql(self):23        if self._converter_cls:24            converted = []25            for value in self._values:26                converter = self._converter_cls(value, **self._converter_options)27                converted.append(converter.get_sql())28            sql = "".join(["[", ",".join(converted), "]"])29 30        else:31            sql = str(self._values)32 33        return format_alias_sql(sql, self.alias)34 35 36class HasAny(Function):37    def __init__(38        self,39        left_array: Array or Field,40        right_array: Array or Field,41        alias: str | None = None,42        schema: str | None = None,43    ):44        self._left_array = left_array45        self._right_array = right_array46        self.alias = alias47        self.schema = schema48        self.args = ()49        self.name = "hasAny"50 51    def get_sql(self, with_alias=False, with_namespace=False, quote_char=None, dialect=None, **kwargs):52        left = self._left_array.get_sql()53        right = self._right_array.get_sql()54        sql = "{name}({left},{right})".format(55            name=self.name,56            left='"%s"' % left if isinstance(self._left_array, Field) else left,57            right='"%s"' % right if isinstance(self._right_array, Field) else right,58        )59        return format_alias_sql(sql, self.alias, **kwargs)60 61 62class _AbstractArrayFunction(Function, metaclass=abc.ABCMeta):63    def __init__(self, array: Array or Field, alias: str | None = None, schema: str | None = None):64        self.schema = schema65        self.alias = alias66        self.name = self.clickhouse_function()67        self._array = array68 69    def get_sql(self, with_namespace=False, quote_char=None, dialect=None, **kwargs):70        array = self._array.get_sql()71        sql = "{name}({array})".format(72            name=self.name,73            array='"%s"' % array if isinstance(self._array, Field) else array,74        )75        return format_alias_sql(sql, self.alias, **kwargs)76 77    @classmethod78    @abc.abstractmethod79    def clickhouse_function(cls) -> str:80        pass81 82 83class NotEmpty(_AbstractArrayFunction):84    @classmethod85    def clickhouse_function(cls) -> str:86        return "notEmpty"87 88 89class Empty(_AbstractArrayFunction):90    @classmethod91    def clickhouse_function(cls) -> str:92        return "empty"93 94 95class Length(_AbstractArrayFunction):96    @classmethod97    def clickhouse_function(cls) -> str:98        return "length"99