blob: 397c52e16201b880714aeb39221e26c11bf56a11 [file] [log] [blame]
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
This module contains `Client` class, that lets you communicate with Apache
Ignite cluster node by the means of Ignite binary client protocol.
To start the communication, you may connect to the node of their choice
by instantiating the `Client` object and calling
:py:meth:`~pyignite.connection.Connection.connect` method with proper
parameters.
The whole storage room of Ignite cluster is split up into named structures,
called caches. For accessing the particular cache in key-value style
(a-la Redis or memcached) you should first create
the :class:`~pyignite.cache.Cache` object by calling
:py:meth:`~pyignite.client.Client.create_cache` or
:py:meth:`~pyignite.client.Client.get_or_create_cache()` methods, than call
:class:`~pyignite.cache.Cache` methods. If you wish to create a cache object
without communicating with server, there is also a
:py:meth:`~pyignite.client.Client.get_cache()` method that does just that.
For using Ignite SQL, call :py:meth:`~pyignite.client.Client.sql` method.
It returns a generator with result rows.
:py:meth:`~pyignite.client.Client.register_binary_type` and
:py:meth:`~pyignite.client.Client.query_binary_type` methods operates
the local (class-wise) registry for Ignite Complex objects.
"""
import time
from collections import defaultdict, OrderedDict
import random
import re
from itertools import chain
from typing import Iterable, Type, Union, Any, Dict, Optional, Sequence
from .api import cache_get_node_partitions
from .api.binary import get_binary_type, put_binary_type
from .api.cache_config import cache_get_names
from .cluster import Cluster
from .cursors import SqlFieldsCursor
from .cache import Cache, create_cache, get_cache, get_or_create_cache, BaseCache
from .connection import Connection
from .constants import IGNITE_DEFAULT_HOST, IGNITE_DEFAULT_PORT, PROTOCOL_BYTE_ORDER, AFFINITY_RETRIES, AFFINITY_DELAY
from .datatypes import BinaryObject, AnyDataObject, TransactionConcurrency, TransactionIsolation
from .datatypes.base import IgniteDataType
from .datatypes.internal import tc_map
from .exceptions import BinaryTypeError, CacheError, ReconnectError, connection_errors
from .queries.cache_info import CacheInfo
from .stream import BinaryStream, READ_BACKWARD
from .transaction import Transaction
from .utils import (
cache_id, capitalize, entity_id, schema_id, process_delimiter, status_to_exception, is_iterable,
get_field_by_id, unsigned
)
from .binary import GenericObjectMeta
from .monitoring import _EventListeners
__all__ = ['Client']
class BaseClient:
# used for Complex object data class names sanitizing
_identifier = re.compile(r'[^0-9a-zA-Z_.+$]', re.UNICODE)
_ident_start = re.compile(r'^[^a-zA-Z_]+', re.UNICODE)
def __init__(self, compact_footer: bool = None, partition_aware: bool = False,
event_listeners: Optional[Sequence] = None, **kwargs):
self._compact_footer = compact_footer
self._partition_aware = partition_aware
self._connection_args = kwargs
self._registry = defaultdict(dict)
self._nodes = []
self._current_node = 0
self._partition_aware = partition_aware
self.affinity_version = (0, 0)
self._affinity = {'version': self.affinity_version, 'partition_mapping': defaultdict(dict)}
self._protocol_context = None
self._event_listeners = _EventListeners(event_listeners)
@property
def protocol_context(self):
"""
Returns protocol context, or None, if no connection to the Ignite
cluster was not yet established.
This method is not a part of the public API. Unless you wish to
extend the `pyignite` capabilities (with additional testing, logging,
examining connections, et c.) you probably should not use it.
"""
return self._protocol_context
@protocol_context.setter
def protocol_context(self, value):
self._protocol_context = value
@property
def partition_aware(self):
return self._partition_aware and self.partition_awareness_supported_by_protocol
@property
def partition_awareness_supported_by_protocol(self):
return self.protocol_context is not None \
and self.protocol_context.is_partition_awareness_supported()
@property
def compact_footer(self) -> bool:
"""
This property remembers Complex object schema encoding approach when
decoding any Complex object, to use the same approach on Complex
object encoding.
:return: True if compact schema was used by server or no Complex
object decoding has yet taken place, False if full schema was used.
"""
# this is an ordinary object property, but its backing storage
# is a class attribute
# use compact schema by default, but leave initial (falsy) backing
# value unchanged
return self._compact_footer or self._compact_footer is None
@compact_footer.setter
def compact_footer(self, value: bool):
# normally schema approach should not change
if self._compact_footer not in (value, None):
raise Warning('Can not change client schema approach.')
else:
self._compact_footer = value
@staticmethod
def _process_connect_args(*args):
if len(args) == 0:
# no parameters − use default Ignite host and port
return [(IGNITE_DEFAULT_HOST, IGNITE_DEFAULT_PORT)]
if len(args) == 1 and is_iterable(args[0]):
# iterable of host-port pairs is given
return args[0]
if len(args) == 2 and isinstance(args[0], str) and isinstance(args[1], int):
# host and port are given
return [args]
raise ConnectionError('Connection parameters are not valid.')
def _process_get_binary_type_result(self, result):
if result.status != 0 or not result.value['type_exists']:
return result
binary_fields = result.value.pop('binary_fields')
old_format_schemas = result.value.pop('schema')
result.value['schemas'] = []
for s_id, field_ids in old_format_schemas.items():
result.value['schemas'].append(self._convert_schema(field_ids, binary_fields))
return result
@staticmethod
def _convert_type(tc_type: int):
try:
return tc_map(tc_type.to_bytes(1, PROTOCOL_BYTE_ORDER))
except (KeyError, OverflowError):
# if conversion to char or type lookup failed,
# we probably have a binary object type ID
return BinaryObject
def _convert_schema(self, field_ids: list, binary_fields: list) -> OrderedDict:
converted_schema = OrderedDict()
for field_id in field_ids:
binary_field = next(x for x in binary_fields if x['field_id'] == field_id)
converted_schema[binary_field['field_name']] = self._convert_type(binary_field['type_id'])
return converted_schema
@staticmethod
def _create_dataclass(type_name: str, schema: OrderedDict = None) -> Type:
"""
Creates default (generic) class for Ignite Complex object.
:param type_name: Complex object type name,
:param schema: Complex object schema,
:return: the resulting class.
"""
schema = schema or {}
return GenericObjectMeta(type_name, (), {}, schema=schema)
@classmethod
def _create_type_name(cls, type_name: str) -> str:
"""
Creates Python data class name from Ignite binary type name.
Handles all the special cases found in
`java.org.apache.ignite.binary.BinaryBasicNameMapper.simpleName()`.
Tries to adhere to PEP8 along the way.
"""
# general sanitizing
type_name = cls._identifier.sub('', type_name)
# - name ending with '$' (Scala)
# - name + '$' + some digits (anonymous class)
# - '$$Lambda$' in the middle
type_name = process_delimiter(type_name, '$')
# .NET outer/inner class delimiter
type_name = process_delimiter(type_name, '+')
# Java fully qualified class name
type_name = process_delimiter(type_name, '.')
# start chars sanitizing
type_name = capitalize(cls._ident_start.sub('', type_name))
return type_name
def _sync_binary_registry(self, type_id: int, type_info: dict):
"""
Sync binary registry
:param type_id: Complex object type ID.
:param type_info: Complex object type info.
"""
if type_info['type_exists']:
for schema in type_info['schemas']:
if not self._registry[type_id].get(schema_id(schema), None):
data_class = self._create_dataclass(
self._create_type_name(type_info['type_name']),
schema,
)
self._registry[type_id][schema_id(schema)] = data_class
def _get_from_registry(self, type_id, schema):
"""
Get binary type info from registry.
:param type_id: Complex object type ID.
:param schema: Complex object schema.
"""
if schema:
try:
return self._registry[type_id][schema_id(schema)]
except KeyError:
return None
return self._registry[type_id]
def register_cache(self, cache_id: int):
if self.partition_aware and cache_id not in self._affinity:
self._affinity['partition_mapping'][cache_id] = {}
def _get_affinity_key(self, cache_id, key, key_hint=None):
if key_hint is None:
key_hint = AnyDataObject.map_python_type(key)
cache_partition_mapping = self._cache_partition_mapping(cache_id)
if cache_partition_mapping and cache_partition_mapping.get('is_applicable'):
config = cache_partition_mapping.get('cache_config')
if config:
affinity_key_id = config.get(key_hint.type_id)
if affinity_key_id and isinstance(key, GenericObjectMeta):
return get_field_by_id(key, affinity_key_id)
return key, key_hint
def _update_affinity(self, full_affinity):
self._affinity['version'] = full_affinity['version']
full_mapping = full_affinity.get('partition_mapping')
if full_mapping:
self._affinity['partition_mapping'].update(full_mapping)
def _caches_to_update_affinity(self):
if self._affinity['version'] < self.affinity_version:
return list(self._affinity['partition_mapping'].keys())
else:
return list(c_id for c_id, c_mapping in self._affinity['partition_mapping'].items() if not c_mapping)
def _cache_partition_mapping(self, cache_id):
return self._affinity['partition_mapping'][cache_id]
def _get_node_by_hashcode(self, cache_id, hashcode, parts):
"""
Get node by key hashcode. Calculate partition and return node on that it is primary.
(algorithm is taken from `RendezvousAffinityFunction.java`)
"""
# calculate partition for key or affinity key
# (algorithm is taken from `RendezvousAffinityFunction.java`)
mask = parts - 1
if parts & mask == 0:
part = (hashcode ^ (unsigned(hashcode) >> 16)) & mask
else:
part = abs(hashcode // parts)
assert 0 <= part < parts, 'Partition calculation has failed'
node_mapping = self._cache_partition_mapping(cache_id).get('node_mapping')
if not node_mapping:
return None
node_uuid, best_conn = None, None
for u, p in node_mapping.items():
if part in p:
node_uuid = u
break
if node_uuid:
for n in self._nodes:
if n.uuid == node_uuid:
best_conn = n
break
if best_conn and best_conn.alive:
return best_conn
class _ConnectionContextManager:
def __init__(self, client, nodes):
self.client = client
self.nodes = nodes
self.client._connect(self.nodes)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.client.close()
class Client(BaseClient):
"""
Synchronous Client implementation.
"""
def __init__(self, compact_footer: bool = None, partition_aware: bool = True,
event_listeners: Optional[Sequence] = None, **kwargs):
"""
Initialize client.
For the use of the SSL-related parameters see
https://docs.python.org/3/library/ssl.html#ssl-certificates.
:param compact_footer: (optional) use compact (True, recommended) or
full (False) schema approach when serializing Complex objects.
Default is to use the same approach the server is using (None).
Apache Ignite binary protocol documentation on this topic:
https://ignite.apache.org/docs/latest/binary-client-protocol/data-format#schema
:param partition_aware: (optional) try to calculate the exact data
placement from the key before to issue the key operation to the
server node, `True` by default,
:param event_listeners: (optional) event listeners,
:param timeout: (optional) sets timeout (in seconds) for each socket
operation including `connect`. 0 means non-blocking mode, which is
virtually guaranteed to fail. Can accept integer or float value.
Default is None (blocking mode),
:param handshake_timeout: (optional) sets timeout (in seconds) for performing handshake (connection)
with node. Default is 10.0 seconds,
:param use_ssl: (optional) set to True if Ignite server uses SSL
on its binary connector. Defaults to use SSL when username
and password has been supplied, not to use SSL otherwise,
:param ssl_version: (optional) SSL version constant from standard
`ssl` module. Defaults to TLS v1.2,
:param ssl_ciphers: (optional) ciphers to use. If not provided,
`ssl` default ciphers are used,
:param ssl_cert_reqs: (optional) determines how the remote side
certificate is treated:
* `ssl.CERT_NONE` − remote certificate is ignored (default),
* `ssl.CERT_OPTIONAL` − remote certificate will be validated,
if provided,
* `ssl.CERT_REQUIRED` − valid remote certificate is required,
:param ssl_keyfile: (optional) a path to SSL key file to identify
local (client) party,
:param ssl_keyfile_password: (optional) password for SSL key file,
can be provided when key file is encrypted to prevent OpenSSL
password prompt,
:param ssl_certfile: (optional) a path to ssl certificate file
to identify local (client) party,
:param ssl_ca_certfile: (optional) a path to a trusted certificate
or a certificate chain. Required to check the validity of the remote
(server-side) certificate,
:param username: (optional) user name to authenticate to Ignite
cluster,
:param password: (optional) password to authenticate to Ignite cluster.
"""
super().__init__(compact_footer, partition_aware, event_listeners, **kwargs)
def connect(self, *args):
"""
Connect to Ignite cluster node(s).
:param args: (optional) host(s) and port(s) to connect to.
"""
nodes = self._process_connect_args(*args)
return _ConnectionContextManager(self, nodes)
def _connect(self, nodes):
# the following code is quite twisted, because the protocol version
# is initially unknown
# TODO: open first node in foreground, others − in background
for i, node in enumerate(nodes):
host, port = node
conn = Connection(self, host, port, **self._connection_args)
try:
if self.protocol_context is None or self.partition_aware:
# open connection before adding to the pool
conn.connect()
# now we have the protocol version
if not self.partition_aware:
# do not try to open more nodes
self._current_node = i
except connection_errors:
if self.partition_aware:
# schedule the reconnection
conn.reconnect()
self._nodes.append(conn)
if self.protocol_context is None:
raise ReconnectError('Can not connect.')
def close(self):
for conn in self._nodes:
conn.close()
self._nodes.clear()
@property
def random_node(self) -> Connection:
"""
Returns random usable node.
This method is not a part of the public API. Unless you wish to
extend the `pyignite` capabilities (with additional testing, logging,
examining connections, et c.) you probably should not use it.
"""
if self.partition_aware:
# if partition awareness is used just pick a random connected node
return self._get_random_node()
else:
# if partition awareness is not used then just return the current
# node if it's alive or the next usable node if connection with the
# current is broken
node = self._nodes[self._current_node]
if node.alive:
return node
# close current (supposedly failed) node
self._nodes[self._current_node].close()
# advance the node index
self._current_node += 1
if self._current_node >= len(self._nodes):
self._current_node = 0
# prepare the list of node indexes to try to connect to
num_nodes = len(self._nodes)
for i in chain(range(self._current_node, num_nodes), range(self._current_node)):
node = self._nodes[i]
try:
node.connect()
except connection_errors:
pass
else:
return node
# no nodes left
raise ReconnectError('Can not reconnect: out of nodes.')
def _get_random_node(self, reconnect=True):
alive_nodes = [n for n in self._nodes if n.alive]
if alive_nodes:
return random.choice(alive_nodes)
elif reconnect:
for n in self._nodes:
n.reconnect()
return self._get_random_node(reconnect=False)
else:
# cannot choose from an empty sequence
raise ReconnectError('Can not reconnect: out of nodes.') from None
@status_to_exception(BinaryTypeError)
def get_binary_type(self, binary_type: Union[str, int]) -> dict:
"""
Gets the binary type information from the Ignite server. This is quite
a low-level implementation of Ignite thin client protocol's
`OP_GET_BINARY_TYPE` operation. You would probably want to use
:py:meth:`~pyignite.client.Client.query_binary_type` instead.
:param binary_type: binary type name or ID,
:return: binary type description − a dict with the following fields:
- `type_exists`: True if the type is registered, False otherwise. In
the latter case all the following fields are omitted,
- `type_id`: Complex object type ID,
- `type_name`: Complex object type name,
- `affinity_key_field`: string value or None,
- `is_enum`: False in case of Complex object registration,
- `schemas`: a list, containing the Complex object schemas in format:
OrderedDict[field name: field type hint]. A schema can be empty.
"""
result = get_binary_type(self.random_node, binary_type)
return self._process_get_binary_type_result(result)
@status_to_exception(BinaryTypeError)
def put_binary_type(
self, type_name: str, affinity_key_field: str = None,
is_enum=False, schema: dict = None
):
"""
Registers binary type information in cluster. Do not update binary
registry. This is a literal implementation of Ignite thin client
protocol's `OP_PUT_BINARY_TYPE` operation. You would probably want
to use :py:meth:`~pyignite.client.Client.register_binary_type` instead.
:param type_name: name of the data type being registered,
:param affinity_key_field: (optional) name of the affinity key field,
:param is_enum: (optional) register enum if True, binary object
otherwise. Defaults to False,
:param schema: (optional) when register enum, pass a dict
of enumerated parameter names as keys and an integers as values.
When register binary type, pass a dict of field names: field types.
Binary type with no fields is OK.
"""
return put_binary_type(self.random_node, type_name, affinity_key_field, is_enum, schema)
def register_binary_type(self, data_class: Type, affinity_key_field: str = None):
"""
Register the given class as a representation of a certain Complex
object type. Discards autogenerated or previously registered class.
:param data_class: Complex object class,
:param affinity_key_field: (optional) affinity parameter.
"""
if not self.query_binary_type(data_class.type_id, data_class.schema_id):
self.put_binary_type(data_class.type_name, affinity_key_field, schema=data_class.schema)
self._registry[data_class.type_id][data_class.schema_id] = data_class
def query_binary_type(self, binary_type: Union[int, str], schema: Union[int, dict] = None):
"""
Queries the registry of Complex object classes.
:param binary_type: Complex object type name or ID,
:param schema: (optional) Complex object schema or schema ID
:return: found dataclass or None, if `schema` parameter is provided,
a dict of {schema ID: dataclass} format otherwise.
"""
type_id = entity_id(binary_type)
result = self._get_from_registry(type_id, schema)
if not result:
type_info = self.get_binary_type(type_id)
self._sync_binary_registry(type_id, type_info)
return self._get_from_registry(type_id, schema)
return result
def unwrap_binary(self, value: Any) -> Any:
"""
Detects and recursively unwraps Binary Object or collections of BinaryObject.
:param value: anything that could be a Binary Object or collection of BinaryObject,
:return: the result of the Binary Object unwrapping with all other data
left intact.
"""
if isinstance(value, tuple) and len(value) == 2:
if type(value[0]) is bytes and type(value[1]) is int:
blob, offset = value
with BinaryStream(self, blob) as stream:
data_class = BinaryObject.parse(stream)
return BinaryObject.to_python(stream.read_ctype(data_class, direction=READ_BACKWARD), client=self)
if isinstance(value[0], int):
col_type, collection = value
if isinstance(collection, list):
return col_type, [self.unwrap_binary(v) for v in collection]
if isinstance(collection, dict):
return col_type, {self.unwrap_binary(k): self.unwrap_binary(v) for k, v in collection.items()}
return value
@status_to_exception(CacheError)
def _get_affinity(self, conn: 'Connection', caches: Iterable[int]) -> Dict:
"""
Queries server for affinity mappings. Retries in case
of an intermittent error (most probably “Getting affinity for topology
version earlier than affinity is calculated”).
:param conn: connection to Ignite server,
:param caches: Ids of caches,
:return: OP_CACHE_PARTITIONS operation result value.
"""
for _ in range(AFFINITY_RETRIES or 1):
result = cache_get_node_partitions(conn, caches)
if result.status == 0:
break
time.sleep(AFFINITY_DELAY)
return result
def get_best_node(
self, cache: Union[int, str, 'BaseCache'], key: Any = None, key_hint: 'IgniteDataType' = None
) -> 'Connection':
"""
Returns the node from the list of the nodes, opened by client, that
most probably contains the needed key-value pair. See IEP-23.
This method is not a part of the public API. Unless you wish to
extend the `pyignite` capabilities (with additional testing, logging,
examining connections, et c.) you probably should not use it.
:param cache: Ignite cache, cache name or cache id,
:param key: (optional) pythonic key,
:param key_hint: (optional) Ignite data type, for which the given key
should be converted,
:return: Ignite connection object.
"""
conn = self.random_node
if self.partition_aware and key is not None:
caches = self._caches_to_update_affinity()
if caches:
# update partition mapping
while True:
try:
full_affinity = self._get_affinity(conn, caches)
break
except connection_errors:
# retry if connection failed
conn = self.random_node
pass
except CacheError:
# server did not create mapping in time
return conn
self._update_affinity(full_affinity)
for node in self._nodes:
if not node.alive:
node.reconnect()
c_id = cache.cache_id if isinstance(cache, BaseCache) else cache_id(cache)
parts = self._cache_partition_mapping(c_id).get('number_of_partitions')
if not parts:
return conn
key, key_hint = self._get_affinity_key(c_id, key, key_hint)
hashcode = key_hint.hashcode(key, client=self)
best_node = self._get_node_by_hashcode(c_id, hashcode, parts)
if best_node:
return best_node
return conn
def create_cache(self, settings: Union[str, dict]) -> 'Cache':
"""
Creates Ignite cache by name. Raises `CacheError` if such a cache is
already exists.
:param settings: cache name or dict of cache properties' codes
and values. All cache properties are documented here:
:ref:`cache_props`. See also the
:ref:`cache creation example <sql_cache_create>`,
:return: :class:`~pyignite.cache.Cache` object.
"""
return create_cache(self, settings)
def get_or_create_cache(self, settings: Union[str, dict]) -> 'Cache':
"""
Creates Ignite cache, if not exist.
:param settings: cache name or dict of cache properties' codes
and values. All cache properties are documented here:
:ref:`cache_props`. See also the
:ref:`cache creation example <sql_cache_create>`,
:return: :class:`~pyignite.cache.Cache` object.
"""
return get_or_create_cache(self, settings)
def get_cache(self, settings: Union[str, dict]) -> 'Cache':
"""
Creates Cache object with a given cache name without checking it up
on server. If such a cache does not exist, some kind of exception
(most probably `CacheError`) may be raised later.
:param settings: cache name or cache properties (but only `PROP_NAME`
property is allowed),
:return: :class:`~pyignite.cache.Cache` object.
"""
return get_cache(self, settings)
@status_to_exception(CacheError)
def get_cache_names(self) -> list:
"""
Gets existing cache names.
:return: list of cache names.
"""
return cache_get_names(self.random_node)
def sql(
self, query_str: str, page_size: int = 1024,
query_args: Iterable = None, schema: str = 'PUBLIC',
statement_type: int = 0, distributed_joins: bool = False,
local: bool = False, replicated_only: bool = False,
enforce_join_order: bool = False, collocated: bool = False,
lazy: bool = False, include_field_names: bool = False,
max_rows: int = -1, timeout: int = 0,
cache: Union[int, str, Cache] = None
) -> SqlFieldsCursor:
"""
Runs an SQL query and returns its result.
:param query_str: SQL query string,
:param page_size: (optional) cursor page size. Default is 1024, which
means that client makes one server call per 1024 rows,
:param query_args: (optional) query arguments. List of values or
(value, type hint) tuples,
:param schema: (optional) schema for the query. Defaults to `PUBLIC`,
:param statement_type: (optional) statement type. Can be:
* StatementType.ALL − any type (default),
* StatementType.SELECT − select,
* StatementType.UPDATE − update.
:param distributed_joins: (optional) distributed joins. Defaults
to False,
:param local: (optional) pass True if this query should be executed
on local node only. Defaults to False,
:param replicated_only: (optional) whether query contains only
replicated tables or not. Defaults to False,
:param enforce_join_order: (optional) enforce join order. Defaults
to False,
:param collocated: (optional) whether your data is co-located or not.
Defaults to False,
:param lazy: (optional) lazy query execution. Defaults to False,
:param include_field_names: (optional) include field names in result.
Defaults to False,
:param max_rows: (optional) query-wide maximum of rows. Defaults to -1
(all rows),
:param timeout: (optional) non-negative timeout value in ms.
Zero disables timeout (default),
:param cache: (optional) Name or ID of the cache to use to infer schema.
If set, 'schema' argument is ignored,
:return: sql fields cursor with result rows as a lists. If
`include_field_names` was set, the first row will hold field names.
"""
if isinstance(cache, (int, str)):
c_info = CacheInfo(cache_id=cache_id(cache), protocol_context=self.protocol_context)
elif isinstance(cache, Cache):
c_info = cache.cache_info
else:
c_info = CacheInfo(protocol_context=self.protocol_context)
if c_info.cache_id:
schema = None
return SqlFieldsCursor(self, c_info, query_str, page_size, query_args, schema, statement_type,
distributed_joins, local, replicated_only, enforce_join_order, collocated, lazy,
include_field_names, max_rows, timeout)
def get_cluster(self) -> 'Cluster':
"""
Get client cluster facade.
:return: :py:class:`~pyignite.cluster.Cluster` instance.
"""
return Cluster(self)
def tx_start(self, concurrency: TransactionConcurrency = TransactionConcurrency.PESSIMISTIC,
isolation: TransactionIsolation = TransactionIsolation.REPEATABLE_READ,
timeout: int = 0, label: Optional[str] = None) -> 'Transaction':
"""
Start thin client transaction.
:param concurrency: (optional) transaction concurrency, see
:py:class:`~pyignite.datatypes.transactions.TransactionConcurrency`,
:param isolation: (optional) transaction isolation level, see
:py:class:`~pyignite.datatypes.transactions.TransactionIsolation`,
:param timeout: (optional) transaction timeout in milliseconds,
:param label: (optional) transaction label.
:return: :py:class:`~pyignite.transaction.Transaction` instance.
"""
return Transaction(self, concurrency, isolation, timeout, label)