client.py 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787
  1. import io
  2. import logging
  3. from datetime import tzinfo
  4. import pytz
  5. from abc import ABC, abstractmethod
  6. from typing import Iterable, Optional, Any, Union, Sequence, Dict, Generator, BinaryIO
  7. from pytz.exceptions import UnknownTimeZoneError
  8. from clickhouse_connect import common
  9. from clickhouse_connect.common import version
  10. from clickhouse_connect.datatypes.registry import get_from_name
  11. from clickhouse_connect.datatypes.base import ClickHouseType
  12. from clickhouse_connect.driver import tzutil
  13. from clickhouse_connect.driver.common import dict_copy, StreamContext, coerce_int, coerce_bool
  14. from clickhouse_connect.driver.constants import CH_VERSION_WITH_PROTOCOL, PROTOCOL_VERSION_WITH_LOW_CARD
  15. from clickhouse_connect.driver.exceptions import ProgrammingError, OperationalError
  16. from clickhouse_connect.driver.external import ExternalData
  17. from clickhouse_connect.driver.insert import InsertContext
  18. from clickhouse_connect.driver.summary import QuerySummary
  19. from clickhouse_connect.driver.models import ColumnDef, SettingDef, SettingStatus
  20. from clickhouse_connect.driver.query import QueryResult, to_arrow, to_arrow_batches, QueryContext, arrow_buffer, \
  21. quote_identifier
  22. io.DEFAULT_BUFFER_SIZE = 1024 * 256
  23. logger = logging.getLogger(__name__)
  24. arrow_str_setting = 'output_format_arrow_string_as_string'
  25. # pylint: disable=too-many-public-methods, too-many-instance-attributes
  26. class Client(ABC):
  27. """
  28. Base ClickHouse Connect client
  29. """
  30. compression: str = None
  31. write_compression: str = None
  32. protocol_version = 0
  33. valid_transport_settings = set()
  34. optional_transport_settings = set()
  35. database = None
  36. max_error_message = 0
  37. apply_server_timezone = False
  38. def __init__(self,
  39. database: str,
  40. query_limit: int,
  41. uri: str,
  42. query_retries: int,
  43. server_host_name: Optional[str],
  44. apply_server_timezone: Optional[Union[str, bool]]):
  45. """
  46. Shared initialization of ClickHouse Connect client
  47. :param database: database name
  48. :param query_limit: default LIMIT for queries
  49. :param uri: uri for error messages
  50. """
  51. self.query_limit = coerce_int(query_limit)
  52. self.query_retries = coerce_int(query_retries)
  53. self.server_host_name = server_host_name
  54. self.server_tz, dst_safe = pytz.UTC, True
  55. self.server_version, server_tz = \
  56. tuple(self.command('SELECT version(), timezone()', use_database=False))
  57. try:
  58. server_tz = pytz.timezone(server_tz)
  59. server_tz, dst_safe = tzutil.normalize_timezone(server_tz)
  60. if apply_server_timezone is None:
  61. apply_server_timezone = dst_safe
  62. self.apply_server_timezone = apply_server_timezone == 'always' or coerce_bool(apply_server_timezone)
  63. except UnknownTimeZoneError:
  64. logger.warning('Warning, server is using an unrecognized timezone %s, will use UTC default', server_tz)
  65. if not self.apply_server_timezone and not tzutil.local_tz_dst_safe:
  66. logger.warning('local timezone %s may return unexpected times due to Daylight Savings Time/' +
  67. 'Summer Time differences', tzutil.local_tz.tzname())
  68. readonly = 'readonly'
  69. if not self.min_version('19.17'):
  70. readonly = common.get_setting('readonly')
  71. server_settings = self.query(f'SELECT name, value, {readonly} as readonly FROM system.settings LIMIT 10000')
  72. self.server_settings = {row['name']: SettingDef(**row) for row in server_settings.named_results()}
  73. if database and not database == '__default__':
  74. self.database = database
  75. if self.min_version(CH_VERSION_WITH_PROTOCOL):
  76. # Unfortunately we have to validate that the client protocol version is actually used by ClickHouse
  77. # since the query parameter could be stripped off (in particular, by CHProxy)
  78. test_data = self.raw_query('SELECT 1 AS check', fmt='Native', settings={
  79. 'client_protocol_version': PROTOCOL_VERSION_WITH_LOW_CARD
  80. })
  81. if test_data[8:16] == b'\x01\x01\x05check':
  82. self.protocol_version = PROTOCOL_VERSION_WITH_LOW_CARD
  83. self.uri = uri
  84. def _validate_settings(self, settings: Optional[Dict[str, Any]]) -> Dict[str, str]:
  85. """
  86. This strips any ClickHouse settings that are not recognized or are read only.
  87. :param settings: Dictionary of setting name and values
  88. :return: A filtered dictionary of settings with values rendered as strings
  89. """
  90. validated = {}
  91. invalid_action = common.get_setting('invalid_setting_action')
  92. for key, value in settings.items():
  93. str_value = self._validate_setting(key, value, invalid_action)
  94. if str_value is not None:
  95. validated[key] = value
  96. return validated
  97. def _validate_setting(self, key: str, value: Any, invalid_action: str) -> Optional[str]:
  98. if key not in self.valid_transport_settings:
  99. setting_def = self.server_settings.get(key)
  100. if setting_def is None or setting_def.readonly:
  101. if key in self.optional_transport_settings:
  102. return None
  103. if invalid_action == 'send':
  104. logger.warning('Attempting to send unrecognized or readonly setting %s', key)
  105. elif invalid_action == 'drop':
  106. logger.warning('Dropping unrecognized or readonly settings %s', key)
  107. return None
  108. else:
  109. raise ProgrammingError(f'Setting {key} is unknown or readonly') from None
  110. if isinstance(value, bool):
  111. return '1' if value else '0'
  112. return str(value)
  113. def _setting_status(self, key: str) -> SettingStatus:
  114. comp_setting = self.server_settings.get(key)
  115. if not comp_setting:
  116. return SettingStatus(False, False)
  117. return SettingStatus(comp_setting.value != '0', comp_setting.readonly != 1)
  118. def _prep_query(self, context: QueryContext):
  119. if context.is_select and not context.has_limit and self.query_limit:
  120. return f'{context.final_query}\n LIMIT {self.query_limit}'
  121. return context.final_query
  122. def _check_tz_change(self, new_tz) -> Optional[tzinfo]:
  123. if new_tz:
  124. try:
  125. new_tzinfo = pytz.timezone(new_tz)
  126. if new_tzinfo != self.server_tz:
  127. return new_tzinfo
  128. except UnknownTimeZoneError:
  129. logger.warning('Unrecognized timezone %s received from ClickHouse', new_tz)
  130. return None
  131. @abstractmethod
  132. def _query_with_context(self, context: QueryContext):
  133. pass
  134. @abstractmethod
  135. def set_client_setting(self, key, value):
  136. """
  137. Set a clickhouse setting for the client after initialization. If a setting is not recognized by ClickHouse,
  138. or the setting is identified as "read_only", this call will either throw a Programming exception or attempt
  139. to send the setting anyway based on the common setting 'invalid_setting_action'
  140. :param key: ClickHouse setting name
  141. :param value: ClickHouse setting value
  142. """
  143. @abstractmethod
  144. def get_client_setting(self, key) -> Optional[str]:
  145. """
  146. :param key: The setting key
  147. :return: The string value of the setting, if it exists, or None
  148. """
  149. # pylint: disable=too-many-arguments,unused-argument,too-many-locals
  150. def query(self,
  151. query: Optional[str] = None,
  152. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  153. settings: Optional[Dict[str, Any]] = None,
  154. query_formats: Optional[Dict[str, str]] = None,
  155. column_formats: Optional[Dict[str, Union[str, Dict[str, str]]]] = None,
  156. encoding: Optional[str] = None,
  157. use_none: Optional[bool] = None,
  158. column_oriented: Optional[bool] = None,
  159. use_numpy: Optional[bool] = None,
  160. max_str_len: Optional[int] = None,
  161. context: QueryContext = None,
  162. query_tz: Optional[Union[str, tzinfo]] = None,
  163. column_tzs: Optional[Dict[str, Union[str, tzinfo]]] = None,
  164. external_data: Optional[ExternalData] = None) -> QueryResult:
  165. """
  166. Main query method for SELECT, DESCRIBE and other SQL statements that return a result matrix. For
  167. parameters, see the create_query_context method
  168. :return: QueryResult -- data and metadata from response
  169. """
  170. if query and query.lower().strip().startswith('select __connect_version__'):
  171. return QueryResult([[f'ClickHouse Connect v.{version()} ⓒ ClickHouse Inc.']], None,
  172. ('connect_version',), (get_from_name('String'),))
  173. kwargs = locals().copy()
  174. del kwargs['self']
  175. query_context = self.create_query_context(**kwargs)
  176. if query_context.is_command:
  177. response = self.command(query,
  178. parameters=query_context.parameters,
  179. settings=query_context.settings,
  180. external_data=query_context.external_data)
  181. if isinstance(response, QuerySummary):
  182. return response.as_query_result()
  183. return QueryResult([response] if isinstance(response, list) else [[response]])
  184. return self._query_with_context(query_context)
  185. def query_column_block_stream(self,
  186. query: Optional[str] = None,
  187. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  188. settings: Optional[Dict[str, Any]] = None,
  189. query_formats: Optional[Dict[str, str]] = None,
  190. column_formats: Optional[Dict[str, Union[str, Dict[str, str]]]] = None,
  191. encoding: Optional[str] = None,
  192. use_none: Optional[bool] = None,
  193. context: QueryContext = None,
  194. query_tz: Optional[Union[str, tzinfo]] = None,
  195. column_tzs: Optional[Dict[str, Union[str, tzinfo]]] = None,
  196. external_data: Optional[ExternalData] = None) -> StreamContext:
  197. """
  198. Variation of main query method that returns a stream of column oriented blocks. For
  199. parameters, see the create_query_context method.
  200. :return: StreamContext -- Iterable stream context that returns column oriented blocks
  201. """
  202. return self._context_query(locals(), use_numpy=False, streaming=True).column_block_stream
  203. def query_row_block_stream(self,
  204. query: Optional[str] = None,
  205. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  206. settings: Optional[Dict[str, Any]] = None,
  207. query_formats: Optional[Dict[str, str]] = None,
  208. column_formats: Optional[Dict[str, Union[str, Dict[str, str]]]] = None,
  209. encoding: Optional[str] = None,
  210. use_none: Optional[bool] = None,
  211. context: QueryContext = None,
  212. query_tz: Optional[Union[str, tzinfo]] = None,
  213. column_tzs: Optional[Dict[str, Union[str, tzinfo]]] = None,
  214. external_data: Optional[ExternalData] = None) -> StreamContext:
  215. """
  216. Variation of main query method that returns a stream of row oriented blocks. For
  217. parameters, see the create_query_context method.
  218. :return: StreamContext -- Iterable stream context that returns blocks of rows
  219. """
  220. return self._context_query(locals(), use_numpy=False, streaming=True).row_block_stream
  221. def query_rows_stream(self,
  222. query: Optional[str] = None,
  223. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  224. settings: Optional[Dict[str, Any]] = None,
  225. query_formats: Optional[Dict[str, str]] = None,
  226. column_formats: Optional[Dict[str, Union[str, Dict[str, str]]]] = None,
  227. encoding: Optional[str] = None,
  228. use_none: Optional[bool] = None,
  229. context: QueryContext = None,
  230. query_tz: Optional[Union[str, tzinfo]] = None,
  231. column_tzs: Optional[Dict[str, Union[str, tzinfo]]] = None,
  232. external_data: Optional[ExternalData] = None) -> StreamContext:
  233. """
  234. Variation of main query method that returns a stream of row oriented blocks. For
  235. parameters, see the create_query_context method.
  236. :return: StreamContext -- Iterable stream context that returns blocks of rows
  237. """
  238. return self._context_query(locals(), use_numpy=False, streaming=True).rows_stream
  239. @abstractmethod
  240. def raw_query(self, query: str,
  241. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  242. settings: Optional[Dict[str, Any]] = None,
  243. fmt: str = None,
  244. use_database: bool = True,
  245. external_data: Optional[ExternalData] = None) -> bytes:
  246. """
  247. Query method that simply returns the raw ClickHouse format bytes
  248. :param query: Query statement/format string
  249. :param parameters: Optional dictionary used to format the query
  250. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  251. :param fmt: ClickHouse output format
  252. :param use_database Send the database parameter to ClickHouse so the command will be executed in the client
  253. database context.
  254. :param external_data External data to send with the query
  255. :return: bytes representing raw ClickHouse return value based on format
  256. """
  257. @abstractmethod
  258. def raw_stream(self, query: str,
  259. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  260. settings: Optional[Dict[str, Any]] = None,
  261. fmt: str = None,
  262. use_database: bool = True,
  263. external_data: Optional[ExternalData] = None) -> io.IOBase:
  264. """
  265. Query method that returns the result as an io.IOBase iterator
  266. :param query: Query statement/format string
  267. :param parameters: Optional dictionary used to format the query
  268. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  269. :param fmt: ClickHouse output format
  270. :param use_database Send the database parameter to ClickHouse so the command will be executed in the client
  271. database context.
  272. :param external_data External data to send with the query
  273. :return: io.IOBase stream/iterator for the result
  274. """
  275. # pylint: disable=duplicate-code,too-many-arguments,unused-argument
  276. def query_np(self,
  277. query: Optional[str] = None,
  278. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  279. settings: Optional[Dict[str, Any]] = None,
  280. query_formats: Optional[Dict[str, str]] = None,
  281. column_formats: Optional[Dict[str, str]] = None,
  282. encoding: Optional[str] = None,
  283. use_none: Optional[bool] = None,
  284. max_str_len: Optional[int] = None,
  285. context: QueryContext = None,
  286. external_data: Optional[ExternalData] = None):
  287. """
  288. Query method that returns the results as a numpy array. For parameter values, see the
  289. create_query_context method
  290. :return: Numpy array representing the result set
  291. """
  292. return self._context_query(locals(), use_numpy=True).np_result
  293. # pylint: disable=duplicate-code,too-many-arguments,unused-argument
  294. def query_np_stream(self,
  295. query: Optional[str] = None,
  296. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  297. settings: Optional[Dict[str, Any]] = None,
  298. query_formats: Optional[Dict[str, str]] = None,
  299. column_formats: Optional[Dict[str, str]] = None,
  300. encoding: Optional[str] = None,
  301. use_none: Optional[bool] = None,
  302. max_str_len: Optional[int] = None,
  303. context: QueryContext = None,
  304. external_data: Optional[ExternalData] = None) -> StreamContext:
  305. """
  306. Query method that returns the results as a stream of numpy arrays. For parameter values, see the
  307. create_query_context method
  308. :return: Generator that yield a numpy array per block representing the result set
  309. """
  310. return self._context_query(locals(), use_numpy=True, streaming=True).np_stream
  311. # pylint: disable=duplicate-code,too-many-arguments,unused-argument
  312. def query_df(self,
  313. query: Optional[str] = None,
  314. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  315. settings: Optional[Dict[str, Any]] = None,
  316. query_formats: Optional[Dict[str, str]] = None,
  317. column_formats: Optional[Dict[str, str]] = None,
  318. encoding: Optional[str] = None,
  319. use_none: Optional[bool] = None,
  320. max_str_len: Optional[int] = None,
  321. use_na_values: Optional[bool] = None,
  322. query_tz: Optional[str] = None,
  323. column_tzs: Optional[Dict[str, Union[str, tzinfo]]] = None,
  324. context: QueryContext = None,
  325. external_data: Optional[ExternalData] = None,
  326. use_extended_dtypes: Optional[bool] = None):
  327. """
  328. Query method that results the results as a pandas dataframe. For parameter values, see the
  329. create_query_context method
  330. :return: Pandas dataframe representing the result set
  331. """
  332. return self._context_query(locals(), use_numpy=True, as_pandas=True).df_result
  333. # pylint: disable=duplicate-code,too-many-arguments,unused-argument
  334. def query_df_stream(self,
  335. query: Optional[str] = None,
  336. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  337. settings: Optional[Dict[str, Any]] = None,
  338. query_formats: Optional[Dict[str, str]] = None,
  339. column_formats: Optional[Dict[str, str]] = None,
  340. encoding: Optional[str] = None,
  341. use_none: Optional[bool] = None,
  342. max_str_len: Optional[int] = None,
  343. use_na_values: Optional[bool] = None,
  344. query_tz: Optional[str] = None,
  345. column_tzs: Optional[Dict[str, Union[str, tzinfo]]] = None,
  346. context: QueryContext = None,
  347. external_data: Optional[ExternalData] = None,
  348. use_extended_dtypes: Optional[bool] = None) -> StreamContext:
  349. """
  350. Query method that returns the results as a StreamContext. For parameter values, see the
  351. create_query_context method
  352. :return: Generator that yields a Pandas dataframe per block representing the result set
  353. """
  354. return self._context_query(locals(), use_numpy=True,
  355. as_pandas=True,
  356. streaming=True).df_stream
  357. def create_query_context(self,
  358. query: Optional[str] = None,
  359. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  360. settings: Optional[Dict[str, Any]] = None,
  361. query_formats: Optional[Dict[str, str]] = None,
  362. column_formats: Optional[Dict[str, Union[str, Dict[str, str]]]] = None,
  363. encoding: Optional[str] = None,
  364. use_none: Optional[bool] = None,
  365. column_oriented: Optional[bool] = None,
  366. use_numpy: Optional[bool] = False,
  367. max_str_len: Optional[int] = 0,
  368. context: Optional[QueryContext] = None,
  369. query_tz: Optional[Union[str, tzinfo]] = None,
  370. column_tzs: Optional[Dict[str, Union[str, tzinfo]]] = None,
  371. use_na_values: Optional[bool] = None,
  372. streaming: bool = False,
  373. as_pandas: bool = False,
  374. external_data: Optional[ExternalData] = None,
  375. use_extended_dtypes: Optional[bool] = None) -> QueryContext:
  376. """
  377. Creates or updates a reusable QueryContext object
  378. :param query: Query statement/format string
  379. :param parameters: Optional dictionary used to format the query
  380. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  381. :param query_formats: See QueryContext __init__ docstring
  382. :param column_formats: See QueryContext __init__ docstring
  383. :param encoding: See QueryContext __init__ docstring
  384. :param use_none: Use None for ClickHouse NULL instead of default values. Note that using None in Numpy
  385. arrays will force the numpy array dtype to 'object', which is often inefficient. This effect also
  386. will impact the performance of Pandas dataframes.
  387. :param column_oriented: Deprecated. Controls orientation of the QueryResult result_set property
  388. :param use_numpy: Return QueryResult columns as one-dimensional numpy arrays
  389. :param max_str_len: Limit returned ClickHouse String values to this length, which allows a Numpy
  390. structured array even with ClickHouse variable length String columns. If 0, Numpy arrays for
  391. String columns will always be object arrays
  392. :param context: An existing QueryContext to be updated with any provided parameter values
  393. :param query_tz Either a string or a pytz tzinfo object. (Strings will be converted to tzinfo objects).
  394. Values for any DateTime or DateTime64 column in the query will be converted to Python datetime.datetime
  395. objects with the selected timezone.
  396. :param column_tzs A dictionary of column names to tzinfo objects (or strings that will be converted to
  397. tzinfo objects). The timezone will be applied to datetime objects returned in the query
  398. :param use_na_values: Deprecated alias for use_advanced_dtypes
  399. :param as_pandas Return the result columns as pandas.Series objects
  400. :param streaming Marker used to correctly configure streaming queries
  401. :param external_data ClickHouse "external data" to send with query
  402. :param use_extended_dtypes: Only relevant to Pandas Dataframe queries. Use Pandas "missing types", such as
  403. pandas.NA and pandas.NaT for ClickHouse NULL values, as well as extended Pandas dtypes such as IntegerArray
  404. and StringArray. Defaulted to True for query_df methods
  405. :return: Reusable QueryContext
  406. """
  407. if context:
  408. return context.updated_copy(query=query,
  409. parameters=parameters,
  410. settings=settings,
  411. query_formats=query_formats,
  412. column_formats=column_formats,
  413. encoding=encoding,
  414. server_tz=self.server_tz,
  415. use_none=use_none,
  416. column_oriented=column_oriented,
  417. use_numpy=use_numpy,
  418. max_str_len=max_str_len,
  419. query_tz=query_tz,
  420. column_tzs=column_tzs,
  421. as_pandas=as_pandas,
  422. use_extended_dtypes=use_extended_dtypes,
  423. streaming=streaming,
  424. external_data=external_data)
  425. if use_numpy and max_str_len is None:
  426. max_str_len = 0
  427. if use_extended_dtypes is None:
  428. use_extended_dtypes = use_na_values
  429. if as_pandas and use_extended_dtypes is None:
  430. use_extended_dtypes = True
  431. return QueryContext(query=query,
  432. parameters=parameters,
  433. settings=settings,
  434. query_formats=query_formats,
  435. column_formats=column_formats,
  436. encoding=encoding,
  437. server_tz=self.server_tz,
  438. use_none=use_none,
  439. column_oriented=column_oriented,
  440. use_numpy=use_numpy,
  441. max_str_len=max_str_len,
  442. query_tz=query_tz,
  443. column_tzs=column_tzs,
  444. use_extended_dtypes=use_extended_dtypes,
  445. as_pandas=as_pandas,
  446. streaming=streaming,
  447. apply_server_tz=self.apply_server_timezone,
  448. external_data=external_data)
  449. def query_arrow(self,
  450. query: str,
  451. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  452. settings: Optional[Dict[str, Any]] = None,
  453. use_strings: Optional[bool] = None,
  454. external_data: Optional[ExternalData] = None):
  455. """
  456. Query method using the ClickHouse Arrow format to return a PyArrow table
  457. :param query: Query statement/format string
  458. :param parameters: Optional dictionary used to format the query
  459. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  460. :param use_strings: Convert ClickHouse String type to Arrow string type (instead of binary)
  461. :param external_data ClickHouse "external data" to send with query
  462. :return: PyArrow.Table
  463. """
  464. settings = self._update_arrow_settings(settings, use_strings)
  465. return to_arrow(self.raw_query(query,
  466. parameters,
  467. settings,
  468. fmt='Arrow',
  469. external_data=external_data))
  470. def query_arrow_stream(self,
  471. query: str,
  472. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  473. settings: Optional[Dict[str, Any]] = None,
  474. use_strings: Optional[bool] = None,
  475. external_data: Optional[ExternalData] = None) -> StreamContext:
  476. """
  477. Query method that returns the results as a stream of Arrow tables
  478. :param query: Query statement/format string
  479. :param parameters: Optional dictionary used to format the query
  480. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  481. :param use_strings: Convert ClickHouse String type to Arrow string type (instead of binary)
  482. :param external_data ClickHouse "external data" to send with query
  483. :return: Generator that yields a PyArrow.Table for per block representing the result set
  484. """
  485. settings = self._update_arrow_settings(settings, use_strings)
  486. return to_arrow_batches(self.raw_stream(query,
  487. parameters,
  488. settings,
  489. fmt='ArrowStream',
  490. external_data=external_data))
  491. def _update_arrow_settings(self,
  492. settings: Optional[Dict[str, Any]],
  493. use_strings: Optional[bool]) -> Dict[str, Any]:
  494. settings = dict_copy(settings)
  495. if self.database:
  496. settings['database'] = self.database
  497. str_status = self._setting_status(arrow_str_setting)
  498. if use_strings is None:
  499. if str_status.is_writable and not str_status.is_set:
  500. settings[arrow_str_setting] = '1' # Default to returning strings if possible
  501. elif use_strings != str_status.is_set:
  502. if not str_status.is_writable:
  503. raise OperationalError(f'Cannot change readonly {arrow_str_setting} to {use_strings}')
  504. settings[arrow_str_setting] = '1' if use_strings else '0'
  505. return settings
  506. @abstractmethod
  507. def command(self,
  508. cmd: str,
  509. parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
  510. data: Union[str, bytes] = None,
  511. settings: Dict[str, Any] = None,
  512. use_database: bool = True,
  513. external_data: Optional[ExternalData] = None) -> Union[str, int, Sequence[str], QuerySummary]:
  514. """
  515. Client method that returns a single value instead of a result set
  516. :param cmd: ClickHouse query/command as a python format string
  517. :param parameters: Optional dictionary of key/values pairs to be formatted
  518. :param data: Optional 'data' for the command (for INSERT INTO in particular)
  519. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  520. :param use_database: Send the database parameter to ClickHouse so the command will be executed in the client
  521. database context. Otherwise, no database will be specified with the command. This is useful for determining
  522. the default user database
  523. :param external_data ClickHouse "external data" to send with command/query
  524. :return: Decoded response from ClickHouse as either a string, int, or sequence of strings, or QuerySummary
  525. if no data returned
  526. """
  527. @abstractmethod
  528. def ping(self) -> bool:
  529. """
  530. Validate the connection, does not throw an Exception (see debug logs)
  531. :return: ClickHouse server is up and reachable
  532. """
  533. # pylint: disable=too-many-arguments
  534. def insert(self,
  535. table: Optional[str] = None,
  536. data: Sequence[Sequence[Any]] = None,
  537. column_names: Union[str, Iterable[str]] = '*',
  538. database: Optional[str] = None,
  539. column_types: Sequence[ClickHouseType] = None,
  540. column_type_names: Sequence[str] = None,
  541. column_oriented: bool = False,
  542. settings: Optional[Dict[str, Any]] = None,
  543. context: InsertContext = None) -> QuerySummary:
  544. """
  545. Method to insert multiple rows/data matrix of native Python objects. If context is specified arguments
  546. other than data are ignored
  547. :param table: Target table
  548. :param data: Sequence of sequences of Python data
  549. :param column_names: Ordered list of column names or '*' if column types should be retrieved from the
  550. ClickHouse table definition
  551. :param database: Target database -- will use client default database if not specified.
  552. :param column_types: ClickHouse column types. If set then column data does not need to be retrieved from
  553. the server
  554. :param column_type_names: ClickHouse column type names. If set then column data does not need to be
  555. retrieved from the server
  556. :param column_oriented: If true the data is already "pivoted" in column form
  557. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  558. :param context: Optional reusable insert context to allow repeated inserts into the same table with
  559. different data batches
  560. :return: QuerySummary with summary information, throws exception if insert fails
  561. """
  562. if (context is None or context.empty) and data is None:
  563. raise ProgrammingError('No data specified for insert') from None
  564. if context is None:
  565. context = self.create_insert_context(table,
  566. column_names,
  567. database,
  568. column_types,
  569. column_type_names,
  570. column_oriented,
  571. settings)
  572. if data is not None:
  573. if not context.empty:
  574. raise ProgrammingError('Attempting to insert new data with non-empty insert context') from None
  575. context.data = data
  576. return self.data_insert(context)
  577. def insert_df(self, table: str = None,
  578. df=None,
  579. database: Optional[str] = None,
  580. settings: Optional[Dict] = None,
  581. column_names: Optional[Sequence[str]] = None,
  582. column_types: Sequence[ClickHouseType] = None,
  583. column_type_names: Sequence[str] = None,
  584. context: InsertContext = None) -> QuerySummary:
  585. """
  586. Insert a pandas DataFrame into ClickHouse. If context is specified arguments other than df are ignored
  587. :param table: ClickHouse table
  588. :param df: two-dimensional pandas dataframe
  589. :param database: Optional ClickHouse database
  590. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  591. :param column_names: An optional list of ClickHouse column names. If not set, the DataFrame column names
  592. will be used
  593. :param column_types: ClickHouse column types. If set then column data does not need to be retrieved from
  594. the server
  595. :param column_type_names: ClickHouse column type names. If set then column data does not need to be
  596. retrieved from the server
  597. :param context: Optional reusable insert context to allow repeated inserts into the same table with
  598. different data batches
  599. :return: QuerySummary with summary information, throws exception if insert fails
  600. """
  601. if context is None:
  602. if column_names is None:
  603. column_names = df.columns
  604. elif len(column_names) != len(df.columns):
  605. raise ProgrammingError('DataFrame column count does not match insert_columns') from None
  606. return self.insert(table,
  607. df,
  608. column_names,
  609. database,
  610. column_types=column_types,
  611. column_type_names=column_type_names,
  612. settings=settings, context=context)
  613. def insert_arrow(self, table: str,
  614. arrow_table, database: str = None,
  615. settings: Optional[Dict] = None) -> QuerySummary:
  616. """
  617. Insert a PyArrow table DataFrame into ClickHouse using raw Arrow format
  618. :param table: ClickHouse table
  619. :param arrow_table: PyArrow Table object
  620. :param database: Optional ClickHouse database
  621. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  622. :return: QuerySummary with summary information, throws exception if insert fails
  623. """
  624. full_table = table if '.' in table or not database else f'{database}.{table}'
  625. column_names, insert_block = arrow_buffer(arrow_table)
  626. return self.raw_insert(full_table, column_names, insert_block, settings, 'Arrow')
  627. def create_insert_context(self,
  628. table: str,
  629. column_names: Optional[Union[str, Sequence[str]]] = None,
  630. database: Optional[str] = None,
  631. column_types: Sequence[ClickHouseType] = None,
  632. column_type_names: Sequence[str] = None,
  633. column_oriented: bool = False,
  634. settings: Optional[Dict[str, Any]] = None,
  635. data: Optional[Sequence[Sequence[Any]]] = None) -> InsertContext:
  636. """
  637. Builds a reusable insert context to hold state for a duration of an insert
  638. :param table: Target table
  639. :param database: Target database. If not set, uses the client default database
  640. :param column_names: Optional ordered list of column names. If not set, all columns ('*') will be assumed
  641. in the order specified by the table definition
  642. :param database: Target database -- will use client default database if not specified
  643. :param column_types: ClickHouse column types. Optional Sequence of ClickHouseType objects. If neither column
  644. types nor column type names are set, actual column types will be retrieved from the server.
  645. :param column_type_names: ClickHouse column type names. Specified column types by name string
  646. :param column_oriented: If true the data is already "pivoted" in column form
  647. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  648. :param data: Initial dataset for insert
  649. :return Reusable insert context
  650. """
  651. full_table = table
  652. if '.' not in table:
  653. if database:
  654. full_table = f'{quote_identifier(database)}.{quote_identifier(table)}'
  655. else:
  656. full_table = quote_identifier(table)
  657. column_defs = []
  658. if column_types is None and column_type_names is None:
  659. describe_result = self.query(f'DESCRIBE TABLE {full_table}')
  660. column_defs = [ColumnDef(**row) for row in describe_result.named_results()
  661. if row['default_type'] not in ('ALIAS', 'MATERIALIZED')]
  662. if column_names is None or isinstance(column_names, str) and column_names == '*':
  663. column_names = [cd.name for cd in column_defs]
  664. column_types = [cd.ch_type for cd in column_defs]
  665. elif isinstance(column_names, str):
  666. column_names = [column_names]
  667. if len(column_names) == 0:
  668. raise ValueError('Column names must be specified for insert')
  669. if not column_types:
  670. if column_type_names:
  671. column_types = [get_from_name(name) for name in column_type_names]
  672. else:
  673. column_map = {d.name: d for d in column_defs}
  674. try:
  675. column_types = [column_map[name].ch_type for name in column_names]
  676. except KeyError as ex:
  677. raise ProgrammingError(f'Unrecognized column {ex} in table {table}') from None
  678. if len(column_names) != len(column_types):
  679. raise ProgrammingError('Column names do not match column types') from None
  680. return InsertContext(full_table,
  681. column_names,
  682. column_types,
  683. column_oriented=column_oriented,
  684. settings=settings,
  685. data=data)
  686. def min_version(self, version_str: str) -> bool:
  687. """
  688. Determine whether the connected server is at least the submitted version
  689. For Altinity Stable versions like 22.8.15.25.altinitystable
  690. the last condition in the first list comprehension expression is added
  691. :param version_str: A version string consisting of up to 4 integers delimited by dots
  692. :return: True if version_str is greater than the server_version, False if less than
  693. """
  694. try:
  695. server_parts = [int(x) for x in self.server_version.split('.') if x.isnumeric()]
  696. server_parts.extend([0] * (4 - len(server_parts)))
  697. version_parts = [int(x) for x in version_str.split('.')]
  698. version_parts.extend([0] * (4 - len(version_parts)))
  699. except ValueError:
  700. logger.warning('Server %s or requested version %s does not match format of numbers separated by dots',
  701. self.server_version, version_str)
  702. return False
  703. for x, y in zip(server_parts, version_parts):
  704. if x > y:
  705. return True
  706. if x < y:
  707. return False
  708. return True
  709. @abstractmethod
  710. def data_insert(self, context: InsertContext) -> QuerySummary:
  711. """
  712. Subclass implementation of the data insert
  713. :context: InsertContext parameter object
  714. :return: No return, throws an exception if the insert fails
  715. """
  716. @abstractmethod
  717. def raw_insert(self, table: str,
  718. column_names: Optional[Sequence[str]] = None,
  719. insert_block: Union[str, bytes, Generator[bytes, None, None], BinaryIO] = None,
  720. settings: Optional[Dict] = None,
  721. fmt: Optional[str] = None,
  722. compression: Optional[str] = None) -> QuerySummary:
  723. """
  724. Insert data already formatted in a bytes object
  725. :param table: Table name (whether qualified with the database name or not)
  726. :param column_names: Sequence of column names
  727. :param insert_block: Binary or string data already in a recognized ClickHouse format
  728. :param settings: Optional dictionary of ClickHouse settings (key/string values)
  729. :param compression: Recognized ClickHouse `Accept-Encoding` header compression value
  730. :param fmt: Valid clickhouse format
  731. """
  732. def close(self):
  733. """
  734. Subclass implementation to close the connection to the server/deallocate the client
  735. """
  736. def _context_query(self, lcls: dict, **overrides):
  737. kwargs = lcls.copy()
  738. kwargs.pop('self')
  739. kwargs.update(overrides)
  740. return self._query_with_context((self.create_query_context(**kwargs)))
  741. def __enter__(self):
  742. return self
  743. def __exit__(self, exc_type, exc_value, exc_traceback):
  744. self.close()