mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
e29f9fd738
* refactor(driver-utils): add type annotations to driver utility modules * refactor(driver-utils): widen QueryContext.parameters to accept Sequence | dict | None * silence macOS-only mypy error, note query_id change --------- Signed-off-by: Joe Spadola <[email protected]> Co-authored-by: Joe S <[email protected]>
37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
from clickhouse_connect.datatypes.registry import get_from_name
|
|
from clickhouse_connect.driver.query import QueryResult
|
|
|
|
|
|
class QuerySummary:
|
|
summary: dict[str, str] = {}
|
|
|
|
def __init__(self, summary: dict[str, str] | None = None):
|
|
if summary is not None:
|
|
self.summary = summary
|
|
|
|
@property
|
|
def written_rows(self) -> int:
|
|
return int(self.summary.get("written_rows", 0))
|
|
|
|
def written_bytes(self) -> int:
|
|
return int(self.summary.get("written_bytes", 0))
|
|
|
|
def query_id(self) -> str:
|
|
return self.summary.get("query_id", "")
|
|
|
|
def as_query_result(self) -> QueryResult:
|
|
data: list[int | str] = []
|
|
column_names = []
|
|
column_types = []
|
|
str_type = get_from_name("String")
|
|
int_type = get_from_name("Int64")
|
|
for key, value in self.summary.items():
|
|
column_names.append(key)
|
|
if value.isnumeric():
|
|
data.append(int(value))
|
|
column_types.append(int_type)
|
|
else:
|
|
data.append(value)
|
|
column_types.append(str_type)
|
|
return QueryResult([data], column_names=tuple(column_names), column_types=tuple(column_types))
|