pyignite/api/key_value.py

# 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.

from typing import Any, Iterable, Optional, Union

from pyignite.connection import AioConnection, Connection
from pyignite.queries.op_codes import (
    OP_CACHE_PUT, OP_CACHE_GET, OP_CACHE_GET_ALL, OP_CACHE_PUT_ALL, OP_CACHE_CONTAINS_KEY, OP_CACHE_CONTAINS_KEYS,
    OP_CACHE_GET_AND_PUT, OP_CACHE_GET_AND_REPLACE, OP_CACHE_GET_AND_REMOVE, OP_CACHE_PUT_IF_ABSENT,
    OP_CACHE_GET_AND_PUT_IF_ABSENT, OP_CACHE_REPLACE, OP_CACHE_REPLACE_IF_EQUALS, OP_CACHE_CLEAR, OP_CACHE_CLEAR_KEY,
    OP_CACHE_CLEAR_KEYS, OP_CACHE_REMOVE_KEY, OP_CACHE_REMOVE_IF_EQUALS, OP_CACHE_REMOVE_KEYS, OP_CACHE_REMOVE_ALL,
    OP_CACHE_GET_SIZE, OP_CACHE_LOCAL_PEEK
)
from pyignite.datatypes import Map, Bool, Byte, Int, Long, AnyDataArray, AnyDataObject, ByteArray
from pyignite.datatypes.base import IgniteDataType
from pyignite.queries import Query, query_perform
from pyignite.utils import cache_id

from .result import APIResult


