How to use aiochclient - 10 common examples

async def chclient(request):
    async with aiohttp.ClientSession() as s:
        yield ChClient(s, **request.param)

async def test_bad_select(self):
        with pytest.raises(ChClientError):
            await self.ch.execute("SELECT * FROM all_types WHERE", 1, 2, 3, 4)

async def test_bad_query(self):
        with pytest.raises(ChClientError):
            await self.ch.execute("SELE")

return b"%a" % str(value.replace(microsecond=0))


class UUIDType(BaseType):
    def p_type(self, string):
        return UUID(string.strip("'"))

    def convert(self, value: bytes) -> UUID:
        return self.p_type(value.decode())

    @staticmethod
    def unconvert(value: UUID) -> bytes:
        return b"%a" % str(value)


class TupleType(BaseType):

    __slots__ = ("name", "types")

    def __init__(self, name: str, **kwargs):
        super().__init__(name, **kwargs)
        tps = RE_TUPLE.findall(name)[0]
        self.types = tuple(what_py_type(tp, container=True) for tp in tps.split(","))

    def p_type(self, string: str) -> tuple:
        return tuple(
            tp.p_type(val)
            for tp, val in zip(self.types, self.seq_parser(string.strip("()")))
        )

    @staticmethod
    def unconvert(value) -> bytes:

super().__init__(name, **kwargs)
        tps = RE_TUPLE.findall(name)[0]
        self.types = tuple(what_py_type(tp, container=True) for tp in tps.split(","))

    def p_type(self, string: str) -> tuple:
        return tuple(
            tp.p_type(val)
            for tp, val in zip(self.types, self.seq_parser(string.strip("()")))
        )

    @staticmethod
    def unconvert(value) -> bytes:
        return b"(" + b",".join(py2ch(elem) for elem in value) + b")"


class ArrayType(BaseType):

    __slots__ = ("name", "type")

    def __init__(self, name: str, **kwargs):
        super().__init__(name, **kwargs)
        self.type = what_py_type(RE_ARRAY.findall(name)[0], container=True)

    def p_type(self, string: str) -> list:
        return [self.type.p_type(val) for val in self.seq_parser(string.strip("[]"))]

    @staticmethod
    def unconvert(value) -> bytes:
        return b"[" + b",".join(py2ch(elem) for elem in value) + b"]"


class NullableType(BaseType):

def what_py_type(name: str, container: bool = False) -> BaseType:
    """ Returns needed type class from clickhouse type name """
    name = name.strip()
    try:
        return CH_TYPES_MAPPING[name.split("(")[0]](name, container=container)
    except KeyError:
        raise ChClientError(f"Unrecognized type name: '{name}'")

"UUID": UUIDType,
    "LowCardinality": LowCardinalityType,
    "Decimal": DecimalType,
    "Decimal32": DecimalType,
    "Decimal64": DecimalType,
    "Decimal128": DecimalType,
}

PY_TYPES_MAPPING = {
    int: IntType.unconvert,
    float: FloatType.unconvert,
    str: StrType.unconvert,
    dt.date: DateType.unconvert,
    dt.datetime: DateTimeType.unconvert,
    tuple: TupleType.unconvert,
    list: ArrayType.unconvert,
    type(None): NullableType.unconvert,
    UUID: UUIDType.unconvert,
    Decimal: DecimalType.unconvert,
}


def what_py_type(name: str, container: bool = False) -> BaseType:
    """ Returns needed type class from clickhouse type name """
    name = name.strip()
    try:
        return CH_TYPES_MAPPING[name.split("(")[0]](name, container=container)
    except KeyError:
        raise ChClientError(f"Unrecognized type name: '{name}'")


def what_py_converter(name: str, container: bool = False) -> Callable:

"Nullable": NullableType,
    "Nothing": NothingType,
    "UUID": UUIDType,
    "LowCardinality": LowCardinalityType,
    "Decimal": DecimalType,
    "Decimal32": DecimalType,
    "Decimal64": DecimalType,
    "Decimal128": DecimalType,
}

PY_TYPES_MAPPING = {
    int: IntType.unconvert,
    float: FloatType.unconvert,
    str: StrType.unconvert,
    dt.date: DateType.unconvert,
    dt.datetime: DateTimeType.unconvert,
    tuple: TupleType.unconvert,
    list: ArrayType.unconvert,
    type(None): NullableType.unconvert,
    UUID: UUIDType.unconvert,
    Decimal: DecimalType.unconvert,
}


def what_py_type(name: str, container: bool = False) -> BaseType:
    """ Returns needed type class from clickhouse type name """
    name = name.strip()
    try:
        return CH_TYPES_MAPPING[name.split("(")[0]](name, container=container)
    except KeyError:
        raise ChClientError(f"Unrecognized type name: '{name}'")

"Array": ArrayType,
    "Nullable": NullableType,
    "Nothing": NothingType,
    "UUID": UUIDType,
    "LowCardinality": LowCardinalityType,
    "Decimal": DecimalType,
    "Decimal32": DecimalType,
    "Decimal64": DecimalType,
    "Decimal128": DecimalType,
}

PY_TYPES_MAPPING = {
    int: IntType.unconvert,
    float: FloatType.unconvert,
    str: StrType.unconvert,
    dt.date: DateType.unconvert,
    dt.datetime: DateTimeType.unconvert,
    tuple: TupleType.unconvert,
    list: ArrayType.unconvert,
    type(None): NullableType.unconvert,
    UUID: UUIDType.unconvert,
    Decimal: DecimalType.unconvert,
}


def what_py_type(name: str, container: bool = False) -> BaseType:
    """ Returns needed type class from clickhouse type name """
    name = name.strip()
    try:
        return CH_TYPES_MAPPING[name.split("(")[0]](name, container=container)
    except KeyError:
        raise ChClientError(f"Unrecognized type name: '{name}'")

"Decimal32": DecimalType,
    "Decimal64": DecimalType,
    "Decimal128": DecimalType,
}

PY_TYPES_MAPPING = {
    int: IntType.unconvert,
    float: FloatType.unconvert,
    str: StrType.unconvert,
    dt.date: DateType.unconvert,
    dt.datetime: DateTimeType.unconvert,
    tuple: TupleType.unconvert,
    list: ArrayType.unconvert,
    type(None): NullableType.unconvert,
    UUID: UUIDType.unconvert,
    Decimal: DecimalType.unconvert,
}


def what_py_type(name: str, container: bool = False) -> BaseType:
    """ Returns needed type class from clickhouse type name """
    name = name.strip()
    try:
        return CH_TYPES_MAPPING[name.split("(")[0]](name, container=container)
    except KeyError:
        raise ChClientError(f"Unrecognized type name: '{name}'")


def what_py_converter(name: str, container: bool = False) -> Callable:
    """ Returns needed type class from clickhouse type name """
    return what_py_type(name, container).convert