o
EDi} @ s d dl Z d dlmZ d dlmZ d dlmZ d dlZd dlmZ ddl m
Z
m Z ddl m
Z
mZ dd lmZ dd
lmZ dd lmZ dd lmZmZmZmZ dd
lmZ ddl m Z m Z m Z G dd de Z!dS ) N)deque)contextmanager)time)urlparse )errorsdefines)ColumnOrientedBlockRowOrientedBlock)
Connection) log_block)ServerPacketTypes)IterQueryResultProgressQueryResult QueryResult QueryInfo)
escape_params)
column_chunkschunks parse_urlc sr e Zd ZdZdZ fddZdd Zdd Zd d
Zd d Z d
d Z
dd Z dGddZ dHddZ
dd Zdd Zdd Zd d Zd d Zed d! Z " dId#d$Z " dId%d&Z " 'dJd(d)Z " *dKd+d,Z " "dLd-d.Z " dMd/d0Z " dMd1d2Z " dNd3d4Z " dOd5d6Z d7d8 Z dPd9d:Z d;d< Z d=d> Z!d?d@ Z"dHdAdBZ#dCdD Z$e%dEdF Z& Z'S )QClienta
Client for communication with the ClickHouse server.
Single connection is established per each connected instance of the client.
:param settings: Dictionary of settings that passed to every query (except
for the client settings, see below). Defaults to ``None``
(no additional settings). See all available settings in
`ClickHouse docs
`_.
:param \**kwargs: All other args are passed to the
:py:class:`~clickhouse_driver.connection.Connection`
constructor.
The following keys when passed in ``settings`` are used for configuring the
client itself:
* ``insert_block_size`` -- chunk size to split rows for ``INSERT``.
Defaults to ``1048576``.
* ``strings_as_bytes`` -- turns off string column encoding/decoding.
* ``strings_encoding`` -- specifies string encoding. UTF-8 by default.
* ``use_numpy`` -- Use NumPy for columns reading. New in version
*0.2.0*.
* ``opentelemetry_traceparent`` -- OpenTelemetry traceparent header as
described by W3C Trace Context recommendation.
New in version *0.2.2*.
* ``opentelemetry_tracestate`` -- OpenTelemetry tracestate header as
described by W3C Trace Context recommendation.
New in version *0.2.2*.
* ``quota_key`` -- A string to differentiate quotas when the user have
keyed quotas configured on server.
New in version *0.2.3*.
* ``input_format_null_as_default`` -- Initialize null fields with
default values if data type of this field is not
nullable. Does not work for NumPy. Default: False.
New in version *0.2.4*.
* ``round_robin`` -- If ``alt_hosts`` are provided the query will be
executed on host picked with round-robin algorithm.
New in version *0.2.5*.
* ``namedtuple_as_json`` -- Controls named tuple and nested types
deserialization. To interpret these column alongside
with ``allow_experimental_object_type=1`` as Python
tuple set ``namedtuple_as_json`` to ``False``.
Default: True.
New in version *0.2.6*.
* ``server_side_params`` -- Species on which side query parameters
should be rendered into placeholders.
Default: False. Means that parameters are rendered
on driver's side.
New in version *0.2.7*.
insert_block_sizestrings_as_bytesstrings_encoding use_numpyopentelemetry_traceparentopentelemetry_tracestate quota_key input_format_null_as_defaultnamedtuple_as_jsonserver_side_paramsc
s | dd pi | _t| j dtj| j dd| j dtj| j dd| j dd | j dd | j d
d | j d d| j d d
| j ddd
| _| jd ruzddlm }m
}m } || _ || _
|| _W n tyt tdw t| _ t| _
t| _| dd}tt|i |g| _|rd|v r| d}|dD ]O}td| } | }
t|} | dkr| j| jf|dd } n!| dkr| jf|dd } | j|
d< n
t } | j|
d< | j|
d< t| i |
}
| j |
q| | _ | t!t"| # d S )Nsettingsr r Fr r r r r r r Tr! r r )NumpyIterQueryResultNumpyProgressQueryResultNumpyQueryResult"Extras for NumPy must be installed round_robin alt_hosts,z
clickhouse:// porthost)$popcopyr" intr DEFAULT_INSERT_BLOCK_SIZESTRINGS_ENCODINGclient_settings numpy.resultr$ r% r& query_result_clsiter_query_result_clsprogress_query_result_cls ImportError RuntimeErrorr r r r r connectionssplitr lenhostnamer, tupleappendget_connection
connectionreset_last_querysuperr __init__)selfargskwargsr$ r% r& r( r) r- urlconnection_kwargsnum_argsconnection_argsrA __class__ Q/var/www/Datamplify/venv/lib/python3.10/site-packages/clickhouse_driver/client.pyrD U s
!
zClient.__init__c C s | S NrN rE rN rN rO __enter__ s zClient.__enter__c C s | d S rP )
disconnect)rE exc_typeexc_valexc_tbrN rN rO __exit__ s zClient.__exit__c C s: t | dr | j| j | j }| j|j_| j|j_|S )NrA )hasattrr: r? rA popleftr" contextr3 rE rA rN rN rO r@ s
zClient.get_connectionc C s | | jD ]}| qd S rP )disconnect_connectionr: rS r[ rN rN rO rS s
zClient.disconnectc C s | j | dS )z.
Disconnects from the server.
N)rA rS rB rQ rN rN rO r\ s
z Client.disconnect_connectionc C s
d | _ d S rP )
last_queryrQ rN rN rO rB s
zClient.reset_last_queryFc C s4 | }|r| j|||dS | j|||d}| S )Nwith_column_typescolumnar)packet_generatorr7 r5
get_result)rE r_ progressr` genresultrN rN rO receive_result s zClient.receive_resultc c s6 | }| j||d}|D ]
}|D ]}|V qqd S Nr_ )ra r6 )rE r_ rd re rowsrowrN rN rO iter_receive_result s zClient.iter_receive_resultc c sL z| }|s W d S |du rW q|V W n ttfy$ | w q)NT)receive_packet ExceptionKeyboardInterruptrS rE packetrN rN rO ra s
zClient.packet_generatorc C s | j }|jtjkr|j|jtjkr | j|j |S |jtj
kr%dS |jtj kr-|S |jtj kr5|S |jtj
kr=|S |jtjkrL| j|j dS dS )NFT)rA rl typer
EXCEPTION exceptionPROGRESSr] store_progressrc
END_OF_STREAMDATATOTALSEXTREMES PROFILE_INFO
store_profile profile_inforo rN rN rO rl s$
zClient.receive_packetc C sd t |pi }| j }| jD ]
}||v r||||< q|| jj_| j }|| || jj_d S rP ) dictr3 r/ available_client_settingsr. rA rZ r" update)rE r" r3 keyquery_settingsrN rN rO make_query_settings s
zClient.make_query_settingsc C s4 | d}| dr|dd | j_d S d S )Nz; zuse )striplower
startswithrA database)rE queryrN rN rO track_current_database s
z Client.track_current_databasec
C s t | j}t| dr|d7 }t|D ]-}z| | _| | | j t | _ W d S t
j t
j fy? ||d k r>Y q w d S )NrA r )
r< r: rX ranger@ rA r
force_connectr r] r SocketTimeoutError NetworkError)rE r" num_connectionsirN rN rO establish_connection# s
zClient.establish_connectionc c sJ z| | d | jj_d V | | W d S ttfy$ | w rP )r rA server_infosession_timezoner rm rn rS )rE r r" rN rN rO disconnect_on_error6 s
zClient.disconnect_on_errorNc
C s t } | ||6 t|tttjf}
|
r!| j||||||d} n | j|||||||d} | j
t | | W d S 1 sBw Y dS )a
Executes query.
Establishes new connection if it wasn't established yet.
After query execution connection remains intact for next queries.
If connection can't be reused it will be closed and new connection will
be created.
:param query: query that will be send to server.
:param params: substitution parameters for SELECT queries and data for
INSERT queries. Data for INSERT can be `list`, `tuple`
or :data:`~types.GeneratorType`.
Defaults to ``None`` (no parameters or data).
:param with_column_types: if specified column names and types will be
returned alongside with result.
Defaults to ``False``.
:param external_tables: external tables to send.
Defaults to ``None`` (no external tables).
:param query_id: the query identifier. If no query id specified
ClickHouse server will generate it.
:param settings: dictionary of query settings.
Defaults to ``None`` (no additional settings).
:param types_check: enables type checking of data for INSERT queries.
Causes additional overhead. Defaults to ``False``.
:param columnar: if specified the result of the SELECT query will be
returned in column-oriented form.
It also allows to INSERT data in columnar form.
Defaults to ``False`` (row-like form).
:return: * number of inserted rows for INSERT queries with data.
Returning rows count from INSERT FROM SELECT is not
supported.
* if `with_column_types=False`: `list` of `tuples` with
rows/columns.
* if `with_column_types=True`: `tuple` of 2 elements:
* The first element is `list` of `tuples` with
rows/columns.
* The second element information is about columns: names
and types.
)external_tablesquery_id types_checkr` paramsr_ r r r r` N) r r
isinstancelistr> types
GeneratorTypeprocess_insert_queryprocess_ordinary_queryr]
store_elapsed) rE r r r_ r r r" r r`
start_time is_insertrvrN rN rO executeD s$ ,$zClient.executec
C sH | || | j|||||||dW d S 1 s w Y dS )a
Executes SELECT query with progress information.
See, :ref:`execute-with-progress`.
:param query: query that will be send to server.
:param params: substitution parameters for SELECT queries and data for
INSERT queries. Data for INSERT can be `list`, `tuple`
or :data:`~types.GeneratorType`.
Defaults to ``None`` (no parameters or data).
:param with_column_types: if specified column names and types will be
returned alongside with result.
Defaults to ``False``.
:param external_tables: external tables to send.
Defaults to ``None`` (no external tables).
:param query_id: the query identifier. If no query id specified
ClickHouse server will generate it.
:param settings: dictionary of query settings.
Defaults to ``None`` (no additional settings).
:param types_check: enables type checking of data for INSERT queries.
Causes additional overhead. Defaults to ``False``.
:param columnar: if specified the result will be returned in
column-oriented form.
Defaults to ``False`` (row-like form).
:return: :ref:`progress-query-result` proxy.
r N)r $process_ordinary_query_with_progress) rE r r r_ r r r" r r` rN rN rO execute_with_progress s $z Client.execute_with_progressr c
C s\ | || | j||||||d} |dkrt| |n| W d S 1 s'w Y dS )a
*New in version 0.0.14.*
Executes SELECT query with results streaming. See, :ref:`execute-iter`.
:param query: query that will be send to server.
:param params: substitution parameters for SELECT queries and data for
INSERT queries. Data for INSERT can be `list`, `tuple`
or :data:`~types.GeneratorType`.
Defaults to ``None`` (no parameters or data).
:param with_column_types: if specified column names and types will be
returned alongside with result.
Defaults to ``False``.
:param external_tables: external tables to send.
Defaults to ``None`` (no external tables).
:param query_id: the query identifier. If no query id specified
ClickHouse server will generate it.
:param settings: dictionary of query settings.
Defaults to ``None`` (no additional settings).
:param types_check: enables type checking of data for INSERT queries.
Causes additional overhead. Defaults to ``False``.
:param chunk_size: chunk query results.
:return: :ref:`iter-query-result` proxy.
)r r_ r r r r N)r iter_process_ordinary_queryr )
rE r r r_ r r r" r
chunk_sizer rN rN rO execute_iter s $zClient.execute_iterTc
C s~ zddl }W n ty tdw | j|dd||||d\}} dd | D } |r0dd | D } |jd d
t|| D | d S ) aQ
*New in version 0.2.0.*
Queries DataFrame with specified SELECT query.
:param query: query that will be send to server.
:param params: substitution parameters.
Defaults to ``None`` (no parameters or data).
:param external_tables: external tables to send.
Defaults to ``None`` (no external tables).
:param query_id: the query identifier. If no query id specified
ClickHouse server will generate it.
:param settings: dictionary of query settings.
Defaults to ``None`` (no additional settings).
:param replace_nonwords: boolean to replace non-words in column names
to underscores. Defaults to ``True``.
:return: pandas DataFrame.
r Nr' T)r` r_ r r r r" c S s g | ]\}}|qS rN rN ).0nametype_rN rN rO
z*Client.query_dataframe..c S s g | ] }t d d|qS )z\W_)resubr xrN rN rO r s c S s i | ]\}}||qS rN rN )r dcolrN rN rO
z*Client.query_dataframe..)columns)pandasr8 r9 r DataFramezip)
rE r r r r r" replace_nonwordspddatar rN rN rO query_dataframe s
zClient.query_dataframec s zddl }W n ty tdw t }| ||^ | jj||d | j| | }d} |rgdd |j D }
t
|
t
j } t | rRd} t
| t| fdd|
D }
| j||
d d
} | | jt | | W d S 1 s|w Y dS ) ay
*New in version 0.2.0.*
Inserts pandas DataFrame with specified query.
:param query: query that will be send to server.
:param dataframe: pandas DataFrame.
:param external_tables: external tables to send.
Defaults to ``None`` (no external tables).
:param query_id: the query identifier. If no query id specified
ClickHouse server will generate it.
:param settings: dictionary of query settings.
Defaults to ``None`` (no additional settings).
:return: number of inserted rows.
r Nr' r c S s g | ]}|d qS )r rN r rN rN rO r r z+Client.insert_dataframe..z&DataFrame missing required columns: {}c s g | ]} | j qS rN )values)r column dataframerN rO r ! r T)r` )r r8 r9 r r rA
send_querysend_external_tablesreceive_sample_blockcolumns_with_typessetr r<
ValueErrorformatr send_datareceive_end_of_queryr] r )rE r r r r r" r r sample_blockr r diffmsgr rN r rO insert_dataframe s. $zClient.insert_dataframec C sL |d ur
| ||| jj}| jj|||d | jj||d | j|d|dS )Nr r r T)r_ rc r` substitute_paramsrA rZ r r rf rE r r r_ r r r r` rN rN rO r ( s
z+Client.process_ordinary_query_with_progressc C sJ |d ur
| ||| jj}| jj|||d | jj||d | j||dS )Nr r r^ r r rN rN rO r 8 s
z Client.process_ordinary_queryc C sH |d ur
| ||| jj}| jj|||d | jj||d | j|dS )Nr r rh )r rA rZ r r rk )rE r r r_ r r r rN rN rO r G s
z"Client.iter_process_ordinary_queryc C sN | j j||d | j j||d | }|r%| j||||d}| |S d S )Nr r )r r` )rA r r r r receive_end_of_insert_query) rE query_without_datar r r r r` r r rN rN rO r V s zClient.process_insert_queryc C sr | j }|jtjkr|jS |jtjkr|j|jtjkr$t |j n|jtj
kr+n
| j d|j}t
|q)NTz$Data, Exception, Log or TableColumns)rA rl rq r
rw blockrr rs LOGr
TABLE_COLUMNSunexpected_packet_messager UnexpectedPacketFromServerErrorrE rp messagerN rN rO r d s
zClient.receive_sample_blockc C s d}| j jj}|r tnt}|d r.zddlm} |r |} ntdW n ty- t dw |r2tnt
} | ||d D ]}
||j |
|d} | j | || j
7 }| q;| j | | |S ) Nr r r )r z0NumPy inserts is only allowed with columnar=Truer' r r )rA rZ r3 r r
numpy.helpersr r r8 r9 r r r num_rowsreceive_profile_events) rE r r r r`
inserted_rowsr3 block_clsnumpy_column_chunksslicerchunkr rN rN rO r z s4
zClient.send_datac C s | j }|jtjkrd S |jtjkr | j|j n7|jtj kr%|j
|jtj kr1t |j
n"|jtjkr8n|jtjkrF| j|j n
| j d|j}t|q)NTzDException, EndOfStream, Progress, TableColumns, ProfileEvents or Log)rA rl rq r
rv rt r] ru rc rr rs r r r r PROFILE_EVENTSr{ r| r r r r rN rN rO r s(
zClient.receive_end_of_queryc C s~ | j }|jtjkrd S |jtjkrt|j n$|jtjkr(| j
|j n|jtj kr1|j
| j d|j}t|q)NTz'EndOfStream, Log, Progress or Exception)rA rl rq r
rv r r r rt r] ru rc rr rs r r r r rN rN rO r s
z"Client.receive_end_of_insert_queryc C s | j jj}|tjk r d S | j }|jtjkr!| j
|j d S |jtj kr/| j
|j n)|jtjkr;t|j n |jtjkrD|j|jtjkrKn
| j d|j}t|q
)NTz9ProfileEvents, Progress, Log, Exception or TimezoneUpdate)rA r
used_revisionr 7DBMS_MIN_PROTOCOL_VERSION_WITH_PROFILE_EVENTS_IN_INSERTrl rq r
r r] r{ r| rt ru rc r r r rr rs TIMEZONE_UPDATEr r r )rE revisionrp r rN rN rO r s0
z Client.receive_profile_eventsc C s | j | j|dS rg )rA send_cancelrf )rE r_ rN rN rO cancel s
z
Client.cancelc C s6 | j jjd r |S t|tstdt||}|| S )a
Substitutes parameters into a provided query.
For example::
client = Client(...)
substituted_query = client.substitute_params(
query='SELECT 1234, %(foo)s',
params={'foo': 'bar'},
context=client.connection.context
)
# prints: SELECT 1234, 'bar'
print(substituted_query)
r! z$Parameters are expected in dict form)rA rZ r3 r r} r r )rE r r rZ escapedrN rN rO r s
zClient.substitute_paramsc C s t |\}}| |fi |S )a
Return a client configured from the given URL.
For example::
clickhouse://[user:password]@localhost:9000/default
clickhouses://[user:password]@localhost:9440/default
Three URL schemes are supported:
* clickhouse:// creates a normal TCP socket connection
* clickhouses:// creates a SSL wrapped TCP socket connection
Any additional querystring arguments will be passed along to
the Connection class's initializer.
)r )clsrH r- rG rN rN rO from_url s zClient.from_url)FFF)F)NFNNNFF)NFNNNFr )NNNNT)NNN)NFNNFF)NFNNF)NNFF)FF)(__name__
__module__ __qualname____doc__r~ rD rR rW r@ rS r\ rB rf rk ra rl r r r r r r r r r r r r r r r r r r r r r classmethodr
__classcell__rN rN rL rO r s~ 3
R
D
&
%
*
0
(
"r )"r collectionsr
contextlibr r r urllib.parser r# r r r r r
rA r logr protocolr
re r r r r util.escaper util.helpersr r r objectr rN rN rN rO s