11405726556

Committed 18 Oct 2024 02:28PM UTC coverage: 85.988% (-0.09%) from 86.082%

Build # 11405726556

Build Type

push

github

Committed by

xzkostyan

Commit Message

Version bumped to 0.2.7

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

63 existing lines in 11 files now uncovered.

2332 of 2712 relevant lines covered (85.99%)

20.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.86

/clickhouse_sqlalchemy/drivers/compilers/typecompiler.py

from sqlalchemy.sql import compiler, type_api
from sqlalchemy.sql.ddl import CreateColumn


class ClickHouseTypeCompiler(compiler.GenericTypeCompiler):
    def visit_string(self, type_, **kw):
        if type_.length is None:
            return 'String'
        else:
            return 'FixedString(%s)' % type_.length

    def visit_array(self, type_, **kw):
        item_type = type_api.to_instance(type_.item_type)
        return 'Array(%s)' % self.process(item_type, **kw)

    def visit_nullable(self, type_, **kw):
        nested_type = type_api.to_instance(type_.nested_type)
        return 'Nullable(%s)' % self.process(nested_type, **kw)

    def visit_lowcardinality(self, type_, **kw):
        nested_type = type_api.to_instance(type_.nested_type)
        return "LowCardinality(%s)" % self.process(nested_type, **kw)

    def visit_int8(self, type_, **kw):
        return 'Int8'

    def visit_uint8(self, type_, **kw):
        return 'UInt8'

    def visit_int16(self, type_, **kw):
        return 'Int16'

    def visit_uint16(self, type_, **kw):
        return 'UInt16'

    def visit_int32(self, type_, **kw):
        return 'Int32'

    def visit_uint32(self, type_, **kw):
        return 'UInt32'

    def visit_int64(self, type_, **kw):
        return 'Int64'

    def visit_uint64(self, type_, **kw):
        return 'UInt64'

    def visit_int128(self, type_, **kw):
        return 'Int128'

    def visit_uint128(self, type_, **kw):
        return 'UInt128'

    def visit_int256(self, type_, **kw):
        return 'Int256'

    def visit_uint256(self, type_, **kw):
        return 'UInt256'

    def visit_date(self, type_, **kw):
        return 'Date'

    def visit_datetime(self, type_, **kw):
        if type_.timezone:
            return "DateTime('%s')" % type_.timezone
        else:
            return 'DateTime'

    def visit_datetime64(self, type_, **kw):
        if type_.timezone:
            return "DateTime64(%s, '%s')" % (type_.precision, type_.timezone)
        else:
            return 'DateTime64(%s)' % type_.precision

    def visit_float32(self, type_, **kw):
        return 'Float32'

    def visit_float64(self, type_, **kw):
        return 'Float64'

    def visit_numeric(self, type_, **kw):
        return 'Decimal(%s, %s)' % (type_.precision, type_.scale)

    def visit_boolean(self, type_, **kw):
        return 'Bool'

    def visit_nested(self, nested, **kwargs):
        ddl_compiler = self.dialect.ddl_compiler(self.dialect, None)
        cols_create = [
            ddl_compiler.visit_create_column(CreateColumn(nested_child))
            for nested_child in nested.columns
        ]
        return 'Nested(%s)' % ', '.join(cols_create)

    def _render_enum(self, db_type, type_, **kw):
        choices = (
            "'%s' = %d" %
            (x.name.replace("'", r"\'"), x.value) for x in type_.enum_class
        )
        return '%s(%s)' % (db_type, ', '.join(choices))

    def visit_enum(self, type_, **kw):
        return self._render_enum('Enum', type_, **kw)

    def visit_enum8(self, type_, **kw):
        return self._render_enum('Enum8', type_, **kw)

    def visit_enum16(self, type_, **kw):
        return self._render_enum('Enum16', type_, **kw)

    def visit_uuid(self, type_, **kw):
        return 'UUID'

    def visit_ipv4(self, type_, **kw):
        return 'IPv4'

    def visit_ipv6(self, type_, **kw):
        return 'IPv6'

    def visit_tuple(self, type_, **kw):
        cols = (
            self.process(type_api.to_instance(nested_type), **kw)
            for nested_type in type_.nested_types
        )
        return 'Tuple(%s)' % ', '.join(cols)

    def visit_map(self, type_, **kw):
        key_type = type_api.to_instance(type_.key_type)
        value_type = type_api.to_instance(type_.value_type)
        return 'Map(%s, %s)' % (
            self.process(key_type, **kw),
            self.process(value_type, **kw)
        )