def cache_put(connection: 'Connection', cache: Union[str, int], key: Any, value: Any,
              key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None, binary: bool = False,
              query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts a value with a given key to cache (overwriting existing value if any).

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param value: value for the key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param value_hint: (optional) Ignite data type, for which the given value
     should be converted.
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status if a value
     is written, non-zero status and an error description otherwise.
    """
    return __cache_put(connection, cache, key, value, key_hint, value_hint, binary, query_id)


async def cache_put_async(connection: 'AioConnection', cache: Union[str, int], key: Any, value: Any,
                          key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None, binary: bool = False,
                          query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_put
    """
    return await __cache_put(connection, cache, key, value, key_hint, value_hint, binary, query_id)


def __cache_put(connection, cache, key, value, key_hint, value_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_PUT,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('value', value_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'value': value
        }
    )


def cache_get(connection: 'Connection', cache: Union[str, int], key: Any, key_hint: 'IgniteDataType' = None,
              binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Retrieves a value from cache by key.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and a value
     retrieved on success, non-zero status and an error description on failure.
    """
    return __cache_get(connection, cache, key, key_hint, binary, query_id)


async def cache_get_async(connection: 'AioConnection', cache: Union[str, int], key: Any,
                          key_hint: 'IgniteDataType' = None, binary: bool = False,
                          query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_get
    """
    return await __cache_get(connection, cache, key, key_hint, binary, query_id)


def __cache_get(connection, cache, key, key_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_GET,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
        },
        response_config=[
            ('value', AnyDataObject),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def cache_get_all(connection: 'Connection', cache: Union[str, int], keys: Iterable, binary: bool = False,
                  query_id: Optional[int] = None) -> 'APIResult':
    """
    Retrieves multiple key-value pairs from cache.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param keys: list of keys or tuples of (key, key_hint),
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and a dict, made of
     retrieved key-value pairs, non-zero status and an error description
     on failure.
    """
    return __cache_get_all(connection, cache, keys, binary, query_id)


async def cache_get_all_async(connection: 'AioConnection', cache: Union[str, int], keys: Iterable, binary: bool = False,
                              query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_get_all.
    """
    return await __cache_get_all(connection, cache, keys, binary, query_id)


def __cache_get_all(connection, cache, keys, binary, query_id):
    query_struct = Query(
        OP_CACHE_GET_ALL,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('keys', AnyDataArray()),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'keys': keys,
        },
        response_config=[
            ('data', Map),
        ],
        post_process_fun=__post_process_value_by_key('data')
    )


def cache_put_all(connection: 'Connection', cache: Union[str, int], pairs: dict, binary: bool = False,
                  query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts multiple key-value pairs to cache (overwriting existing associations
    if any).

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param pairs: dictionary type parameters, contains key-value pairs to save.
     Each key or value can be an item of representable Python type or a tuple
     of (item, hint),
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status if key-value pairs
     are written, non-zero status and an error description otherwise.
    """
    return __cache_put_all(connection, cache, pairs, binary, query_id)


async def cache_put_all_async(connection: 'AioConnection', cache: Union[str, int], pairs: dict, binary: bool = False,
                              query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_put_all.
    """
    return await __cache_put_all(connection, cache, pairs, binary, query_id)


def __cache_put_all(connection, cache, pairs, binary, query_id):
    query_struct = Query(
        OP_CACHE_PUT_ALL,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('data', Map),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'data': pairs,
        },
    )


def cache_contains_key(connection: 'Connection', cache: Union[str, int], key: Any, key_hint: 'IgniteDataType' = None,
                       binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Returns a value indicating whether given key is present in cache.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param binary: pass True to keep the value in binary form. False
     by default,
    :param query_id: a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and a bool value
     retrieved on success: `True` when key is present, `False` otherwise,
     non-zero status and an error description on failure.
    """
    return __cache_contains_key(connection, cache, key, key_hint, binary, query_id)


async def cache_contains_key_async(connection: 'AioConnection', cache: Union[str, int], key: Any,
                                   key_hint: 'IgniteDataType' = None, binary: bool = False,
                                   query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_contains_key.
    """
    return await __cache_contains_key(connection, cache, key, key_hint, binary, query_id)


def __cache_contains_key(connection, cache, key, key_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_CONTAINS_KEY,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
        },
        response_config=[
            ('value', Bool),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def cache_contains_keys(connection: 'Connection', cache: Union[str, int], keys: Iterable, binary: bool = False,
                        query_id: Optional[int] = None) -> 'APIResult':
    """
    Returns a value indicating whether all given keys are present in cache.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param keys: a list of keys or (key, type hint) tuples,
    :param binary: pass True to keep the value in binary form. False
     by default,
    :param query_id: a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and a bool value
     retrieved on success: `True` when all keys are present, `False` otherwise,
     non-zero status and an error description on failure.
    """
    return __cache_contains_keys(connection, cache, keys, binary, query_id)


async def cache_contains_keys_async(connection: 'AioConnection', cache: Union[str, int], keys: Iterable,
                                    binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_contains_keys.
    """
    return await __cache_contains_keys(connection, cache, keys, binary, query_id)


def __cache_contains_keys(connection, cache, keys, binary, query_id):
    query_struct = Query(
        OP_CACHE_CONTAINS_KEYS,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('keys', AnyDataArray()),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'keys': keys,
        },
        response_config=[
            ('value', Bool),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def cache_get_and_put(connection: 'Connection', cache: Union[str, int], key: Any, value: Any,
                      key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None, binary: bool = False,
                      query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts a value with a given key to cache, and returns the previous value
    for that key, or null value if there was not such key.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param value: value for the key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param value_hint: (optional) Ignite data type, for which the given value
     should be converted.
    :param binary: pass True to keep the value in binary form. False
     by default,
    :param query_id: a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and an old value
     or None if a value is written, non-zero status and an error description
     in case of error.
    """
    return __cache_get_and_put(connection, cache, key, value, key_hint, value_hint, binary, query_id)


async def cache_get_and_put_async(connection: 'AioConnection', cache: Union[str, int], key: Any, value: Any,
                                  key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
                                  binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_get_and_put.
    """
    return await __cache_get_and_put(connection, cache, key, value, key_hint, value_hint, binary, query_id)


def __cache_get_and_put(connection, cache, key, value, key_hint, value_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_GET_AND_PUT,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('value', value_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'value': value,
        },
        response_config=[
            ('value', AnyDataObject),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def cache_get_and_replace(connection: 'Connection', cache: Union[str, int], key: Any, value: Any,
                          key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None, binary: bool = False,
                          query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts a value with a given key to cache, returning previous value
    for that key, if and only if there is a value currently mapped
    for that key.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param value: value for the key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param value_hint: (optional) Ignite data type, for which the given value
     should be converted.
    :param binary: pass True to keep the value in binary form. False
     by default,
    :param query_id: a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and an old value
     or None on success, non-zero status and an error description otherwise.
    """
    return __cache_get_and_replace(connection, cache, key, key_hint, value, value_hint, binary, query_id)


async def cache_get_and_replace_async(connection: 'AioConnection', cache: Union[str, int], key: Any, value: Any,
                                      key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
                                      binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_get_and_replace.
    """
    return await __cache_get_and_replace(connection, cache, key, key_hint, value, value_hint, binary, query_id)


def __cache_get_and_replace(connection, cache, key, key_hint, value, value_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_GET_AND_REPLACE, [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('value', value_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'value': value,
        },
        response_config=[
            ('value', AnyDataObject),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def cache_get_and_remove(connection: 'Connection', cache: Union[str, int], key: Any, key_hint: 'IgniteDataType' = None,
                         binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Removes the cache entry with specified key, returning the value.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param binary: pass True to keep the value in binary form. False
     by default,
    :param query_id: a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and an old value
     or None, non-zero status and an error description otherwise.
    """
    return __cache_get_and_remove(connection, cache, key, key_hint, binary, query_id)


async def cache_get_and_remove_async(connection: 'AioConnection', cache: Union[str, int], key: Any,
                                     key_hint: 'IgniteDataType' = None, binary: bool = False,
                                     query_id: Optional[int] = None) -> 'APIResult':
    return await __cache_get_and_remove(connection, cache, key, key_hint, binary, query_id)


def __cache_get_and_remove(connection, cache, key, key_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_GET_AND_REMOVE, [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
        },
        response_config=[
            ('value', AnyDataObject),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def cache_put_if_absent(connection: 'Connection', cache: Union[str, int], key: Any, value: Any,
                        key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
                        binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts a value with a given key to cache only if the key
    does not already exist.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param value: value for the key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param value_hint: (optional) Ignite data type, for which the given value
     should be converted.
    :param binary: (optional) pass True to keep the value in binary form. False
     by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status on success,
     non-zero status and an error description otherwise.
    """
    return __cache_put_if_absent(connection, cache, key, value, key_hint, value_hint, binary, query_id)


async def cache_put_if_absent_async(connection: 'AioConnection', cache: Union[str, int], key: Any, value: Any,
                                    key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
                                    binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_put_if_absent.
    """
    return await __cache_put_if_absent(connection, cache, key, value, key_hint, value_hint, binary, query_id)


def __cache_put_if_absent(connection, cache, key, value, key_hint, value_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_PUT_IF_ABSENT,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('value', value_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'value': value,
        },
        response_config=[
            ('success', Bool),
        ],
        post_process_fun=__post_process_value_by_key('success')
    )


def cache_get_and_put_if_absent(connection: 'Connection', cache: Union[str, int], key: Any, value: Any,
                                key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
                                binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts a value with a given key to cache only if the key does not
    already exist.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param value: value for the key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param value_hint: (optional) Ignite data type, for which the given value
     should be converted.
    :param binary: (optional) pass True to keep the value in binary form. False
     by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and an old value
     or None on success, non-zero status and an error description otherwise.
    """
    return __cache_get_and_put_if_absent(connection, cache, key, value, key_hint, value_hint, binary, query_id)


async def cache_get_and_put_if_absent_async(connection: 'AioConnection', cache: Union[str, int], key: Any, value: Any,
                                            key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
                                            binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_get_and_put_if_absent.
    """
    return await __cache_get_and_put_if_absent(connection, cache, key, value, key_hint, value_hint, binary, query_id)


def __cache_get_and_put_if_absent(connection, cache, key, value, key_hint, value_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_GET_AND_PUT_IF_ABSENT,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('value', value_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'value': value,
        },
        response_config=[
            ('value', AnyDataObject),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def cache_replace(connection: 'Connection', cache: Union[str, int], key: Any, value: Any,
                  key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None, binary: bool = False,
                  query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts a value with a given key to cache only if the key already exist.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: key for the cache entry. Can be of any supported type,
    :param value: value for the key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param value_hint: (optional) Ignite data type, for which the given value
     should be converted.
    :param binary: pass True to keep the value in binary form. False
     by default,
    :param query_id: a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and a boolean
     success code, or non-zero status and an error description if something
     has gone wrong.
    """
    return __cache_replace(connection, cache, key, value, key_hint, value_hint, binary, query_id)


async def cache_replace_async(connection: 'AioConnection', cache: Union[str, int], key: Any, value: Any,
                              key_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
                              binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_replace.
    """
    return await __cache_replace(connection, cache, key, value, key_hint, value_hint, binary, query_id)


def __cache_replace(connection, cache, key, value, key_hint, value_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_REPLACE,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('value', value_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'value': value,
        },
        response_config=[
            ('success', Bool),
        ],
        post_process_fun=__post_process_value_by_key('success')
    )


def cache_replace_if_equals(connection: 'Connection', cache: Union[str, int], key: Any, sample: Any, value: Any,
                            key_hint: 'IgniteDataType' = None, sample_hint: 'IgniteDataType' = None,
                            value_hint: 'IgniteDataType' = None, binary: bool = False,
                            query_id: Optional[int] = None) -> 'APIResult':
    """
    Puts a value with a given key to cache only if the key already exists
    and value equals provided sample.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key:  key for the cache entry,
    :param sample: a sample to compare the stored value with,
    :param value: new value for the given key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param sample_hint: (optional) Ignite data type, for whic
     the given sample should be converted
    :param value_hint: (optional) Ignite data type, for which the given value
     should be converted,
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned
     as-is in response.query_id. When the parameter is omitted, a random
     value is generated,
    :return: API result data object. Contains zero status and a boolean
     success code, or non-zero status and an error description if something
     has gone wrong.
    """
    return __cache_replace_if_equals(connection, cache, key, sample, value, key_hint, sample_hint, value_hint, binary,
                                     query_id)


async def cache_replace_if_equals_async(
        connection: 'AioConnection', cache: Union[str, int], key: Any, sample: Any, value: Any,
        key_hint: 'IgniteDataType' = None, sample_hint: 'IgniteDataType' = None, value_hint: 'IgniteDataType' = None,
        binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_replace_if_equals.
    """
    return await __cache_replace_if_equals(connection, cache, key, sample, value, key_hint, sample_hint, value_hint,
                                           binary, query_id)


def __cache_replace_if_equals(connection, cache, key, sample, value, key_hint, sample_hint, value_hint, binary,
                              query_id):
    query_struct = Query(
        OP_CACHE_REPLACE_IF_EQUALS,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('sample', sample_hint or AnyDataObject),
            ('value', value_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'sample': sample,
            'value': value,
        },
        response_config=[
            ('success', Bool),
        ],
        post_process_fun=__post_process_value_by_key('success')
    )


def cache_clear(connection: 'Connection', cache: Union[str, int], binary: bool = False,
                query_id: Optional[int] = None) -> 'APIResult':
    """
    Clears the cache without notifying listeners or cache writers.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned
     as-is in response.query_id. When the parameter is omitted, a random
     value is generated,
    :return: API result data object. Contains zero status on success,
     non-zero status and an error description otherwise.
    """
    return __cache_clear(connection, cache, binary, query_id)


async def cache_clear_async(connection: 'AioConnection', cache: Union[str, int], binary: bool = False,
                            query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_clear.
    """
    return await __cache_clear(connection, cache, binary, query_id)


def __cache_clear(connection, cache, binary, query_id):
    query_struct = Query(
        OP_CACHE_CLEAR,
        [
            ('hash_code', Int),
            ('flag', Byte),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
        },
    )


def cache_clear_key(connection: 'Connection', cache: Union[str, int], key: Any, key_hint: 'IgniteDataType' = None,
                    binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Clears the cache key without notifying listeners or cache writers.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key:  key for the cache entry,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned
     as-is in response.query_id. When the parameter is omitted, a random
     value is generated,
    :return: API result data object. Contains zero status on success,
     non-zero status and an error description otherwise.
    """
    return __cache_clear_key(connection, cache, key, key_hint, binary, query_id)


async def cache_clear_key_async(connection: 'AioConnection', cache: Union[str, int], key: Any,
                                key_hint: 'IgniteDataType' = None, binary: bool = False,
                                query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_clear_key.
    """
    return await __cache_clear_key(connection, cache, key, key_hint, binary, query_id)


def __cache_clear_key(connection, cache, key, key_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_CLEAR_KEY,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
        },
    )


def cache_clear_keys(connection: 'Connection', cache: Union[str, int], keys: Iterable, binary: bool = False,
                     query_id: Optional[int] = None) -> 'APIResult':
    """
    Clears the cache keys without notifying listeners or cache writers.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param keys: list of keys or tuples of (key, key_hint),
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status on success,
     non-zero status and an error description otherwise.
    """
    return __cache_clear_keys(connection, cache, keys, binary, query_id)


async def cache_clear_keys_async(connection: 'AioConnection', cache: Union[str, int], keys: Iterable,
                                 binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_clear_keys.
    """
    return await __cache_clear_keys(connection, cache, keys, binary, query_id)


def __cache_clear_keys(connection, cache, keys, binary, query_id):
    query_struct = Query(
        OP_CACHE_CLEAR_KEYS,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('keys', AnyDataArray()),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'keys': keys,
        },
    )


def cache_remove_key(connection: 'Connection', cache: Union[str, int], key: Any, key_hint: 'IgniteDataType' = None,
                     binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Clears the cache key without notifying listeners or cache writers.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key:  key for the cache entry,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned
     as-is in response.query_id. When the parameter is omitted, a random
     value is generated,
    :return: API result data object. Contains zero status and a boolean
     success code, or non-zero status and an error description if something
     has gone wrong.
    """
    return __cache_remove_key(connection, cache, key, key_hint, binary, query_id)


async def cache_remove_key_async(connection: 'AioConnection', cache: Union[str, int], key: Any,
                                 key_hint: 'IgniteDataType' = None, binary: bool = False,
                                 query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_remove_key.
    """
    return await __cache_remove_key(connection, cache, key, key_hint, binary, query_id)


def __cache_remove_key(connection, cache, key, key_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_REMOVE_KEY,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
        },
        response_config=[
            ('success', Bool),
        ],
        post_process_fun=__post_process_value_by_key('success')
    )


def cache_remove_if_equals(connection: 'Connection', cache: Union[str, int], key: Any, sample: Any,
                           key_hint: 'IgniteDataType' = None, sample_hint: 'IgniteDataType' = None,
                           binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Removes an entry with a given key if provided value is equal to
    actual value, notifying listeners and cache writers.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key:  key for the cache entry,
    :param sample: a sample to compare the stored value with,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param sample_hint: (optional) Ignite data type, for whic
     the given sample should be converted
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned
     as-is in response.query_id. When the parameter is omitted, a random
     value is generated,
    :return: API result data object. Contains zero status and a boolean
     success code, or non-zero status and an error description if something
     has gone wrong.
    """
    return __cache_remove_if_equals(connection, cache, key, sample, key_hint, sample_hint, binary, query_id)


async def cache_remove_if_equals_async(
        connection: 'AioConnection', cache: Union[str, int], key: Any, sample: Any, key_hint: 'IgniteDataType' = None,
        sample_hint: 'IgniteDataType' = None, binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_remove_if_equals.
    """
    return await __cache_remove_if_equals(connection, cache, key, sample, key_hint, sample_hint, binary, query_id)


def __cache_remove_if_equals(connection, cache, key, sample, key_hint, sample_hint, binary, query_id):
    query_struct = Query(
        OP_CACHE_REMOVE_IF_EQUALS,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('sample', sample_hint or AnyDataObject),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'sample': sample,
        },
        response_config=[
            ('success', Bool),
        ],
        post_process_fun=__post_process_value_by_key('success')
    )


def cache_remove_keys(connection: 'Connection', cache: Union[str, int], keys: Iterable, binary: bool = False,
                      query_id: Optional[int] = None) -> 'APIResult':
    """
    Removes entries with given keys, notifying listeners and cache writers.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param keys: list of keys or tuples of (key, key_hint),
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status on success,
     non-zero status and an error description otherwise.
    """
    return __cache_remove_keys(connection, cache, keys, binary, query_id)


async def cache_remove_keys_async(connection: 'AioConnection', cache: Union[str, int], keys: Iterable,
                                  binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_remove_keys.
    """
    return await __cache_remove_keys(connection, cache, keys, binary, query_id)


def __cache_remove_keys(connection, cache, keys, binary, query_id):
    query_struct = Query(
        OP_CACHE_REMOVE_KEYS,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('keys', AnyDataArray()),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'keys': keys,
        },
    )


def cache_remove_all(connection: 'Connection', cache: Union[str, int], binary: bool = False,
                     query_id: Optional[int] = None) -> 'APIResult':
    """
    Removes all entries from cache, notifying listeners and cache writers.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status on success,
     non-zero status and an error description otherwise.
    """
    return __cache_remove_all(connection, cache, binary, query_id)


async def cache_remove_all_async(connection: 'AioConnection', cache: Union[str, int], binary: bool = False,
                                 query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_remove_all.
    """
    return await __cache_remove_all(connection, cache, binary, query_id)


def __cache_remove_all(connection, cache, binary, query_id):
    query_struct = Query(
        OP_CACHE_REMOVE_ALL,
        [
            ('hash_code', Int),
            ('flag', Byte),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
        },
    )


def cache_get_size(connection: 'Connection', cache: Union[str, int], peek_modes: Union[int, list, tuple] = None,
                   binary: bool = False, query_id: Optional[int] = None) -> 'APIResult':
    """
    Gets the number of entries in cache.

    :param connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param peek_modes: (optional) limit count to near cache partition
     (PeekModes.NEAR), primary cache (PeekModes.PRIMARY), or backup cache
     (PeekModes.BACKUP). Defaults to pimary cache partitions (PeekModes.PRIMARY),
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and a number of
     cache entries on success, non-zero status and an error description
     otherwise.
    """
    return __cache_get_size(connection, cache, peek_modes, binary, query_id)


async def cache_get_size_async(connection: 'AioConnection', cache: Union[str, int],
                               peek_modes: Union[int, list, tuple] = None, binary: bool = False,
                               query_id: Optional[int] = None) -> 'APIResult':
    return await __cache_get_size(connection, cache, peek_modes, binary, query_id)


def __cache_get_size(connection, cache, peek_modes, binary, query_id):
    if peek_modes is None:
        peek_modes = []
    elif not isinstance(peek_modes, (list, tuple)):
        peek_modes = [peek_modes]

    query_struct = Query(
        OP_CACHE_GET_SIZE,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('peek_modes', ByteArray),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, connection,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'peek_modes': peek_modes,
        },
        response_config=[
            ('count', Long),
        ],
        post_process_fun=__post_process_value_by_key('count')
    )


def cache_local_peek(conn: 'Connection', cache: Union[str, int], key: Any, key_hint: 'IgniteDataType' = None,
                     peek_modes: Union[int, list, tuple] = None, binary: bool = False,
                     query_id: Optional[int] = None) -> 'APIResult':
    """
    Peeks at in-memory cached value using default optional peek mode.

    This method will not load value from any cache store or from a remote
    node.

    :param conn: connection: connection to Ignite server,
    :param cache: name or ID of the cache,
    :param key: entry key,
    :param key_hint: (optional) Ignite data type, for which the given key
     should be converted,
    :param peek_modes: (optional) limit count to near cache partition
     (PeekModes.NEAR), primary cache (PeekModes.PRIMARY), or backup cache
     (PeekModes.BACKUP). Defaults to primary cache partitions (PeekModes.PRIMARY),
    :param binary: (optional) pass True to keep the value in binary form.
     False by default,
    :param query_id: (optional) a value generated by client and returned as-is
     in response.query_id. When the parameter is omitted, a random value
     is generated,
    :return: API result data object. Contains zero status and a peeked value
     (null if not found).
    """
    return __cache_local_peek(conn, cache, key, key_hint, peek_modes, binary, query_id)


async def cache_local_peek_async(
        conn: 'AioConnection', cache: Union[str, int], key: Any, key_hint: 'IgniteDataType' = None,
        peek_modes: Union[int, list, tuple] = None, binary: bool = False,
        query_id: Optional[int] = None) -> 'APIResult':
    """
    Async version of cache_local_peek.
    """
    return await __cache_local_peek(conn, cache, key, key_hint, peek_modes, binary, query_id)


def __cache_local_peek(conn, cache, key, key_hint, peek_modes, binary, query_id):
    if peek_modes is None:
        peek_modes = []
    elif not isinstance(peek_modes, (list, tuple)):
        peek_modes = [peek_modes]

    query_struct = Query(
        OP_CACHE_LOCAL_PEEK,
        [
            ('hash_code', Int),
            ('flag', Byte),
            ('key', key_hint or AnyDataObject),
            ('peek_modes', ByteArray),
        ],
        query_id=query_id,
    )
    return query_perform(
        query_struct, conn,
        query_params={
            'hash_code': cache_id(cache),
            'flag': 1 if binary else 0,
            'key': key,
            'peek_modes': peek_modes,
        },
        response_config=[
            ('value', AnyDataObject),
        ],
        post_process_fun=__post_process_value_by_key('value')
    )


def __post_process_value_by_key(key):
    def internal(result):
        if result.status == 0:
            result.value = result.value[key]

        return result
    return internal