1	from sqlalchemy.sql import compiler, type_api	16✔
2	from sqlalchemy.sql.ddl import CreateColumn	24✔
3
4
5	class ClickHouseTypeCompiler(compiler.GenericTypeCompiler):	24✔
6	def visit_string(self, type_, **kw):	24✔
7	if type_.length is None:	24✔
8	return 'String'	24✔
9	else:
10	return 'FixedString(%s)' % type_.length	24✔
11
12	def visit_array(self, type_, **kw):	24✔
13	item_type = type_api.to_instance(type_.item_type)	24✔
14	return 'Array(%s)' % self.process(item_type, **kw)	24✔
15
16	def visit_nullable(self, type_, **kw):	24✔
17	nested_type = type_api.to_instance(type_.nested_type)	24✔
18	return 'Nullable(%s)' % self.process(nested_type, **kw)	24✔
19
20	def visit_lowcardinality(self, type_, **kw):	24✔
21	nested_type = type_api.to_instance(type_.nested_type)	18✔
22	return "LowCardinality(%s)" % self.process(nested_type, **kw)	18✔
23
24	def visit_int8(self, type_, **kw):	24✔
25	return 'Int8'	24✔
26
27	def visit_uint8(self, type_, **kw):	24✔
28	return 'UInt8'	24✔
29
30	def visit_int16(self, type_, **kw):	24✔
31	return 'Int16'	×
32
33	def visit_uint16(self, type_, **kw):	24✔
UNCOV 34	return 'UInt16'	×
35
36	def visit_int32(self, type_, **kw):	24✔
37	return 'Int32'	24✔
38
39	def visit_uint32(self, type_, **kw):	24✔
40	return 'UInt32'	24✔
41
42	def visit_int64(self, type_, **kw):	24✔
43	return 'Int64'	×
44
45	def visit_uint64(self, type_, **kw):	24✔
UNCOV 46	return 'UInt64'	×
47
48	def visit_int128(self, type_, **kw):	24✔
49	return 'Int128'	12✔
50
51	def visit_uint128(self, type_, **kw):	24✔
52	return 'UInt128'	12✔
53
54	def visit_int256(self, type_, **kw):	24✔
55	return 'Int256'	12✔
56
57	def visit_uint256(self, type_, **kw):	24✔
58	return 'UInt256'	12✔
59
60	def visit_date(self, type_, **kw):	24✔
61	return 'Date'	24✔
62
63	def visit_datetime(self, type_, **kw):	24✔
64	if type_.timezone:	24✔
65	return "DateTime('%s')" % type_.timezone	24✔
66	else:
67	return 'DateTime'	24✔
68
69	def visit_datetime64(self, type_, **kw):	24✔
70	if type_.timezone:	24✔
71	return "DateTime64(%s, '%s')" % (type_.precision, type_.timezone)	24✔
72	else:
73	return 'DateTime64(%s)' % type_.precision	24✔
74
75	def visit_float32(self, type_, **kw):	24✔
76	return 'Float32'	24✔
77
78	def visit_float64(self, type_, **kw):	24✔
UNCOV 79	return 'Float64'	×
80
81	def visit_numeric(self, type_, **kw):	24✔
82	return 'Decimal(%s, %s)' % (type_.precision, type_.scale)	24✔
83
84	def visit_boolean(self, type_, **kw):	24✔
85	return 'Bool'	24✔
86
87	def visit_nested(self, nested, **kwargs):	24✔
88	ddl_compiler = self.dialect.ddl_compiler(self.dialect, None)	24✔
89	cols_create = [	24✔
90	ddl_compiler.visit_create_column(CreateColumn(nested_child))
91	for nested_child in nested.columns
92	]
93	return 'Nested(%s)' % ', '.join(cols_create)	24✔
94
95	def _render_enum(self, db_type, type_, **kw):	24✔
96	choices = (	24✔
97	"'%s' = %d" %
98	(x.name.replace("'", r"\'"), x.value) for x in type_.enum_class
99	)
100	return '%s(%s)' % (db_type, ', '.join(choices))	24✔
101
102	def visit_enum(self, type_, **kw):	24✔
103	return self._render_enum('Enum', type_, **kw)	24✔
104
105	def visit_enum8(self, type_, **kw):	24✔
106	return self._render_enum('Enum8', type_, **kw)	24✔
107
108	def visit_enum16(self, type_, **kw):	24✔
109	return self._render_enum('Enum16', type_, **kw)	24✔
110
111	def visit_uuid(self, type_, **kw):	24✔
UNCOV 112	return 'UUID'	×
113
114	def visit_ipv4(self, type_, **kw):	24✔
115	return 'IPv4'	18✔
116
117	def visit_ipv6(self, type_, **kw):	24✔
118	return 'IPv6'	18✔
119
120	def visit_tuple(self, type_, **kw):	24✔
121	cols = (	24✔
122	self.process(type_api.to_instance(nested_type), **kw)
123	for nested_type in type_.nested_types
124	)
125	return 'Tuple(%s)' % ', '.join(cols)	24✔
126
127	def visit_map(self, type_, **kw):	24✔
128	key_type = type_api.to_instance(type_.key_type)	12✔
129	value_type = type_api.to_instance(type_.value_type)	12✔
130	return 'Map(%s, %s)' % (	12✔
131	self.process(key_type, **kw),
132	self.process(value_type, **kw)
133	)

xzkostyan / clickhouse-sqlalchemy / 11405726556

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous