332 integrate tarantool driver abstraction (#339)

* renamed bigchain_pool -> validator_obj
* renamed the flask connection pool (class name)
* prepared AsyncIO separation
* renamed abci/core.py and class names, merged utils files
* removed obsolete file
* tidy up of ABCI application logic interface
* updated to newest driver tarantool 0.12.1
* Added new start options: --abci-only and --web-api-only to enable seperate execution of the services
* Added exception handling to the ABCI app
* removed space() object usage and thereby halved the amount of DB lookups
* removed async_io handling in the connection object but left some basics of the potential initialization
* tidied up the import structure/order
* tidied up imports
* set version number and changelog

Signed-off-by: Jürgen Eckel <juergen@riddleandcode.com>

CHANGELOG.md

@@ -25,6 +25,12 @@ For reference, the possible headings are:
 * **Known Issues**
 * **Notes**
 
+## [2.3.0] - 2023-01-03
+* **Fixed** double usage of the tarantool driver in one instance that lead to crashes
+* **Changed** refactored a lot of classes and the structure
+* **Changed** upgraded to tarantool driver 0.12.1
+
+
 ## [2.2.4] - 2023-15-02
 * **Fixed** subcondition instantiation now works recursively
 * **Changed** migrated dependency management to poetry

docker-compose.yml

@@ -23,7 +23,7 @@ services:
       - "8081:8081"
     volumes:
       - ./planetmint/backend/tarantool/init.lua:/opt/tarantool/init.lua
-    command: tarantool /opt/tarantool/init.lua
+    entrypoint: tarantool /opt/tarantool/init.lua
     restart: always
   planetmint:
     depends_on:

planetmint/abci/application_logic.py

@@ -9,9 +9,10 @@ with Tendermint.
 import logging
 import sys
 
-from tendermint.abci import types_pb2
 from abci.application import BaseApplication
 from abci.application import OkCode
+
+from tendermint.abci import types_pb2
 from tendermint.abci.types_pb2 import (
     ResponseInfo,
     ResponseInitChain,
@@ -23,47 +24,50 @@ from tendermint.abci.types_pb2 import (
 )
 
 from planetmint.application.validator import Validator
-from planetmint.model.models import Models
+from planetmint.abci.utils import decode_validator, decode_transaction, calculate_hash
-from planetmint.abci.tendermint_utils import decode_transaction, calculate_hash, decode_validator
 from planetmint.abci.block import Block
 from planetmint.ipc.events import EventTypes, Event
+from planetmint.backend.exceptions import DBConcurrencyError
 
 CodeTypeError = 1
 logger = logging.getLogger(__name__)
 
 
-class App(BaseApplication):
+class ApplicationLogic(BaseApplication):
     """Bridge between Planetmint and Tendermint.
 
     The role of this class is to expose the Planetmint
     transaction logic to Tendermint Core.
     """
 
-    def __init__(self, planetmint_node=None, events_queue=None, models: Models = None, validator: Validator = None):
+    def __init__(
+        self,
+        validator: Validator = None,
+        events_queue=None,
+    ):
         # super().__init__(abci)
         logger.debug("Checking values of types")
         logger.debug(dir(types_pb2))
         self.events_queue = events_queue
-        self.validator = Validator()
+        self.validator = validator if validator else Validator()
-        self.models = models or Models()
         self.block_txn_ids = []
         self.block_txn_hash = ""
         self.block_transactions = []
         self.validators = None
         self.new_height = None
-        self.chain = self.models.get_latest_abci_chain()
+        self.chain = self.validator.models.get_latest_abci_chain()
 
     def log_abci_migration_error(self, chain_id, validators):
         logger.error(
             "An ABCI chain migration is in process. "
-            "Download theself.planetmint_node.get_latest_abci_chain new ABCI client and configure it with "
+            "Download the self.planetmint_node.get_latest_abci_chain new ABCI client and configure it with "
             f"chain_id={chain_id} and validators={validators}."
         )
 
     def abort_if_abci_chain_is_not_synced(self):
         if self.chain is None or self.chain["is_synced"]:
             return
-        validators = self.models.get_validators()
+        validators = self.validator.models.get_validators()
         self.log_abci_migration_error(self.chain["chain_id"], validators)
         sys.exit(1)
 
@@ -71,33 +75,44 @@ class App(BaseApplication):
         """Initialize chain upon genesis or a migration"""
         app_hash = ""
         height = 0
-        known_chain = self.models.get_latest_abci_chain()
+        try:
+            known_chain = self.validator.models.get_latest_abci_chain()
             if known_chain is not None:
                 chain_id = known_chain["chain_id"]
 
                 if known_chain["is_synced"]:
-                msg = f"Got invalid InitChain ABCI request ({genesis}) - " f"the chain {chain_id} is already synced."
+                    msg = (
+                        f"Got invalid InitChain ABCI request ({genesis}) - " f"the chain {chain_id} is already synced."
+                    )
                     logger.error(msg)
                     sys.exit(1)
                 if chain_id != genesis.chain_id:
-                validators = self.models.get_validators()
+                    validators = self.validator.models.get_validators()
                     self.log_abci_migration_error(chain_id, validators)
                     sys.exit(1)
                 # set migration values for app hash and height
-            block = self.models.get_latest_block()
+                block = self.validator.models.get_latest_block()
                 app_hash = "" if block is None else block["app_hash"]
                 height = 0 if block is None else block["height"] + 1
-        known_validators = self.models.get_validators()
+            known_validators = self.validator.models.get_validators()
             validator_set = [decode_validator(v) for v in genesis.validators]
             if known_validators and known_validators != validator_set:
                 self.log_abci_migration_error(known_chain["chain_id"], known_validators)
                 sys.exit(1)
             block = Block(app_hash=app_hash, height=height, transactions=[])
-        self.models.store_block(block._asdict())
+            self.validator.models.store_block(block._asdict())
-        self.models.store_validator_set(height + 1, validator_set)
+            self.validator.models.store_validator_set(height + 1, validator_set)
             abci_chain_height = 0 if known_chain is None else known_chain["height"]
-        self.models.store_abci_chain(abci_chain_height, genesis.chain_id, True)
+            self.validator.models.store_abci_chain(abci_chain_height, genesis.chain_id, True)
-        self.chain = {"height": abci_chain_height, "is_synced": True, "chain_id": genesis.chain_id}
+            self.chain = {
+                "height": abci_chain_height,
+                "is_synced": True,
+                "chain_id": genesis.chain_id,
+            }
+        except DBConcurrencyError:
+            sys.exit(1)
+        except ValueError:
+            sys.exit(1)
         return ResponseInitChain()
 
     def info(self, request):
@@ -114,7 +129,13 @@ class App(BaseApplication):
         # logger.info(f"Tendermint version: {request.version}")
 
         r = ResponseInfo()
-        block = self.models.get_latest_block()
+        block = None
+        try:
+            block = self.validator.models.get_latest_block()
+        except DBConcurrencyError:
+            block = None
+        except ValueError:
+            block = None
         if block:
             chain_shift = 0 if self.chain is None else self.chain["height"]
             r.last_block_height = block["height"] - chain_shift
@@ -136,12 +157,17 @@ class App(BaseApplication):
 
         logger.debug("check_tx: %s", raw_transaction)
         transaction = decode_transaction(raw_transaction)
+        try:
             if self.validator.is_valid_transaction(transaction):
                 logger.debug("check_tx: VALID")
                 return ResponseCheckTx(code=OkCode)
             else:
                 logger.debug("check_tx: INVALID")
                 return ResponseCheckTx(code=CodeTypeError)
+        except DBConcurrencyError:
+            sys.exit(1)
+        except ValueError:
+            sys.exit(1)
 
     def begin_block(self, req_begin_block):
         """Initialize list of transaction.
@@ -169,7 +195,15 @@ class App(BaseApplication):
         self.abort_if_abci_chain_is_not_synced()
 
         logger.debug("deliver_tx: %s", raw_transaction)
-        transaction = self.validator.is_valid_transaction(decode_transaction(raw_transaction), self.block_transactions)
+        transaction = None
+        try:
+            transaction = self.validator.is_valid_transaction(
+                decode_transaction(raw_transaction), self.block_transactions
+            )
+        except DBConcurrencyError:
+            sys.exit(1)
+        except ValueError:
+            sys.exit(1)
 
         if not transaction:
             logger.debug("deliver_tx: INVALID")
@@ -198,10 +232,11 @@ class App(BaseApplication):
         # `end_block` or `commit`
         logger.debug(f"Updating pre-commit state: {self.new_height}")
         pre_commit_state = dict(height=self.new_height, transactions=self.block_txn_ids)
-        self.models.store_pre_commit_state(pre_commit_state)
+        try:
+            self.validator.models.store_pre_commit_state(pre_commit_state)
 
             block_txn_hash = calculate_hash(self.block_txn_ids)
-        block = self.models.get_latest_block()
+            block = self.validator.models.get_latest_block()
 
             logger.debug("BLOCK: ", block)
 
@@ -211,6 +246,10 @@ class App(BaseApplication):
                 self.block_txn_hash = block["app_hash"]
 
             validator_update = self.validator.process_block(self.new_height, self.block_transactions)
+        except DBConcurrencyError:
+            sys.exit(1)
+        except ValueError:
+            sys.exit(1)
 
         return ResponseEndBlock(validator_updates=validator_update)
 
@@ -220,15 +259,23 @@ class App(BaseApplication):
         self.abort_if_abci_chain_is_not_synced()
 
         data = self.block_txn_hash.encode("utf-8")
-
+        try:
             # register a new block only when new transactions are received
             if self.block_txn_ids:
-            self.models.store_bulk_transactions(self.block_transactions)
+                self.validator.models.store_bulk_transactions(self.block_transactions)
 
-        block = Block(app_hash=self.block_txn_hash, height=self.new_height, transactions=self.block_txn_ids)
+            block = Block(
+                app_hash=self.block_txn_hash,
+                height=self.new_height,
+                transactions=self.block_txn_ids,
+            )
             # NOTE: storing the block should be the last operation during commit
             # this effects crash recovery. Refer BEP#8 for details
-        self.models.store_block(block._asdict())
+            self.validator.models.store_block(block._asdict())
+        except DBConcurrencyError:
+            sys.exit(1)
+        except ValueError:
+            sys.exit(1)
 
         logger.debug(
             "Commit-ing new block with hash: apphash=%s ," "height=%s, txn ids=%s",
@@ -240,7 +287,11 @@ class App(BaseApplication):
         if self.events_queue:
             event = Event(
                 EventTypes.BLOCK_VALID,
-                {"height": self.new_height, "hash": self.block_txn_hash, "transactions": self.block_transactions},
+                {
+                    "height": self.new_height,
+                    "hash": self.block_txn_hash,
+                    "transactions": self.block_transactions,
+                },
             )
             self.events_queue.put(event)
 

planetmint/abci/parallel_validation.py

@@ -6,9 +6,9 @@
 import multiprocessing
 
 from collections import defaultdict
-from planetmint.abci.core import App
+from planetmint.abci.application_logic import ApplicationLogic
 from planetmint.application.validator import Validator
-from planetmint.abci.tendermint_utils import decode_transaction
+from planetmint.abci.utils import decode_transaction
 from abci.application import OkCode
 from tendermint.abci.types_pb2 import (
     ResponseCheckTx,
@@ -16,7 +16,7 @@ from tendermint.abci.types_pb2 import (
 )
 
 
-class ParallelValidationApp(App):
+class ParallelValidationApp(ApplicationLogic):
     def __init__(self, planetmint=None, events_queue=None):
         super().__init__(planetmint, events_queue)
         self.parallel_validator = ParallelValidator()

planetmint/abci/rpc.py

@@ -7,8 +7,7 @@ from transactions.common.transaction_mode_types import (
     BROADCAST_TX_SYNC,
 )
 
-from planetmint.utils import Singleton
+from planetmint.abci.utils import encode_transaction
-from planetmint.abci.tendermint_utils import encode_transaction
 from planetmint.application.validator import logger
 from planetmint.config_utils import autoconfigure
 from planetmint.config import Config

planetmint/abci/tendermint_utils.py

@@ -1,151 +0,0 @@
-# Copyright © 2020 Interplanetary Database Association e.V.,
-# Planetmint and IPDB software contributors.
-# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
-# Code is Apache-2.0 and docs are CC-BY-4.0
-
-import base64
-import hashlib
-import json
-import codecs
-
-from binascii import hexlify
-from tendermint.abci import types_pb2
-from tendermint.crypto import keys_pb2
-from hashlib import sha3_256
-from transactions.common.exceptions import InvalidPublicKey
-
-
-def encode_validator(v):
-    ed25519_public_key = v["public_key"]["value"]
-    pub_key = keys_pb2.PublicKey(ed25519=bytes.fromhex(ed25519_public_key))
-
-    return types_pb2.ValidatorUpdate(pub_key=pub_key, power=v["power"])
-
-
-def decode_validator(v):
-    return {
-        "public_key": {
-            "type": "ed25519-base64",
-            "value": codecs.encode(v.pub_key.ed25519, "base64").decode().rstrip("\n"),
-        },
-        "voting_power": v.power,
-    }
-
-
-def new_validator_set(validators, updates):
-    validators_dict = {}
-    for v in validators:
-        validators_dict[v["public_key"]["value"]] = v
-
-    updates_dict = {}
-    for u in updates:
-        decoder = get_public_key_decoder(u["public_key"])
-        public_key64 = base64.b64encode(decoder(u["public_key"]["value"])).decode("utf-8")
-        updates_dict[public_key64] = {
-            "public_key": {"type": "ed25519-base64", "value": public_key64},
-            "voting_power": u["power"],
-        }
-
-    new_validators_dict = {**validators_dict, **updates_dict}
-    return list(new_validators_dict.values())
-
-
-def get_public_key_decoder(pk):
-    encoding = pk["type"]
-    decoder = base64.b64decode
-
-    if encoding == "ed25519-base16":
-        decoder = base64.b16decode
-    elif encoding == "ed25519-base32":
-        decoder = base64.b32decode
-    elif encoding == "ed25519-base64":
-        decoder = base64.b64decode
-    else:
-        raise InvalidPublicKey("Invalid `type` specified for public key `value`")
-
-    return decoder
-
-
-def encode_transaction(value):
-    """Encode a transaction (dict) to Base64."""
-
-    return base64.b64encode(json.dumps(value).encode("utf8")).decode("utf8")
-
-
-def decode_transaction(raw):
-    """Decode a transaction from bytes to a dict."""
-
-    return json.loads(raw.decode("utf8"))
-
-
-def decode_transaction_base64(value):
-    """Decode a transaction from Base64."""
-
-    return json.loads(base64.b64decode(value.encode("utf8")).decode("utf8"))
-
-
-def calculate_hash(key_list):
-    if not key_list:
-        return ""
-
-    full_hash = sha3_256()
-    for key in key_list:
-        full_hash.update(key.encode("utf8"))
-
-    return full_hash.hexdigest()
-
-
-def merkleroot(hashes):
-    """Computes the merkle root for a given list.
-
-    Args:
-        hashes (:obj:`list` of :obj:`bytes`): The leaves of the tree.
-
-    Returns:
-        str: Merkle root in hexadecimal form.
-
-    """
-    # XXX TEMPORARY -- MUST REVIEW and possibly CHANGE
-    # The idea here is that the UTXO SET would be empty and this function
-    # would be invoked to compute the merkle root, and since there is nothing,
-    # i.e. an empty list, then the hash of the empty string is returned.
-    # This seems too easy but maybe that is good enough? TO REVIEW!
-    if not hashes:
-        return sha3_256(b"").hexdigest()
-    # XXX END TEMPORARY -- MUST REVIEW ...
-    if len(hashes) == 1:
-        return hexlify(hashes[0]).decode()
-    if len(hashes) % 2 == 1:
-        hashes.append(hashes[-1])
-    parent_hashes = [sha3_256(hashes[i] + hashes[i + 1]).digest() for i in range(0, len(hashes) - 1, 2)]
-    return merkleroot(parent_hashes)
-
-
-# ripemd160 is only available below python 3.9.13
-@DeprecationWarning
-def public_key64_to_address(base64_public_key):
-    """Note this only compatible with Tendermint 0.19.x"""
-    ed25519_public_key = public_key_from_base64(base64_public_key)
-    encoded_public_key = amino_encoded_public_key(ed25519_public_key)
-    return hashlib.new("ripemd160", encoded_public_key).hexdigest().upper()
-
-
-def public_key_from_base64(base64_public_key):
-    return key_from_base64(base64_public_key)
-
-
-def key_from_base64(base64_key):
-    return base64.b64decode(base64_key).hex().upper()
-
-
-def public_key_to_base64(ed25519_public_key):
-    return key_to_base64(ed25519_public_key)
-
-
-def key_to_base64(ed25519_key):
-    ed25519_key = bytes.fromhex(ed25519_key)
-    return base64.b64encode(ed25519_key).decode("utf-8")
-
-
-def amino_encoded_public_key(ed25519_public_key):
-    return bytes.fromhex("1624DE6220{}".format(ed25519_public_key))

planetmint/abci/utils.py

@@ -1,9 +1,16 @@
+import base64
+import codecs
+import hashlib
 import json
+from binascii import hexlify
+from hashlib import sha3_256
 
 from packaging import version
+from tendermint.abci import types_pb2
+from tendermint.crypto import keys_pb2
 from transactions.common.crypto import key_pair_from_ed25519_key
+from transactions.common.exceptions import InvalidPublicKey
 
-from planetmint.abci.tendermint_utils import key_from_base64
 from planetmint.version import __tm_supported_versions__
 
 
@@ -33,3 +40,138 @@ def tendermint_version_is_compatible(running_tm_ver):
         if version.parse(ver) == version.parse(tm_ver[0]):
             return True
     return False
+
+
+def encode_validator(v):
+    ed25519_public_key = v["public_key"]["value"]
+    pub_key = keys_pb2.PublicKey(ed25519=bytes.fromhex(ed25519_public_key))
+
+    return types_pb2.ValidatorUpdate(pub_key=pub_key, power=v["power"])
+
+
+def decode_validator(v):
+    return {
+        "public_key": {
+            "type": "ed25519-base64",
+            "value": codecs.encode(v.pub_key.ed25519, "base64").decode().rstrip("\n"),
+        },
+        "voting_power": v.power,
+    }
+
+
+def new_validator_set(validators, updates):
+    validators_dict = {}
+    for v in validators:
+        validators_dict[v["public_key"]["value"]] = v
+
+    updates_dict = {}
+    for u in updates:
+        decoder = get_public_key_decoder(u["public_key"])
+        public_key64 = base64.b64encode(decoder(u["public_key"]["value"])).decode("utf-8")
+        updates_dict[public_key64] = {
+            "public_key": {"type": "ed25519-base64", "value": public_key64},
+            "voting_power": u["power"],
+        }
+
+    new_validators_dict = {**validators_dict, **updates_dict}
+    return list(new_validators_dict.values())
+
+
+def get_public_key_decoder(pk):
+    encoding = pk["type"]
+    decoder = base64.b64decode
+
+    if encoding == "ed25519-base16":
+        decoder = base64.b16decode
+    elif encoding == "ed25519-base32":
+        decoder = base64.b32decode
+    elif encoding == "ed25519-base64":
+        decoder = base64.b64decode
+    else:
+        raise InvalidPublicKey("Invalid `type` specified for public key `value`")
+
+    return decoder
+
+
+def encode_transaction(value):
+    """Encode a transaction (dict) to Base64."""
+
+    return base64.b64encode(json.dumps(value).encode("utf8")).decode("utf8")
+
+
+def decode_transaction(raw):
+    """Decode a transaction from bytes to a dict."""
+
+    return json.loads(raw.decode("utf8"))
+
+
+def decode_transaction_base64(value):
+    """Decode a transaction from Base64."""
+
+    return json.loads(base64.b64decode(value.encode("utf8")).decode("utf8"))
+
+
+def calculate_hash(key_list):
+    if not key_list:
+        return ""
+
+    full_hash = sha3_256()
+    for key in key_list:
+        full_hash.update(key.encode("utf8"))
+
+    return full_hash.hexdigest()
+
+
+def merkleroot(hashes):
+    """Computes the merkle root for a given list.
+
+    Args:
+        hashes (:obj:`list` of :obj:`bytes`): The leaves of the tree.
+
+    Returns:
+        str: Merkle root in hexadecimal form.
+
+    """
+    # XXX TEMPORARY -- MUST REVIEW and possibly CHANGE
+    # The idea here is that the UTXO SET would be empty and this function
+    # would be invoked to compute the merkle root, and since there is nothing,
+    # i.e. an empty list, then the hash of the empty string is returned.
+    # This seems too easy but maybe that is good enough? TO REVIEW!
+    if not hashes:
+        return sha3_256(b"").hexdigest()
+    # XXX END TEMPORARY -- MUST REVIEW ...
+    if len(hashes) == 1:
+        return hexlify(hashes[0]).decode()
+    if len(hashes) % 2 == 1:
+        hashes.append(hashes[-1])
+    parent_hashes = [sha3_256(hashes[i] + hashes[i + 1]).digest() for i in range(0, len(hashes) - 1, 2)]
+    return merkleroot(parent_hashes)
+
+
+@DeprecationWarning
+def public_key64_to_address(base64_public_key):
+    """Note this only compatible with Tendermint 0.19.x"""
+    ed25519_public_key = public_key_from_base64(base64_public_key)
+    encoded_public_key = amino_encoded_public_key(ed25519_public_key)
+    return hashlib.new("ripemd160", encoded_public_key).hexdigest().upper()
+
+
+def public_key_from_base64(base64_public_key):
+    return key_from_base64(base64_public_key)
+
+
+def key_from_base64(base64_key):
+    return base64.b64decode(base64_key).hex().upper()
+
+
+def public_key_to_base64(ed25519_public_key):
+    return key_to_base64(ed25519_public_key)
+
+
+def key_to_base64(ed25519_key):
+    ed25519_key = bytes.fromhex(ed25519_key)
+    return base64.b64encode(ed25519_key).decode("utf-8")
+
+
+def amino_encoded_public_key(ed25519_public_key):
+    return bytes.fromhex("1624DE6220{}".format(ed25519_public_key))

planetmint/application/validator.py

@@ -22,16 +22,10 @@ from transactions.common.transaction import VALIDATOR_ELECTION, CHAIN_MIGRATION_
 from transactions.types.elections.election import Election
 from transactions.types.elections.validator_utils import election_id_to_public_key
 
-from planetmint.abci.tendermint_utils import (
+from planetmint.abci.utils import encode_validator, new_validator_set, key_from_base64, public_key_to_base64
-    merkleroot,
-    key_from_base64,
-    public_key_to_base64,
-    encode_validator,
-    new_validator_set,
-)
 from planetmint.application.basevalidationrules import BaseValidationRules
 from planetmint.backend.models.output import Output
-from planetmint.model.models import Models
+from planetmint.model.dataaccessor import DataAccessor
 from planetmint.config import Config
 from planetmint.config_utils import load_validation_plugin
 
@@ -39,12 +33,13 @@ logger = logging.getLogger(__name__)
 
 
 class Validator:
-    def __init__(self):
+    def __init__(self, async_io: bool = False):
-        self.models = Models()
+        self.async_io = async_io
-        self.validation = Validator._get_validationmethod()
+        self.models = DataAccessor(async_io=async_io)
+        self.validation = Validator._get_validation_method()
 
     @staticmethod
-    def _get_validationmethod():
+    def _get_validation_method():
         validationPlugin = Config().get().get("validation_plugin")
 
         if validationPlugin:

planetmint/backend/connection.py

@@ -11,7 +11,7 @@ from transactions.common.exceptions import ConfigurationError
 from planetmint.config import Config
 
 BACKENDS = {
-    "tarantool_db": "planetmint.backend.tarantool.connection.TarantoolDBConnection",
+    "tarantool_db": "planetmint.backend.tarantool.sync_io.connection.TarantoolDBConnection",
     "localmongodb": "planetmint.backend.localmongodb.connection.LocalMongoDBConnection",
 }
 
@@ -64,6 +64,7 @@ class DBConnection(metaclass=DBSingleton):
         backend: str = None,
         connection_timeout: int = None,
         max_tries: int = None,
+        async_io: bool = False,
         **kwargs
     ):
         """Create a new :class:`~.Connection` instance.

planetmint/backend/exceptions.py

@@ -22,5 +22,9 @@ class OperationDataInsertionError(BackendError):
     """Exception raised when a Database operation fails."""
 
 
+class DBConcurrencyError(BackendError):
+    """Exception raised when a Database operation fails."""
+
+
 class DuplicateKeyError(OperationError):
     """Exception raised when an insert fails because the key is not unique"""

planetmint/backend/tarantool/__init__.py

@@ -1,5 +1,2 @@
 # Register the single dispatched modules on import.
-from planetmint.backend.tarantool import query, connection, schema  # noqa
+from planetmint.backend.tarantool.sync_io import connection, query, schema
-
-# MongoDBConnection should always be accessed via
-# ``planetmint.backend.connect()``.

planetmint/backend/tarantool/init.lua

@@ -1,5 +1,10 @@
 box.cfg{listen = 3303}
 
+box.once("bootstrap", function()
+    box.schema.user.grant('guest','read,write,execute,create,drop','universe')
+end)
+
+
 function init()
     -- ABCI chains
     abci_chains = box.schema.create_space('abci_chains', { if_not_exists = true })

planetmint/backend/tarantool/sync_io/connection.py

@@ -6,6 +6,7 @@
 import logging
 import tarantool
 
+
 from planetmint.config import Config
 from transactions.common.exceptions import ConfigurationError
 from planetmint.utils import Lazy
@@ -55,11 +56,15 @@ class TarantoolDBConnection(DBConnection):
         with open(path, "r") as f:
             execute = f.readlines()
             f.close()
-        return "".join(execute).encode()
+        return "".join(execute).encode(encoding="utf-8")
 
     def connect(self):
         if not self.__conn:
-            self.__conn = tarantool.connect(host=self.host, port=self.port)
+            self.__conn = tarantool.Connection(
+                host=self.host, port=self.port, encoding="utf-8", connect_now=True, reconnect_delay=0.1
+            )
+        elif self.__conn.connected == False:
+            self.__conn.connect()
         return self.__conn
 
     def close(self):
@@ -74,65 +79,8 @@ class TarantoolDBConnection(DBConnection):
     def get_space(self, space_name: str):
         return self.connect().space(space_name)
 
-    def space(self, space_name: str):
-        return self.query().space(space_name)
-
-    def exec(self, query, only_data=True):
-        try:
-            conn = self.connect()
-            conn.execute(query) if only_data else conn.execute(query)
-        except tarantool.error.OperationalError as op_error:
-            raise op_error
-        except tarantool.error.NetworkError as net_error:
-            raise net_error
-
-    def run(self, query, only_data=True):
-        try:
-            conn = self.connect()
-            return query.run(conn).data if only_data else query.run(conn)
-        except tarantool.error.OperationalError as op_error:
-            raise op_error
-        except tarantool.error.NetworkError as net_error:
-            raise net_error
-
     def drop_database(self):
         self.connect().call("drop")
 
     def init_database(self):
         self.connect().call("init")
-
-    def run_command(self, command: str, config: dict):
-        from subprocess import run
-
-        try:
-            self.close()
-        except ConnectionError:
-            pass
-
-        print(f" commands: {command}")
-        host_port = "%s:%s" % (self.host, self.port)
-        execute_cmd = self._file_content_to_bytes(path=command)
-        output = run(
-            ["tarantoolctl", "connect", host_port],
-            input=execute_cmd,
-            capture_output=True,
-        ).stderr
-        output = output.decode()
-        return output
-
-    def run_command_with_output(self, command: str):
-        from subprocess import run
-
-        try:
-            self.close()
-        except ConnectionError:
-            pass
-
-        host_port = "%s:%s" % (
-            Config().get()["database"]["host"],
-            Config().get()["database"]["port"],
-        )
-        output = run(["tarantoolctl", "connect", host_port], input=command, capture_output=True)
-        if output.returncode != 0:
-            raise Exception(f"Error while trying to execute cmd {command} on host:port {host_port}: {output.stderr}")
-        return output.stdout

planetmint/backend/tarantool/sync_io/query.py

@@ -4,7 +4,6 @@
 # Code is Apache-2.0 and docs are CC-BY-4.0
 
 """Query implementation for Tarantool"""
-import json
 import logging
 from uuid import uuid4
 from operator import itemgetter
@@ -15,9 +14,8 @@ from planetmint.backend import query
 from planetmint.backend.models.dbtransaction import DbTransaction
 from planetmint.backend.exceptions import OperationDataInsertionError
 from planetmint.exceptions import CriticalDoubleSpend
+from planetmint.backend.exceptions import DBConcurrencyError
 from planetmint.backend.tarantool.const import (
-    TARANT_TABLE_META_DATA,
-    TARANT_TABLE_ASSETS,
     TARANT_TABLE_TRANSACTION,
     TARANT_TABLE_OUTPUT,
     TARANT_TABLE_SCRIPT,
@@ -35,13 +33,43 @@ from planetmint.backend.tarantool.const import (
 )
 from planetmint.backend.utils import module_dispatch_registrar
 from planetmint.backend.models import Asset, Block, Output
-from planetmint.backend.tarantool.connection import TarantoolDBConnection
+from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 from transactions.common.transaction import Transaction
 
-
 logger = logging.getLogger(__name__)
 register_query = module_dispatch_registrar(query)
 
+from tarantool.error import OperationalError, NetworkError, SchemaError
+from functools import wraps
+
+
+def catch_db_exception(function_to_decorate):
+    @wraps(function_to_decorate)
+    def wrapper(*args, **kw):
+        try:
+            output = function_to_decorate(*args, **kw)
+        except OperationalError as op_error:
+            raise op_error
+        except SchemaError as schema_error:
+            raise schema_error
+        except NetworkError as net_error:
+            raise net_error
+        except ValueError:
+            logger.info(f"ValueError in Query/DB instruction: {e}: raising DBConcurrencyError")
+            raise DBConcurrencyError
+        except AttributeError:
+            logger.info(f"Attribute in Query/DB instruction: {e}: raising DBConcurrencyError")
+            raise DBConcurrencyError
+        except Exception as e:
+            logger.info(f"Could not insert transactions: {e}")
+            if e.args[0] == 3 and e.args[1].startswith("Duplicate key exists in"):
+                raise CriticalDoubleSpend()
+            else:
+                raise OperationDataInsertionError()
+        return output
+
+    return wrapper
+
 
 @register_query(TarantoolDBConnection)
 def get_complete_transactions_by_ids(connection, txids: list) -> list[DbTransaction]:
@@ -59,8 +87,9 @@ def get_complete_transactions_by_ids(connection, txids: list) -> list[DbTransact
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_outputs_by_tx_id(connection, tx_id: str) -> list[Output]:
-    _outputs = connection.run(connection.space(TARANT_TABLE_OUTPUT).select(tx_id, index=TARANT_TX_ID_SEARCH))
+    _outputs = connection.connect().select(TARANT_TABLE_OUTPUT, tx_id, index=TARANT_TX_ID_SEARCH).data
     _sorted_outputs = sorted(_outputs, key=itemgetter(4))
     return [Output.from_tuple(output) for output in _sorted_outputs]
 
@@ -75,28 +104,34 @@ def get_transaction(connection, tx_id: str) -> Union[DbTransaction, None]:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_transactions_by_asset(connection, asset: str, limit: int = 1000) -> list[DbTransaction]:
-    txs = connection.run(
+    txs = (
-        connection.space(TARANT_TABLE_TRANSACTION).select(asset, limit=limit, index="transactions_by_asset_cid")
+        connection.connect()
+        .select(TARANT_TABLE_TRANSACTION, asset, limit=limit, index="transactions_by_asset_cid")
+        .data
     )
     tx_ids = [tx[0] for tx in txs]
     return get_complete_transactions_by_ids(connection, tx_ids)
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_transactions_by_metadata(connection, metadata: str, limit: int = 1000) -> list[DbTransaction]:
-    txs = connection.run(
+    txs = (
-        connection.space(TARANT_TABLE_TRANSACTION).select(metadata, limit=limit, index="transactions_by_metadata_cid")
+        connection.connect()
+        .select(TARANT_TABLE_TRANSACTION, metadata, limit=limit, index="transactions_by_metadata_cid")
+        .data
     )
     tx_ids = [tx[0] for tx in txs]
     return get_complete_transactions_by_ids(connection, tx_ids)
 
 
+@catch_db_exception
 def store_transaction_outputs(connection, output: Output, index: int) -> str:
     output_id = uuid4().hex
-    try:
+    connection.connect().insert(
-        connection.run(
+        TARANT_TABLE_OUTPUT,
-            connection.space(TARANT_TABLE_OUTPUT).insert(
         (
             output_id,
             int(output.amount),
@@ -104,13 +139,9 @@ def store_transaction_outputs(connection, output: Output, index: int) -> str:
             output.condition.to_dict(),
             index,
             output.transaction_id,
-                )
+        ),
-            )
+    ).data
-        )
     return output_id
-    except Exception as e:
-        logger.info(f"Could not insert Output: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
@@ -124,6 +155,7 @@ def store_transactions(connection, signed_transactions: list, table=TARANT_TABLE
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_transaction(connection, transaction, table=TARANT_TABLE_TRANSACTION):
     scripts = None
     if TARANT_TABLE_SCRIPT in transaction:
@@ -142,19 +174,13 @@ def store_transaction(connection, transaction, table=TARANT_TABLE_TRANSACTION):
         transaction["inputs"],
         scripts,
     )
-    try:
+    connection.connect().insert(table, tx)
-        connection.run(connection.space(table).insert(tx), only_data=False)
-    except Exception as e:
-        logger.info(f"Could not insert transactions: {e}")
-        if e.args[0] == 3 and e.args[1].startswith("Duplicate key exists in"):
-            raise CriticalDoubleSpend()
-        else:
-            raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_transaction_by_id(connection, transaction_id, table=TARANT_TABLE_TRANSACTION):
-    txs = connection.run(connection.space(table).select(transaction_id, index=TARANT_ID_SEARCH), only_data=False)
+    txs = connection.connect().select(table, transaction_id, index=TARANT_ID_SEARCH)
     if len(txs) == 0:
         return None
     return DbTransaction.from_tuple(txs[0])
@@ -172,18 +198,18 @@ def get_transactions(connection, transactions_ids: list) -> list[DbTransaction]:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_asset(connection, asset_id: str) -> Asset:
-    _data = connection.run(
+    connection.connect().select(TARANT_TABLE_TRANSACTION, asset_id, index=TARANT_INDEX_TX_BY_ASSET_ID).data
-        connection.space(TARANT_TABLE_TRANSACTION).select(asset_id, index=TARANT_INDEX_TX_BY_ASSET_ID)
-    )
     return Asset.from_dict(_data[0])
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_assets(connection, assets_ids: list) -> list[Asset]:
     _returned_data = []
     for _id in list(set(assets_ids)):
-        res = connection.run(connection.space(TARANT_TABLE_TRANSACTION).select(_id, index=TARANT_INDEX_TX_BY_ASSET_ID))
+        res = connection.connect().select(TARANT_TABLE_TRANSACTION, _id, index=TARANT_INDEX_TX_BY_ASSET_ID).data
         if len(res) == 0:
             continue
         _returned_data.append(res[0])
@@ -193,18 +219,24 @@ def get_assets(connection, assets_ids: list) -> list[Asset]:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_spent(connection, fullfil_transaction_id: str, fullfil_output_index: str) -> list[DbTransaction]:
-    _inputs = connection.run(
+    _inputs = (
-        connection.space(TARANT_TABLE_TRANSACTION).select(
+        connection.connect()
-            [fullfil_transaction_id, fullfil_output_index], index=TARANT_INDEX_SPENDING_BY_ID_AND_OUTPUT_INDEX
+        .select(
+            TARANT_TABLE_TRANSACTION,
+            [fullfil_transaction_id, fullfil_output_index],
+            index=TARANT_INDEX_SPENDING_BY_ID_AND_OUTPUT_INDEX,
         )
+        .data
     )
     return get_complete_transactions_by_ids(txids=[inp[0] for inp in _inputs], connection=connection)
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_latest_block(connection) -> Union[dict, None]:
-    blocks = connection.run(connection.space(TARANT_TABLE_BLOCKS).select())
+    blocks = connection.connect().select(TARANT_TABLE_BLOCKS).data
     if not blocks:
         return None
 
@@ -214,37 +246,32 @@ def get_latest_block(connection) -> Union[dict, None]:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_block(connection, block: dict):
     block_unique_id = uuid4().hex
-    try:
+    connection.connect().insert(
-        connection.run(
+        TARANT_TABLE_BLOCKS, (block_unique_id, block["app_hash"], block["height"], block[TARANT_TABLE_TRANSACTION])
-            connection.space(TARANT_TABLE_BLOCKS).insert(
-                (block_unique_id, block["app_hash"], block["height"], block[TARANT_TABLE_TRANSACTION])
-            ),
-            only_data=False,
     )
-    except Exception as e:
-        logger.info(f"Could not insert block: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_txids_filtered(connection, asset_ids: list[str], operation: str = "", last_tx: bool = False) -> list[str]:
     transactions = []
     if operation == "CREATE":
-        transactions = connection.run(
+        transactions = (
-            connection.space(TARANT_TABLE_TRANSACTION).select(
+            connection.connect()
-                [asset_ids[0], operation], index="transactions_by_id_and_operation"
+            .select(TARANT_TABLE_TRANSACTION, [asset_ids[0], operation], index="transactions_by_id_and_operation")
-            )
+            .data
         )
     elif operation == "TRANSFER":
-        transactions = connection.run(
+        transactions = (
-            connection.space(TARANT_TABLE_TRANSACTION).select(asset_ids, index=TARANT_INDEX_TX_BY_ASSET_ID)
+            connection.connect().select(TARANT_TABLE_TRANSACTION, asset_ids, index=TARANT_INDEX_TX_BY_ASSET_ID).data
         )
     else:
-        txs = connection.run(connection.space(TARANT_TABLE_TRANSACTION).select(asset_ids, index=TARANT_ID_SEARCH))
+        txs = connection.connect().select(TARANT_TABLE_TRANSACTION, asset_ids, index=TARANT_ID_SEARCH).data
-        asset_txs = connection.run(
+        asset_txs = (
-            connection.space(TARANT_TABLE_TRANSACTION).select(asset_ids, index=TARANT_INDEX_TX_BY_ASSET_ID)
+            connection.connect().select(TARANT_TABLE_TRANSACTION, asset_ids, index=TARANT_INDEX_TX_BY_ASSET_ID).data
         )
         transactions = txs + asset_txs
 
@@ -258,8 +285,9 @@ def get_txids_filtered(connection, asset_ids: list[str], operation: str = "", la
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_owned_ids(connection, owner: str) -> list[DbTransaction]:
-    outputs = connection.run(connection.space(TARANT_TABLE_OUTPUT).select(owner, index="public_keys"))
+    outputs = connection.connect().select(TARANT_TABLE_OUTPUT, owner, index="public_keys").data
     if len(outputs) == 0:
         return []
     txids = [output[5] for output in outputs]
@@ -283,8 +311,9 @@ def get_spending_transactions(connection, inputs):
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_block(connection, block_id=None) -> Union[dict, None]:
-    _block = connection.run(connection.space(TARANT_TABLE_BLOCKS).select(block_id, index="height", limit=1))
+    _block = connection.connect().select(TARANT_TABLE_BLOCKS, block_id, index="height", limit=1).data
     if len(_block) == 0:
         return
     _block = Block.from_tuple(_block[0])
@@ -292,8 +321,9 @@ def get_block(connection, block_id=None) -> Union[dict, None]:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_block_with_transaction(connection, txid: str) -> Union[dict, None]:
-    _block = connection.run(connection.space(TARANT_TABLE_BLOCKS).select(txid, index="block_by_transaction_id"))
+    _block = connection.connect().select(TARANT_TABLE_BLOCKS, txid, index="block_by_transaction_id").data
     if len(_block) == 0:
         return
     _block = Block.from_tuple(_block[0])
@@ -301,30 +331,28 @@ def get_block_with_transaction(connection, txid: str) -> Union[dict, None]:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def delete_transactions(connection, txn_ids: list):
-    try:
     for _id in txn_ids:
         _outputs = get_outputs_by_tx_id(connection, _id)
         for x in range(len(_outputs)):
             connection.connect().call("delete_output", (_outputs[x].id))
     for _id in txn_ids:
-            connection.run(connection.space(TARANT_TABLE_TRANSACTION).delete(_id), only_data=False)
+        connection.connect().delete(TARANT_TABLE_TRANSACTION, _id)
-            connection.run(connection.space(TARANT_TABLE_GOVERNANCE).delete(_id), only_data=False)
+        connection.connect().delete(TARANT_TABLE_GOVERNANCE, _id)
-    except Exception as e:
-        logger.info(f"Could not insert unspent output: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_unspent_outputs(connection, *unspent_outputs: list):
     result = []
     if unspent_outputs:
         for utxo in unspent_outputs:
             try:
-                output = connection.run(
+                output = (
-                    connection.space(TARANT_TABLE_UTXOS).insert(
+                    connection.connect()
-                        (uuid4().hex, utxo["transaction_id"], utxo["output_index"], utxo)
+                    .insert(TARANT_TABLE_UTXOS, (uuid4().hex, utxo["transaction_id"], utxo["output_index"], utxo))
-                    )
+                    .data
                 )
                 result.append(output)
             except Exception as e:
@@ -334,50 +362,51 @@ def store_unspent_outputs(connection, *unspent_outputs: list):
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def delete_unspent_outputs(connection, *unspent_outputs: list):
     result = []
     if unspent_outputs:
         for utxo in unspent_outputs:
-            output = connection.run(
+            output = (
-                connection.space(TARANT_TABLE_UTXOS).delete(
+                connection.connect()
-                    (utxo["transaction_id"], utxo["output_index"]), index="utxo_by_transaction_id_and_output_index"
+                .delete(
+                    TARANT_TABLE_UTXOS,
+                    (utxo["transaction_id"], utxo["output_index"]),
+                    index="utxo_by_transaction_id_and_output_index",
                 )
+                .data
             )
             result.append(output)
     return result
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_unspent_outputs(connection, query=None):  # for now we don't have implementation for 'query'.
-    _utxos = connection.run(connection.space(TARANT_TABLE_UTXOS).select([]))
+    _utxos = connection.connect().select(TARANT_TABLE_UTXOS, []).data
     return [utx[3] for utx in _utxos]
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_pre_commit_state(connection, state: dict):
-    _precommit = connection.run(connection.space(TARANT_TABLE_PRE_COMMITS).select([], limit=1))
+    _precommit = connection.connect().select(TARANT_TABLE_PRE_COMMITS, [], limit=1).data
     _precommitTuple = (
         (uuid4().hex, state["height"], state[TARANT_TABLE_TRANSACTION])
         if _precommit is None or len(_precommit) == 0
         else _precommit[0]
     )
-    try:
+    connection.connect().upsert(
-        connection.run(
+        TARANT_TABLE_PRE_COMMITS,
-            connection.space(TARANT_TABLE_PRE_COMMITS).upsert(
         _precommitTuple,
         op_list=[("=", 1, state["height"]), ("=", 2, state[TARANT_TABLE_TRANSACTION])],
-                limit=1,
-            ),
-            only_data=False,
     )
-    except Exception as e:
-        logger.info(f"Could not insert pre commit state: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_pre_commit_state(connection) -> dict:
-    _commit = connection.run(connection.space(TARANT_TABLE_PRE_COMMITS).select([], index=TARANT_ID_SEARCH))
+    _commit = connection.connect().select(TARANT_TABLE_PRE_COMMITS, [], index=TARANT_ID_SEARCH).data
     if _commit is None or len(_commit) == 0:
         return None
     _commit = sorted(_commit, key=itemgetter(1), reverse=False)[0]
@@ -385,71 +414,56 @@ def get_pre_commit_state(connection) -> dict:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_validator_set(conn, validators_update: dict):
-    _validator = conn.run(
+    _validator = (
-        conn.space(TARANT_TABLE_VALIDATOR_SETS).select(validators_update["height"], index="height", limit=1)
+        conn.connect().select(TARANT_TABLE_VALIDATOR_SETS, validators_update["height"], index="height", limit=1).data
     )
     unique_id = uuid4().hex if _validator is None or len(_validator) == 0 else _validator[0][0]
-    try:
+    conn.connect().upsert(
-        conn.run(
+        TARANT_TABLE_VALIDATOR_SETS,
-            conn.space(TARANT_TABLE_VALIDATOR_SETS).upsert(
         (unique_id, validators_update["height"], validators_update["validators"]),
         op_list=[("=", 1, validators_update["height"]), ("=", 2, validators_update["validators"])],
-                limit=1,
-            ),
-            only_data=False,
     )
-    except Exception as e:
-        logger.info(f"Could not insert validator set: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def delete_validator_set(connection, height: int):
-    _validators = connection.run(connection.space(TARANT_TABLE_VALIDATOR_SETS).select(height, index="height"))
+    _validators = connection.connect().select(TARANT_TABLE_VALIDATOR_SETS, height, index="height").data
     for _valid in _validators:
-        connection.run(connection.space(TARANT_TABLE_VALIDATOR_SETS).delete(_valid[0]), only_data=False)
+        connection.connect().delete(TARANT_TABLE_VALIDATOR_SETS, _valid[0])
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_election(connection, election_id: str, height: int, is_concluded: bool):
-    try:
+    connection.connect().upsert(
-        connection.run(
+        TARANT_TABLE_ELECTIONS, (election_id, height, is_concluded), op_list=[("=", 1, height), ("=", 2, is_concluded)]
-            connection.space(TARANT_TABLE_ELECTIONS).upsert(
-                (election_id, height, is_concluded), op_list=[("=", 1, height), ("=", 2, is_concluded)], limit=1
-            ),
-            only_data=False,
     )
-    except Exception as e:
-        logger.info(f"Could not insert election: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_elections(connection, elections: list):
-    try:
     for election in elections:
-            _election = connection.run(  # noqa: F841
+        _election = connection.connect().insert(
-                connection.space(TARANT_TABLE_ELECTIONS).insert(
+            TARANT_TABLE_ELECTIONS, (election["election_id"], election["height"], election["is_concluded"])
-                    (election["election_id"], election["height"], election["is_concluded"])
-                ),
-                only_data=False,
         )
-    except Exception as e:
-        logger.info(f"Could not insert elections: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def delete_elections(connection, height: int):
-    _elections = connection.run(connection.space(TARANT_TABLE_ELECTIONS).select(height, index="height"))
+    _elections = connection.connect().select(TARANT_TABLE_ELECTIONS, height, index="height").data
     for _elec in _elections:
-        connection.run(connection.space(TARANT_TABLE_ELECTIONS).delete(_elec[0]), only_data=False)
+        connection.connect().delete(TARANT_TABLE_ELECTIONS, _elec[0])
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_validator_set(connection, height: int = None):
-    _validators = connection.run(connection.space(TARANT_TABLE_VALIDATOR_SETS).select())
+    _validators = connection.connect().select(TARANT_TABLE_VALIDATOR_SETS).data
     if height is not None and _validators is not None:
         _validators = [
             {"height": validator[1], "validators": validator[2]} for validator in _validators if validator[1] <= height
@@ -462,8 +476,9 @@ def get_validator_set(connection, height: int = None):
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_election(connection, election_id: str) -> dict:
-    _elections = connection.run(connection.space(TARANT_TABLE_ELECTIONS).select(election_id, index=TARANT_ID_SEARCH))
+    _elections = connection.connect().select(TARANT_TABLE_ELECTIONS, election_id, index=TARANT_ID_SEARCH).data
     if _elections is None or len(_elections) == 0:
         return None
     _election = sorted(_elections, key=itemgetter(0), reverse=True)[0]
@@ -471,39 +486,38 @@ def get_election(connection, election_id: str) -> dict:
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_asset_tokens_for_public_key(connection, asset_id: str, public_key: str) -> list[DbTransaction]:
-    id_transactions = connection.run(connection.space(TARANT_TABLE_GOVERNANCE).select([asset_id]))
+    id_transactions = connection.connect().select(TARANT_TABLE_GOVERNANCE, [asset_id]).data
-    asset_id_transactions = connection.run(
+    asset_id_transactions = (
-        connection.space(TARANT_TABLE_GOVERNANCE).select([asset_id], index="governance_by_asset_id")
+        connection.connect().select(TARANT_TABLE_GOVERNANCE, [asset_id], index="governance_by_asset_id").data
     )
+
     transactions = id_transactions + asset_id_transactions
     return get_complete_transactions_by_ids(connection, [_tx[0] for _tx in transactions])
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def store_abci_chain(connection, height: int, chain_id: str, is_synced: bool = True):
-    try:
+    connection.connect().upsert(
-        connection.run(
+        TARANT_TABLE_ABCI_CHAINS,
-            connection.space(TARANT_TABLE_ABCI_CHAINS).upsert(
         (chain_id, height, is_synced),
         op_list=[("=", 0, chain_id), ("=", 1, height), ("=", 2, is_synced)],
-            ),
-            only_data=False,
     )
-    except Exception as e:
-        logger.info(f"Could not insert abci-chain: {e}")
-        raise OperationDataInsertionError()
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def delete_abci_chain(connection, height: int):
-    chains = connection.run(connection.space(TARANT_TABLE_ABCI_CHAINS).select(height, index="height"), only_data=False)
+    chains = connection.connect().select(TARANT_TABLE_ABCI_CHAINS, height, index="height")
-    connection.run(connection.space(TARANT_TABLE_ABCI_CHAINS).delete(chains[0][0], index="id"), only_data=False)
+    connection.connect().delete(TARANT_TABLE_ABCI_CHAINS, chains[0][0], index="id")
 
 
 @register_query(TarantoolDBConnection)
+@catch_db_exception
 def get_latest_abci_chain(connection) -> Union[dict, None]:
-    _all_chains = connection.run(connection.space(TARANT_TABLE_ABCI_CHAINS).select())
+    _all_chains = connection.connect().select(TARANT_TABLE_ABCI_CHAINS).data
     if _all_chains is None or len(_all_chains) == 0:
         return None
     _chain = sorted(_all_chains, key=itemgetter(1), reverse=True)[0]

planetmint/backend/tarantool/sync_io/schema.py

@@ -3,7 +3,7 @@ import logging
 from planetmint.config import Config
 from planetmint.backend.utils import module_dispatch_registrar
 from planetmint import backend
-from planetmint.backend.tarantool.connection import TarantoolDBConnection
+from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 
 logger = logging.getLogger(__name__)
 register_schema = module_dispatch_registrar(backend.schema)
@@ -32,19 +32,6 @@ def create_database(connection, dbname):
     logger.info("Create database `%s`.", dbname)
 
 
-def run_command_with_output(command):
-    from subprocess import run
-
-    host_port = "%s:%s" % (
-        Config().get()["database"]["host"],
-        Config().get()["database"]["port"],
-    )
-    output = run(["tarantoolctl", "connect", host_port], input=command, capture_output=True)
-    if output.returncode != 0:
-        raise Exception(f"Error while trying to execute cmd {command} on host:port {host_port}: {output.stderr}")
-    return output.stdout
-
-
 @register_schema(TarantoolDBConnection)
 def create_tables(connection, dbname):
     connection.connect().call("init")

planetmint/commands/planetmint.py

@@ -24,19 +24,16 @@ from transactions.types.elections.validator_election import ValidatorElection
 from transactions.common.transaction import Transaction
 
 
-from planetmint.abci.rpc import ABCI_RPC
-from planetmint.abci.utils import load_node_key
 from planetmint.application.validator import Validator
 from planetmint.backend import schema
 from planetmint.commands import utils
 from planetmint.commands.utils import configure_planetmint, input_on_stderr
 from planetmint.config_utils import setup_logging
-from planetmint.abci.tendermint_utils import public_key_from_base64
 from planetmint.abci.rpc import MODE_COMMIT, MODE_LIST
+from planetmint.abci.utils import load_node_key, public_key_from_base64
 from planetmint.commands.election_types import elections
 from planetmint.version import __tm_supported_versions__
 from planetmint.config import Config
-from planetmint.model.models import Models
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -113,7 +110,7 @@ def run_configure(args):
 def run_election(args):
     """Initiate and manage elections"""
 
-    b = Planetmint()
+    b = Validator()
 
     # Call the function specified by args.action, as defined above
     globals()[f"run_election_{args.action}"](args, b)
@@ -293,6 +290,7 @@ def run_start(args):
 
     validator = Validator()
     validator.rollback()
+    del validator
 
     logger.info("Starting Planetmint main process.")
     from planetmint.start import start
@@ -386,6 +384,21 @@ def create_parser():
         help="💀 EXPERIMENTAL: parallelize validation for better throughput 💀",
     )
 
+    start_parser.add_argument(
+        "--web-api-only",
+        dest="web_api_only",
+        default=False,
+        action="store_true",
+        help="💀 EXPERIMENTAL: seperate web API from ABCI server 💀",
+    )
+    start_parser.add_argument(
+        "--abci-only",
+        dest="abci_only",
+        default=False,
+        action="store_true",
+        help="💀 EXPERIMENTAL: seperate web API from ABCI server 💀",
+    )
+
     return parser
 
 

planetmint/model/dataaccessor.py

@@ -8,11 +8,10 @@ from transactions.common.exceptions import InputDoesNotExist
 
 from planetmint import config_utils, backend
 from planetmint.const import GOVERNANCE_TRANSACTION_TYPES
+from planetmint.model.fastquery import FastQuery
+from planetmint.abci.utils import key_from_base64
 from planetmint.backend.connection import Connection
 from planetmint.backend.tarantool.const import TARANT_TABLE_TRANSACTION, TARANT_TABLE_GOVERNANCE
-from planetmint.model.fastquery import FastQuery
-from planetmint.abci.tendermint_utils import key_from_base64
-
 from planetmint.backend.models.block import Block
 from planetmint.backend.models.output import Output
 from planetmint.backend.models.asset import Asset
@@ -20,10 +19,10 @@ from planetmint.backend.models.metadata import MetaData
 from planetmint.backend.models.dbtransaction import DbTransaction
 
 
-class Models:
+class DataAccessor:
-    def __init__(self, database_connection=None):
+    def __init__(self, database_connection=None, async_io: bool = False):
         config_utils.autoconfigure()
-        self.connection = database_connection if database_connection is not None else Connection()
+        self.connection = database_connection if database_connection is not None else Connection(async_io=async_io)
 
     def store_bulk_transactions(self, transactions):
         txns = []

planetmint/start.py

@@ -8,7 +8,7 @@ import setproctitle
 
 from planetmint.config import Config
 from planetmint.application.validator import Validator
-from planetmint.abci.core import App
+from planetmint.abci.application_logic import ApplicationLogic
 from planetmint.abci.parallel_validation import ParallelValidationApp
 from planetmint.web import server, websocket_server
 from planetmint.ipc.events import EventTypes
@@ -35,18 +35,20 @@ BANNER = """
 """
 
 
-def start(args):
+def start_web_api(args):
-    # Exchange object for event stream api
-    logger.info("Starting Planetmint")
-    exchange = Exchange()
-    # start the web api
     app_server = server.create_server(
         settings=Config().get()["server"], log_config=Config().get()["log"], planetmint_factory=Validator
     )
+    if args.web_api_only:
+        app_server.run()
+    else:
         p_webapi = Process(name="planetmint_webapi", target=app_server.run, daemon=True)
         p_webapi.start()
 
+
+def start_abci_server(args):
     logger.info(BANNER.format(__version__, Config().get()["server"]["bind"]))
+    exchange = Exchange()
 
     # start websocket server
     p_websocket_server = Process(
@@ -67,21 +69,29 @@ def start(args):
 
     setproctitle.setproctitle("planetmint")
 
-    # Start the ABCIServer
+    abci_server_app = None
+
+    publisher_queue = exchange.get_publisher_queue()
     if args.experimental_parallel_validation:
-        app = ABCIServer(
+        abci_server_app = ParallelValidationApp(events_queue=publisher_queue)
-            app=ParallelValidationApp(
-                events_queue=exchange.get_publisher_queue(),
-            )
-        )
     else:
-        app = ABCIServer(
+        abci_server_app = ApplicationLogic(events_queue=publisher_queue)
-            app=App(
+
-                events_queue=exchange.get_publisher_queue(),
+    app = ABCIServer(abci_server_app)
-            )
-        )
     app.run()
 
 
+def start(args):
+    logger.info("Starting Planetmint")
+
+    if args.web_api_only:
+        start_web_api(args)
+    elif args.abci_only:
+        start_abci_server(args)
+    else:
+        start_web_api(args)
+        start_abci_server(args)
+
+
 if __name__ == "__main__":
     start()

planetmint/version.py

@@ -3,8 +3,8 @@
 # SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
 # Code is Apache-2.0 and docs are CC-BY-4.0
 
-__version__ = "2.2.4"
+__version__ = "2.3.0"
-__short_version__ = "2.2"
+__short_version__ = "2.3"
 
 # Supported Tendermint versions
 __tm_supported_versions__ = ["0.34.15"]

planetmint/web/server.py

@@ -81,7 +81,7 @@ def create_app(*, debug=False, threads=1, planetmint_factory=None):
 
     app.debug = debug
 
-    app.config["bigchain_pool"] = utils.pool(planetmint_factory, size=threads)
+    app.config["validator_class_name"] = utils.pool(planetmint_factory, size=threads)
 
     add_routes(app)
 

planetmint/web/views/assets.py

@@ -26,9 +26,9 @@ class AssetListApi(Resource):
         if not args["limit"]:
             del args["limit"]
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
-        with pool() as validator:
+        with validator_class() as validator:
             assets = validator.models.get_assets_by_cid(cid, **args)
 
         try:

planetmint/web/views/blocks.py

@@ -20,9 +20,9 @@ class LatestBlock(Resource):
             A JSON string containing the data about the block.
         """
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
-        with pool() as validator:
+        with validator_class() as validator:
             block = validator.models.get_latest_block()
 
         if not block:
@@ -42,9 +42,9 @@ class BlockApi(Resource):
             A JSON string containing the data about the block.
         """
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
-        with pool() as validator:
+        with validator_class() as validator:
             block = validator.models.get_block(block_id=block_id)
 
         if not block:
@@ -68,9 +68,9 @@ class BlockListApi(Resource):
         args = parser.parse_args(strict=True)
         tx_id = args["transaction_id"]
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
-        with pool() as validator:
+        with validator_class() as validator:
             block = validator.models.get_block_containing_tx(tx_id)
 
         if not block:

planetmint/web/views/metadata.py

@@ -34,9 +34,9 @@ class MetadataApi(Resource):
         if not args["limit"]:
             del args["limit"]
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
-        with pool() as validator:
+        with validator_class() as validator:
             metadata = validator.models.get_metadata_by_cid(cid, **args)
 
         try:

planetmint/web/views/outputs.py

@@ -22,8 +22,8 @@ class OutputListApi(Resource):
         parser.add_argument("spent", type=parameters.valid_bool)
         args = parser.parse_args(strict=True)
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
-        with pool() as validator:
+        with validator_class() as validator:
             try:
                 outputs = validator.models.get_outputs_filtered(args["public_key"], args["spent"])
             except Exception as e:

planetmint/web/views/transactions.py

@@ -11,17 +11,15 @@ import logging
 
 from flask import current_app, request, jsonify
 from flask_restful import Resource, reqparse
+from transactions.common.transaction import Transaction
 from transactions.common.transaction_mode_types import BROADCAST_TX_ASYNC
 from transactions.common.exceptions import (
     SchemaValidationError,
     ValidationError,
 )
-
+from planetmint.abci.rpc import ABCI_RPC, MODE_COMMIT, MODE_LIST
-from planetmint.abci.rpc import ABCI_RPC
-from planetmint.web.views.base import make_error
 from planetmint.web.views import parameters
-from transactions.common.transaction import Transaction
+from planetmint.web.views.base import make_error
-from planetmint.abci.rpc import MODE_COMMIT, MODE_LIST
 
 logger = logging.getLogger(__name__)
 
@@ -36,9 +34,9 @@ class TransactionApi(Resource):
         Return:
             A JSON string containing the data about the transaction.
         """
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
-        with pool() as validator:
+        with validator_class() as validator:
             tx = validator.models.get_transaction(tx_id)
 
         if not tx:
@@ -54,7 +52,7 @@ class TransactionListApi(Resource):
         parser.add_argument("asset_ids", type=parameters.valid_txid_list, required=True)
         parser.add_argument("last_tx", type=parameters.valid_bool, required=False)
         args = parser.parse_args()
-        with current_app.config["bigchain_pool"]() as validator:
+        with current_app.config["validator_class_name"]() as validator:
             txs = validator.models.get_transactions_filtered(**args)
 
         return [tx.to_dict() for tx in txs]
@@ -70,7 +68,7 @@ class TransactionListApi(Resource):
         args = parser.parse_args()
         mode = str(args["mode"])
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
         # `force` will try to format the body of the POST request even if the
         # `content-type` header is not set to `application/json`
@@ -89,9 +87,9 @@ class TransactionListApi(Resource):
         except Exception as e:
             return make_error(500, "Invalid transaction ({}): {} - {}".format(type(e).__name__, e, tx), level="error")
 
-        with pool() as planet:
+        with validator_class() as validator:
             try:
-                planet.validate_transaction(tx_obj)
+                validator.validate_transaction(tx_obj)
             except ValidationError as e:
                 return make_error(400, "Invalid transaction ({}): {}".format(type(e).__name__, e))
             except Exception as e:

planetmint/web/views/validators.py

@@ -15,9 +15,9 @@ class ValidatorsApi(Resource):
             A JSON string containing the validator set of the current node.
         """
 
-        pool = current_app.config["bigchain_pool"]
+        validator_class = current_app.config["validator_class_name"]
 
-        with pool() as validator:
+        with validator_class() as validator:
             validators = validator.models.get_validators()
 
         return validators

poetry.lock

@@ -209,6 +209,55 @@ files = [
     {file = "async_timeout-4.0.2-py3-none-any.whl", hash = "sha256:8ca1e4fcf50d07413d66d1a5e416e42cfdf5851c981d679a09851a6853383b3c"},
 ]
 
+[[package]]
+name = "asynctnt"
+version = "2.0.1"
+description = "A fast Tarantool Database connector for Python/asyncio."
+category = "main"
+optional = false
+python-versions = "*"
+files = [
+    {file = "asynctnt-2.0.1-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:7c4b9e3c5cb557e4417b8845aca997441320788a0ea7d94d40692ca669239571"},
+    {file = "asynctnt-2.0.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:448959f8a1273bf7659c464c6159fccd426cbefabced9cfdc71e4cb48d518125"},
+    {file = "asynctnt-2.0.1-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c8851db8f629568e5249a1676a0b9e849a65185533a24308822b18d93601184a"},
+    {file = "asynctnt-2.0.1-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:47cd2d10c54a17fda502144bd9c77af83589ca25a1cfdfa0e59c1042aeb4df89"},
+    {file = "asynctnt-2.0.1-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:2ed53209c29fe488bc42e362ccd45f1353da59338dea7bf77811ba41369f81ad"},
+    {file = "asynctnt-2.0.1-cp310-cp310-win32.whl", hash = "sha256:0bc9c09cf5e63a5ae82d4e335410f7375e9b9f5c89c44ba94202441811b2d0e3"},
+    {file = "asynctnt-2.0.1-cp310-cp310-win_amd64.whl", hash = "sha256:eaca7eae2e0161bb16f775f4997e7b71aadf3b18deec8ddbe7a3d7e129cb3b2d"},
+    {file = "asynctnt-2.0.1-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:ef238ee5abf21c465d881ff94e8b5709637ce070ea84bc2071b5396320bd1eb9"},
+    {file = "asynctnt-2.0.1-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:15a2ab6be4b5cb12826a39ef57c027068d8c8ac8aee804913d288241236fc03d"},
+    {file = "asynctnt-2.0.1-cp36-cp36m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:18a7a9a6630336e97fa64401235e0647795c9b1633ce991638f1fa1e5fa8c97c"},
+    {file = "asynctnt-2.0.1-cp36-cp36m-musllinux_1_1_i686.whl", hash = "sha256:18086835b4db2aa6987097348156016854a43bfc6ed57de09f033393323afa7f"},
+    {file = "asynctnt-2.0.1-cp36-cp36m-musllinux_1_1_x86_64.whl", hash = "sha256:1bc9ec81543ea73e65c5849faa13763e054c6894b51b2fe0d05a9de075d11e48"},
+    {file = "asynctnt-2.0.1-cp36-cp36m-win32.whl", hash = "sha256:45bb81873916267d15d2d71600297a579bcf1698e4f75d5123b17de20e291c7b"},
+    {file = "asynctnt-2.0.1-cp36-cp36m-win_amd64.whl", hash = "sha256:400220c233561a494656bdf485319918cf0faf454d10b3be87ed652f93f63da2"},
+    {file = "asynctnt-2.0.1-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:4b1d9e1a389bc39ec304c00544a7ce865174079a559477eba68a19532a4313b7"},
+    {file = "asynctnt-2.0.1-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8c0dc3d0a7b9b2f12280abab0fe976984e1495407a1c7502e09d886ffeb124d4"},
+    {file = "asynctnt-2.0.1-cp37-cp37m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1b9e6dd3ec4a4396c2309393d828a2c5e7e9ae9b05e207219b3edd3ea691519b"},
+    {file = "asynctnt-2.0.1-cp37-cp37m-musllinux_1_1_i686.whl", hash = "sha256:61ac8b944b2a149947f07f8d5c9db10efd177deee1206c729039b8bb5e31fcb9"},
+    {file = "asynctnt-2.0.1-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:d1411d14fc31bdd7b9eb7b6e059a03fa373b8b6f6c7bd78d69a5488575182e65"},
+    {file = "asynctnt-2.0.1-cp37-cp37m-win32.whl", hash = "sha256:dd04c6bf05ddd78be50c8aee595aeeb4a7a49ab7ae84685567d2ba0cfcd0a606"},
+    {file = "asynctnt-2.0.1-cp37-cp37m-win_amd64.whl", hash = "sha256:bf6b94630b134c400daa1c7c1e2821afe7850e454c34795bdf11d2ff619e32fa"},
+    {file = "asynctnt-2.0.1-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:d5956086983cda76a5a3534d1bbf076476f36de3deb77d74f25f5b067ca92752"},
+    {file = "asynctnt-2.0.1-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:631c6e68e9efd1a1992c154b53bb3d0aa3ae361c47535ee72538efd92d5d9209"},
+    {file = "asynctnt-2.0.1-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:602f5e860773c6640cd4260f588dc4659962d2d810e0c7d02e872c6440bbc6c8"},
+    {file = "asynctnt-2.0.1-cp38-cp38-musllinux_1_1_i686.whl", hash = "sha256:d286d82d0f5c11ded09d9a067caf23c37a19906351ca8b9991d1e61fb6b31bf9"},
+    {file = "asynctnt-2.0.1-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:884897f42dc0f0e5eef9fd1a8a90e1ab811d1aaae3cdda2d6021125c7c5ddce9"},
+    {file = "asynctnt-2.0.1-cp38-cp38-win32.whl", hash = "sha256:98ee3194e834380e5cd79cc0b3fc07510c1023b81e1b341fe442fa94614a82f7"},
+    {file = "asynctnt-2.0.1-cp38-cp38-win_amd64.whl", hash = "sha256:b95fc8b7c0be5086a65b4b629114bf214f66a0cf20599e7464ddbbb06720ecd6"},
+    {file = "asynctnt-2.0.1-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:2da8dae75300cab35fedd8dbded6954bb82c05a596af9383d5a18b9d59d5fe47"},
+    {file = "asynctnt-2.0.1-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:afba9b5f4a1cbf3ba16ca15a0e190697d3f7e70c93e43a4fe12f39f03274ac8e"},
+    {file = "asynctnt-2.0.1-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:871b3ac7bc20bb3b383b4244e46dbad3de684a8562bdcd6765733d47c9da51c7"},
+    {file = "asynctnt-2.0.1-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:a834c3334430dd2fa3c93ce45b2c35eef1312c9931860c25b379ea69da9b96ed"},
+    {file = "asynctnt-2.0.1-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:8bf3d7e6ac447bf3504db837c5ec1ee9d8c275a0f9cb2319f9b5405bd228120b"},
+    {file = "asynctnt-2.0.1-cp39-cp39-win32.whl", hash = "sha256:6d77b0988b7c9cc15883442f85c23bb284986532a0ed9b9bf9ff710872133bfc"},
+    {file = "asynctnt-2.0.1-cp39-cp39-win_amd64.whl", hash = "sha256:a743511b91f4217a05e158277148c8d9523a152343a54c58b5a950ae52661c58"},
+    {file = "asynctnt-2.0.1.tar.gz", hash = "sha256:b11efe38122e2b9a658aadc4a4ad1182cbf7100b5d7040206a71f9f107206ebc"},
+]
+
+[package.dependencies]
+PyYAML = ">=5.0"
+
 [[package]]
 name = "attrs"
 version = "22.2.0"
@@ -1590,6 +1639,44 @@ files = [
 [package.dependencies]
 setuptools = "*"
 
+[[package]]
+name = "numpy"
+version = "1.24.2"
+description = "Fundamental package for array computing in Python"
+category = "main"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "numpy-1.24.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:eef70b4fc1e872ebddc38cddacc87c19a3709c0e3e5d20bf3954c147b1dd941d"},
+    {file = "numpy-1.24.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:e8d2859428712785e8a8b7d2b3ef0a1d1565892367b32f915c4a4df44d0e64f5"},
+    {file = "numpy-1.24.2-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6524630f71631be2dabe0c541e7675db82651eb998496bbe16bc4f77f0772253"},
+    {file = "numpy-1.24.2-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a51725a815a6188c662fb66fb32077709a9ca38053f0274640293a14fdd22978"},
+    {file = "numpy-1.24.2-cp310-cp310-win32.whl", hash = "sha256:2620e8592136e073bd12ee4536149380695fbe9ebeae845b81237f986479ffc9"},
+    {file = "numpy-1.24.2-cp310-cp310-win_amd64.whl", hash = "sha256:97cf27e51fa078078c649a51d7ade3c92d9e709ba2bfb97493007103c741f1d0"},
+    {file = "numpy-1.24.2-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:7de8fdde0003f4294655aa5d5f0a89c26b9f22c0a58790c38fae1ed392d44a5a"},
+    {file = "numpy-1.24.2-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4173bde9fa2a005c2c6e2ea8ac1618e2ed2c1c6ec8a7657237854d42094123a0"},
+    {file = "numpy-1.24.2-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:4cecaed30dc14123020f77b03601559fff3e6cd0c048f8b5289f4eeabb0eb281"},
+    {file = "numpy-1.24.2-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9a23f8440561a633204a67fb44617ce2a299beecf3295f0d13c495518908e910"},
+    {file = "numpy-1.24.2-cp311-cp311-win32.whl", hash = "sha256:e428c4fbfa085f947b536706a2fc349245d7baa8334f0c5723c56a10595f9b95"},
+    {file = "numpy-1.24.2-cp311-cp311-win_amd64.whl", hash = "sha256:557d42778a6869c2162deb40ad82612645e21d79e11c1dc62c6e82a2220ffb04"},
+    {file = "numpy-1.24.2-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:d0a2db9d20117bf523dde15858398e7c0858aadca7c0f088ac0d6edd360e9ad2"},
+    {file = "numpy-1.24.2-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:c72a6b2f4af1adfe193f7beb91ddf708ff867a3f977ef2ec53c0ffb8283ab9f5"},
+    {file = "numpy-1.24.2-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c29e6bd0ec49a44d7690ecb623a8eac5ab8a923bce0bea6293953992edf3a76a"},
+    {file = "numpy-1.24.2-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2eabd64ddb96a1239791da78fa5f4e1693ae2dadc82a76bc76a14cbb2b966e96"},
+    {file = "numpy-1.24.2-cp38-cp38-win32.whl", hash = "sha256:e3ab5d32784e843fc0dd3ab6dcafc67ef806e6b6828dc6af2f689be0eb4d781d"},
+    {file = "numpy-1.24.2-cp38-cp38-win_amd64.whl", hash = "sha256:76807b4063f0002c8532cfeac47a3068a69561e9c8715efdad3c642eb27c0756"},
+    {file = "numpy-1.24.2-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:4199e7cfc307a778f72d293372736223e39ec9ac096ff0a2e64853b866a8e18a"},
+    {file = "numpy-1.24.2-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:adbdce121896fd3a17a77ab0b0b5eedf05a9834a18699db6829a64e1dfccca7f"},
+    {file = "numpy-1.24.2-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:889b2cc88b837d86eda1b17008ebeb679d82875022200c6e8e4ce6cf549b7acb"},
+    {file = "numpy-1.24.2-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f64bb98ac59b3ea3bf74b02f13836eb2e24e48e0ab0145bbda646295769bd780"},
+    {file = "numpy-1.24.2-cp39-cp39-win32.whl", hash = "sha256:63e45511ee4d9d976637d11e6c9864eae50e12dc9598f531c035265991910468"},
+    {file = "numpy-1.24.2-cp39-cp39-win_amd64.whl", hash = "sha256:a77d3e1163a7770164404607b7ba3967fb49b24782a6ef85d9b5f54126cc39e5"},
+    {file = "numpy-1.24.2-pp38-pypy38_pp73-macosx_10_9_x86_64.whl", hash = "sha256:92011118955724465fb6853def593cf397b4a1367495e0b59a7e69d40c4eb71d"},
+    {file = "numpy-1.24.2-pp38-pypy38_pp73-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f9006288bcf4895917d02583cf3411f98631275bc67cce355a7f39f8c14338fa"},
+    {file = "numpy-1.24.2-pp38-pypy38_pp73-win_amd64.whl", hash = "sha256:150947adbdfeceec4e5926d956a06865c1c690f2fd902efede4ca6fe2e657c3f"},
+    {file = "numpy-1.24.2.tar.gz", hash = "sha256:003a9f530e880cb2cd177cba1af7220b9aa42def9c4afc2a2fc3ee6be7eb2b22"},
+]
+
 [[package]]
 name = "packaging"
 version = "23.0"
@@ -1602,6 +1689,55 @@ files = [
     {file = "packaging-23.0.tar.gz", hash = "sha256:b6ad297f8907de0fa2fe1ccbd26fdaf387f5f47c7275fedf8cce89f99446cf97"},
 ]
 
+[[package]]
+name = "pandas"
+version = "1.5.3"
+description = "Powerful data structures for data analysis, time series, and statistics"
+category = "main"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "pandas-1.5.3-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:3749077d86e3a2f0ed51367f30bf5b82e131cc0f14260c4d3e499186fccc4406"},
+    {file = "pandas-1.5.3-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:972d8a45395f2a2d26733eb8d0f629b2f90bebe8e8eddbb8829b180c09639572"},
+    {file = "pandas-1.5.3-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:50869a35cbb0f2e0cd5ec04b191e7b12ed688874bd05dd777c19b28cbea90996"},
+    {file = "pandas-1.5.3-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c3ac844a0fe00bfaeb2c9b51ab1424e5c8744f89860b138434a363b1f620f354"},
+    {file = "pandas-1.5.3-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7a0a56cef15fd1586726dace5616db75ebcfec9179a3a55e78f72c5639fa2a23"},
+    {file = "pandas-1.5.3-cp310-cp310-win_amd64.whl", hash = "sha256:478ff646ca42b20376e4ed3fa2e8d7341e8a63105586efe54fa2508ee087f328"},
+    {file = "pandas-1.5.3-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:6973549c01ca91ec96199e940495219c887ea815b2083722821f1d7abfa2b4dc"},
+    {file = "pandas-1.5.3-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:c39a8da13cede5adcd3be1182883aea1c925476f4e84b2807a46e2775306305d"},
+    {file = "pandas-1.5.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:f76d097d12c82a535fda9dfe5e8dd4127952b45fea9b0276cb30cca5ea313fbc"},
+    {file = "pandas-1.5.3-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e474390e60ed609cec869b0da796ad94f420bb057d86784191eefc62b65819ae"},
+    {file = "pandas-1.5.3-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5f2b952406a1588ad4cad5b3f55f520e82e902388a6d5a4a91baa8d38d23c7f6"},
+    {file = "pandas-1.5.3-cp311-cp311-win_amd64.whl", hash = "sha256:bc4c368f42b551bf72fac35c5128963a171b40dce866fb066540eeaf46faa003"},
+    {file = "pandas-1.5.3-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:14e45300521902689a81f3f41386dc86f19b8ba8dd5ac5a3c7010ef8d2932813"},
+    {file = "pandas-1.5.3-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:9842b6f4b8479e41968eced654487258ed81df7d1c9b7b870ceea24ed9459b31"},
+    {file = "pandas-1.5.3-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:26d9c71772c7afb9d5046e6e9cf42d83dd147b5cf5bcb9d97252077118543792"},
+    {file = "pandas-1.5.3-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:5fbcb19d6fceb9e946b3e23258757c7b225ba450990d9ed63ccceeb8cae609f7"},
+    {file = "pandas-1.5.3-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:565fa34a5434d38e9d250af3c12ff931abaf88050551d9fbcdfafca50d62babf"},
+    {file = "pandas-1.5.3-cp38-cp38-win32.whl", hash = "sha256:87bd9c03da1ac870a6d2c8902a0e1fd4267ca00f13bc494c9e5a9020920e1d51"},
+    {file = "pandas-1.5.3-cp38-cp38-win_amd64.whl", hash = "sha256:41179ce559943d83a9b4bbacb736b04c928b095b5f25dd2b7389eda08f46f373"},
+    {file = "pandas-1.5.3-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:c74a62747864ed568f5a82a49a23a8d7fe171d0c69038b38cedf0976831296fa"},
+    {file = "pandas-1.5.3-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:c4c00e0b0597c8e4f59e8d461f797e5d70b4d025880516a8261b2817c47759ee"},
+    {file = "pandas-1.5.3-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:a50d9a4336a9621cab7b8eb3fb11adb82de58f9b91d84c2cd526576b881a0c5a"},
+    {file = "pandas-1.5.3-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:dd05f7783b3274aa206a1af06f0ceed3f9b412cf665b7247eacd83be41cf7bf0"},
+    {file = "pandas-1.5.3-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9f69c4029613de47816b1bb30ff5ac778686688751a5e9c99ad8c7031f6508e5"},
+    {file = "pandas-1.5.3-cp39-cp39-win32.whl", hash = "sha256:7cec0bee9f294e5de5bbfc14d0573f65526071029d036b753ee6507d2a21480a"},
+    {file = "pandas-1.5.3-cp39-cp39-win_amd64.whl", hash = "sha256:dfd681c5dc216037e0b0a2c821f5ed99ba9f03ebcf119c7dac0e9a7b960b9ec9"},
+    {file = "pandas-1.5.3.tar.gz", hash = "sha256:74a3fd7e5a7ec052f183273dc7b0acd3a863edf7520f5d3a1765c04ffdb3b0b1"},
+]
+
+[package.dependencies]
+numpy = [
+    {version = ">=1.20.3", markers = "python_version < \"3.10\""},
+    {version = ">=1.21.0", markers = "python_version >= \"3.10\""},
+    {version = ">=1.23.2", markers = "python_version >= \"3.11\""},
+]
+python-dateutil = ">=2.8.1"
+pytz = ">=2020.1"
+
+[package.extras]
+test = ["hypothesis (>=5.5.3)", "pytest (>=6.0)", "pytest-xdist (>=1.31)"]
+
 [[package]]
 name = "parso"
 version = "0.8.3"
@@ -2175,7 +2311,7 @@ cffi = ">=1.4.1"
 six = "*"
 
 [package.extras]
-docs = ["sphinx (>=1.6.5)", "sphinx-rtd-theme"]
+docs = ["sphinx (>=1.6.5)", "sphinx_rtd_theme"]
 tests = ["hypothesis (>=3.27.0)", "pytest (>=3.2.1,!=3.3.0)"]
 
 [[package]]
@@ -2382,6 +2518,21 @@ files = [
     {file = "python-baseconv-1.2.2.tar.gz", hash = "sha256:0539f8bd0464013b05ad62e0a1673f0ac9086c76b43ebf9f833053527cd9931b"},
 ]
 
+[[package]]
+name = "python-dateutil"
+version = "2.8.2"
+description = "Extensions to the standard Python datetime module"
+category = "main"
+optional = false
+python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,>=2.7"
+files = [
+    {file = "python-dateutil-2.8.2.tar.gz", hash = "sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86"},
+    {file = "python_dateutil-2.8.2-py2.py3-none-any.whl", hash = "sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9"},
+]
+
+[package.dependencies]
+six = ">=1.5"
+
 [[package]]
 name = "python-decouple"
 version = "3.7"
@@ -2952,18 +3103,20 @@ tests = ["cython", "littleutils", "pygments", "pytest", "typeguard"]
 
 [[package]]
 name = "tarantool"
-version = "0.7.1"
+version = "0.12.1"
-description = "Python client library for Tarantool 1.6 Database"
+description = "Python client library for Tarantool"
 category = "main"
 optional = false
-python-versions = "*"
+python-versions = ">=3.6"
 files = [
-    {file = "tarantool-0.7.1-py3-none-any.whl", hash = "sha256:e557e5faf5337e6040eb324b43a21701986e0c37fae87d4c80011632faa20ff6"},
+    {file = "tarantool-0.12.1-py3-none-any.whl", hash = "sha256:711b47671aba6e6faedc71f57bc07a10f6d7ac728b696183e99a31f20082f187"},
-    {file = "tarantool-0.7.1.tar.gz", hash = "sha256:a4bf212e86c5f43dcb6baf89487f0db478a45e2c5a1b16926fbbc0e9aa6eae22"},
+    {file = "tarantool-0.12.1.tar.gz", hash = "sha256:80370cb5de0e35572f9515f09d8fc616367162d858ec8aacd3b537820b695c0e"},
 ]
 
 [package.dependencies]
-msgpack = ">=0.4.0"
+msgpack = "*"
+pandas = "*"
+pytz = "*"
 
 [[package]]
 name = "tomli"
@@ -3308,4 +3461,4 @@ testing = ["func-timeout", "jaraco.itertools", "pytest (>=6)", "pytest-black (>=
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "ce9a30d4e4ed6d8bdd7c6cfe8f1908ffb787fc111e77fc1f709dbe8c4dba7fbb"
+content-hash = "8b9c0f4ebc41bca0af100c124f3522e2e1052271ebde525b0b06b44496a2d105"

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "planetmint"
-version = "2.2.4"
+version = "2.3.0"
 description = "Planetmint: The Blockchain Database"
 authors = ["Planetmint contributors"]
 license = "AGPLv3"
@@ -26,7 +26,6 @@ python = "^3.9"
 chardet = "3.0.4"
 base58 = "2.1.1"
 aiohttp = "3.8.1"
-abci = "0.8.3"
 flask-cors = "3.0.10"
 flask-restful = "0.3.9"
 flask = "2.1.2"
@@ -35,7 +34,7 @@ jsonschema = "4.16.0"
 logstats = "0.3.0"
 packaging = ">=22.0"
 pymongo = "3.11.4"
-tarantool = "0.7.1"
+tarantool = ">=0.12.1"
 python-rapidjson = ">=1.0"
 pyyaml = "6.0.0"
 requests = "2.25.1"
@@ -47,6 +46,8 @@ planetmint-ipld = ">=0.0.3"
 pyasn1 = ">=0.4.8"
 python-decouple = "^3.7"
 planetmint-transactions = ">=0.7.0"
+asynctnt = "^2.0.1"
+abci = "^0.8.3"
 
 [tool.poetry.group.dev.dependencies]
 aafigure = "0.6"

tests/backend/tarantool/test_queries.py

@@ -10,7 +10,7 @@ pytestmark = pytest.mark.bdb
 
 
 def test_get_txids_filtered(signed_create_tx, signed_transfer_tx, db_conn):
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
     # create and insert two blocks, one for the create and one for the
     # transfer transaction
@@ -36,7 +36,7 @@ def test_get_txids_filtered(signed_create_tx, signed_transfer_tx, db_conn):
 
 
 def test_get_owned_ids(signed_create_tx, user_pk, db_conn):
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
     # insert a transaction
     query.store_transactions(connection=db_conn, signed_transactions=[signed_create_tx.to_dict()])
@@ -49,18 +49,18 @@ def test_get_owned_ids(signed_create_tx, user_pk, db_conn):
 
 def test_store_block(db_conn):
     from planetmint.abci.block import Block
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
     block = Block(app_hash="random_utxo", height=3, transactions=[])
     query.store_block(connection=db_conn, block=block._asdict())
     # block = query.get_block(connection=db_conn)
-    blocks = db_conn.run(db_conn.space("blocks").select([]))
+    blocks = db_conn.connect().select("blocks", []).data
     assert len(blocks) == 1
 
 
 def test_get_block(db_conn):
     from planetmint.abci.block import Block
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
     block = Block(app_hash="random_utxo", height=3, transactions=[])
 
@@ -71,7 +71,7 @@ def test_get_block(db_conn):
 
 
 def test_store_pre_commit_state(db_conn):
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
     state = dict(height=3, transactions=[])
 
@@ -84,11 +84,11 @@ def test_store_pre_commit_state(db_conn):
 
 
 def test_get_pre_commit_state(db_conn):
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
-    all_pre = db_conn.run(db_conn.space("pre_commits").select([]))
+    all_pre = db_conn.connect().select("pre_commits", []).data
     for pre in all_pre:
-        db_conn.run(db_conn.space("pre_commits").delete(pre[0]), only_data=False)
+        db_conn.connect().delete("pre_commits", pre[0])
     #  TODO First IN, First OUT
     state = dict(height=3, transactions=[])
     # db_context.conn.db.pre_commit.insert_one
@@ -98,10 +98,14 @@ def test_get_pre_commit_state(db_conn):
 
 
 def test_validator_update(db_conn):
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
     def gen_validator_update(height):
-        return {"validators": [], "height": height, "election_id": f"election_id_at_height_{height}"}
+        return {
+            "validators": [],
+            "height": height,
+            "election_id": f"election_id_at_height_{height}",
+        }
         # return {'data': 'somedata', 'height': height, 'election_id': f'election_id_at_height_{height}'}
 
     for i in range(1, 100, 10):
@@ -160,7 +164,7 @@ def test_validator_update(db_conn):
     ],
 )
 def test_store_abci_chain(description, stores, expected, db_conn):
-    from planetmint.backend.tarantool import query
+    from planetmint.backend.tarantool.sync_io import query
 
     for store in stores:
         query.store_abci_chain(db_conn, **store)

tests/backend/test_connection.py

@@ -8,7 +8,7 @@ import pytest
 
 def test_get_connection_raises_a_configuration_error(monkeypatch):
     from planetmint.backend.exceptions import ConnectionError
-    from planetmint.backend.tarantool.connection import TarantoolDBConnection
+    from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 
     with pytest.raises(ConnectionError):
         TarantoolDBConnection("localhost", "1337", "mydb", "password")
@@ -17,7 +17,6 @@ def test_get_connection_raises_a_configuration_error(monkeypatch):
 @pytest.mark.skip(reason="we currently do not suppport mongodb.")
 def test_get_connection_raises_a_configuration_error_mongodb(monkeypatch):
     from planetmint.backend.localmongodb.connection import LocalMongoDBConnection
-    from transactions.common.exceptions import ConfigurationError
 
     with pytest.raises(ConnectionError):
         conn = LocalMongoDBConnection("localhost", "1337", "mydb", "password")

tests/commands/test_commands.py

@@ -109,7 +109,7 @@ def test_bigchain_show_config(capsys):
 
 
 def test__run_init(mocker):
-    init_db_mock = mocker.patch("planetmint.backend.tarantool.connection.TarantoolDBConnection.init_database")
+    init_db_mock = mocker.patch("planetmint.backend.tarantool.sync_io.connection.TarantoolDBConnection.init_database")
 
     conn = Connection()
     conn.init_database()

tests/conftest.py

@@ -22,15 +22,14 @@ from logging import getLogger
 from logging.config import dictConfig
 
 from planetmint.backend.connection import Connection
-from planetmint.backend.tarantool.connection import TarantoolDBConnection
+from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 from transactions.common import crypto
 from transactions.common.transaction_mode_types import BROADCAST_TX_COMMIT
-from planetmint.abci.tendermint_utils import key_from_base64
+from planetmint.abci.utils import key_from_base64
 from planetmint.backend import schema, query
 from transactions.common.crypto import key_pair_from_ed25519_key, public_key_from_ed25519_key
 from planetmint.abci.block import Block
 from planetmint.abci.rpc import MODE_LIST
-from planetmint.model.models import Models
 from tests.utils import gen_vote
 from planetmint.config import Config
 from transactions.types.elections.validator_election import ValidatorElection  # noqa
@@ -142,6 +141,8 @@ def _bdb(_setup_database):
     from planetmint.config import Config
 
     conn = Connection()
+    conn.close()
+    conn.connect()
     yield
     dbname = Config().get()["database"]["name"]
     flush_db(conn, dbname)
@@ -241,7 +242,6 @@ def merlin():
 
 
 @pytest.fixture
-# def a():
 def abci_fixture():
     from tendermint.abci import types_pb2
 
@@ -250,9 +250,9 @@ def abci_fixture():
 
 @pytest.fixture
 def test_models():
-    from planetmint.model.models import Models
+    from planetmint.model.dataaccessor import DataAccessor
 
-    return Models()
+    return DataAccessor()
 
 
 @pytest.fixture
@@ -273,7 +273,10 @@ def test_abci_rpc():
 def b():
     from planetmint.application import Validator
 
-    return Validator()
+    validator = Validator()
+    validator.models.connection.close()
+    validator.models.connection.connect()
+    return validator
 
 
 @pytest.fixture
@@ -385,7 +388,10 @@ def db_name(db_config):
 
 @pytest.fixture
 def db_conn():
-    return Connection()
+    conn = Connection()
+    conn.close()
+    conn.connect()
+    return conn
 
 
 @pytest.fixture
@@ -453,10 +459,10 @@ def abci_server():
     from abci.server import ABCIServer
 
     # from tendermint.abci import types_pb2 as types_v0_34_11
-    from planetmint.abci.core import App
+    from planetmint.abci.application_logic import ApplicationLogic
     from planetmint.utils import Process
 
-    app = ABCIServer(app=App())
+    app = ABCIServer(app=ApplicationLogic())
     abci_proxy = Process(name="ABCI", target=app.run)
     yield abci_proxy.start()
     abci_proxy.terminate()

tests/db/test_planetmint_api.py

@@ -3,7 +3,6 @@
 # SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
 # Code is Apache-2.0 and docs are CC-BY-4.0
 import random
-import warnings
 from unittest.mock import patch
 import pytest
 from base58 import b58decode
@@ -14,7 +13,6 @@ from transactions.common.transaction import TransactionLink
 from transactions.common.transaction import Transaction
 from transactions.types.assets.create import Create
 from transactions.types.assets.transfer import Transfer
-from planetmint.model.fastquery import FastQuery
 from planetmint.exceptions import CriticalDoubleSpend
 
 pytestmark = pytest.mark.bdb
@@ -48,10 +46,8 @@ class TestBigchainApi(object):
             b.models.store_bulk_transactions([transfer_tx2])
 
     def test_double_inclusion(self, b, alice):
-        from tarantool.error import DatabaseError
-
         from planetmint.backend.exceptions import OperationError
-        from planetmint.backend.tarantool.connection import TarantoolDBConnection
+        from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 
         tx = Create.generate([alice.public_key], [([alice.public_key], 1)])
         tx = tx.sign([alice.private_key])

tests/tendermint/test_core.py

@@ -15,12 +15,11 @@ from transactions import ValidatorElection, ChainMigrationElection
 from transactions.common.crypto import generate_key_pair
 from transactions.types.assets.create import Create
 from transactions.types.assets.transfer import Transfer
-from planetmint.abci.core import App
+from planetmint.abci.application_logic import ApplicationLogic
 from planetmint.backend import query
-from planetmint.abci.core import OkCode, CodeTypeError
+from planetmint.abci.application_logic import OkCode, CodeTypeError
 from planetmint.abci.block import Block
-from planetmint.abci.tendermint_utils import new_validator_set
+from planetmint.abci.utils import new_validator_set, public_key_to_base64
-from planetmint.abci.tendermint_utils import public_key_to_base64
 from planetmint.version import __tm_supported_versions__
 from tests.utils import generate_election, generate_validators
 
@@ -49,7 +48,7 @@ def generate_init_chain_request(chain_id, vals=None):
 
 def test_init_chain_successfully_registers_chain(b):
     request = generate_init_chain_request("chain-XYZ")
-    res = App(b).init_chain(request)
+    res = ApplicationLogic(validator=b).init_chain(request)
     assert res == types.ResponseInitChain()
     chain = query.get_latest_abci_chain(b.models.connection)
     assert chain == {"height": 0, "chain_id": "chain-XYZ", "is_synced": True}
@@ -63,7 +62,7 @@ def test_init_chain_successfully_registers_chain(b):
 def test_init_chain_ignores_invalid_init_chain_requests(b):
     validators = [generate_validator()]
     request = generate_init_chain_request("chain-XYZ", validators)
-    res = App(b).init_chain(request)
+    res = ApplicationLogic(validator=b).init_chain(request)
     assert res == types.ResponseInitChain()
 
     validator_set = query.get_validator_set(b.models.connection)
@@ -77,7 +76,7 @@ def test_init_chain_ignores_invalid_init_chain_requests(b):
     ]
     for r in invalid_requests:
         with pytest.raises(SystemExit):
-            App(b).init_chain(r)
+            ApplicationLogic(validator=b).init_chain(r)
         # assert nothing changed - neither validator set, nor chain ID
         new_validator_set = query.get_validator_set(b.models.connection)
         assert new_validator_set == validator_set
@@ -93,7 +92,7 @@ def test_init_chain_ignores_invalid_init_chain_requests(b):
 def test_init_chain_recognizes_new_chain_after_migration(b):
     validators = [generate_validator()]
     request = generate_init_chain_request("chain-XYZ", validators)
-    res = App(b).init_chain(request)
+    res = ApplicationLogic(validator=b).init_chain(request)
     assert res == types.ResponseInitChain()
 
     validator_set = query.get_validator_set(b.models.connection)["validators"]
@@ -111,7 +110,7 @@ def test_init_chain_recognizes_new_chain_after_migration(b):
     ]
     for r in invalid_requests:
         with pytest.raises(SystemExit):
-            App(b).init_chain(r)
+            ApplicationLogic(validator=b).init_chain(r)
         assert query.get_latest_abci_chain(b.models.connection) == {
             "chain_id": "chain-XYZ-migrated-at-height-1",
             "is_synced": False,
@@ -123,7 +122,7 @@ def test_init_chain_recognizes_new_chain_after_migration(b):
     # a request with the matching chain ID and matching validator set
     # completes the migration
     request = generate_init_chain_request("chain-XYZ-migrated-at-height-1", validators)
-    res = App(b).init_chain(request)
+    res = ApplicationLogic(validator=b).init_chain(request)
     assert res == types.ResponseInitChain()
     assert query.get_latest_abci_chain(b.models.connection) == {
         "chain_id": "chain-XYZ-migrated-at-height-1",
@@ -144,7 +143,7 @@ def test_init_chain_recognizes_new_chain_after_migration(b):
     ]
     for r in invalid_requests:
         with pytest.raises(SystemExit):
-            App(b).init_chain(r)
+            ApplicationLogic(validator=b).init_chain(r)
         assert query.get_latest_abci_chain(b.models.connection) == {
             "chain_id": "chain-XYZ-migrated-at-height-1",
             "is_synced": True,
@@ -161,7 +160,7 @@ def test_init_chain_recognizes_new_chain_after_migration(b):
 
 def test_info(b):
     r = types.RequestInfo(version=__tm_supported_versions__[0])
-    app = App(b)
+    app = ApplicationLogic(validator=b)
 
     res = app.info(r)
     assert res.last_block_height == 0
@@ -174,7 +173,7 @@ def test_info(b):
 
     # simulate a migration and assert the height is shifted
     b.models.store_abci_chain(2, "chain-XYZ")
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     b.models.store_block(Block(app_hash="2", height=2, transactions=[])._asdict())
     res = app.info(r)
     assert res.last_block_height == 0
@@ -187,7 +186,7 @@ def test_info(b):
 
     # it's always the latest migration that is taken into account
     b.models.store_abci_chain(4, "chain-XYZ-new")
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     b.models.store_block(Block(app_hash="4", height=4, transactions=[])._asdict())
     res = app.info(r)
     assert res.last_block_height == 0
@@ -200,7 +199,7 @@ def test_check_tx__signed_create_is_ok(b):
 
     tx = Create.generate([alice.public_key], [([bob.public_key], 1)]).sign([alice.private_key])
 
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     result = app.check_tx(encode_tx_to_bytes(tx))
     assert result.code == OkCode
 
@@ -211,7 +210,7 @@ def test_check_tx__unsigned_create_is_error(b):
 
     tx = Create.generate([alice.public_key], [([bob.public_key], 1)])
 
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     result = app.check_tx(encode_tx_to_bytes(tx))
     assert result.code == CodeTypeError
 
@@ -223,7 +222,7 @@ def test_deliver_tx__valid_create_updates_db_and_emits_event(b, init_chain_reque
 
     tx = Create.generate([alice.public_key], [([bob.public_key], 1)]).sign([alice.private_key])
 
-    app = App(b, events)
+    app = ApplicationLogic(validator=b, events_queue=events)
 
     app.init_chain(init_chain_request)
 
@@ -253,7 +252,7 @@ def test_deliver_tx__double_spend_fails(b, init_chain_request):
 
     tx = Create.generate([alice.public_key], [([bob.public_key], 1)]).sign([alice.private_key])
 
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     app.init_chain(init_chain_request)
 
     begin_block = types.RequestBeginBlock()
@@ -270,7 +269,7 @@ def test_deliver_tx__double_spend_fails(b, init_chain_request):
 
 
 def test_deliver_transfer_tx__double_spend_fails(b, init_chain_request):
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     app.init_chain(init_chain_request)
 
     begin_block = types.RequestBeginBlock()
@@ -303,7 +302,7 @@ def test_deliver_transfer_tx__double_spend_fails(b, init_chain_request):
 
 
 def test_end_block_return_validator_updates(b, init_chain_request):
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     app.init_chain(init_chain_request)
 
     begin_block = types.RequestBeginBlock()
@@ -335,7 +334,7 @@ def test_end_block_return_validator_updates(b, init_chain_request):
 
 
 def test_store_pre_commit_state_in_end_block(b, alice, init_chain_request):
-    from planetmint.abci.core import App
+    from planetmint.abci.application_logic import ApplicationLogic
     from planetmint.backend import query
 
     tx = Create.generate(
@@ -344,7 +343,7 @@ def test_store_pre_commit_state_in_end_block(b, alice, init_chain_request):
         assets=[{"data": "QmaozNR7DZHQK1ZcU9p7QdrshMvXqWK6gpu5rmrkPdT3L4"}],
     ).sign([alice.private_key])
 
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     app.init_chain(init_chain_request)
 
     begin_block = types.RequestBeginBlock()
@@ -365,7 +364,7 @@ def test_store_pre_commit_state_in_end_block(b, alice, init_chain_request):
 
     # simulate a chain migration and assert the height is shifted
     b.models.store_abci_chain(100, "new-chain")
-    app = App(b)
+    app = ApplicationLogic(validator=b)
     app.begin_block(begin_block)
     app.deliver_tx(encode_tx_to_bytes(tx))
     app.end_block(types.RequestEndBlock(height=1))
@@ -470,39 +469,39 @@ def test_info_aborts_if_chain_is_not_synced(b):
     b.models.store_abci_chain(0, "chain-XYZ", False)
 
     with pytest.raises(SystemExit):
-        App(b).info(types.RequestInfo())
+        ApplicationLogic(validator=b).info(types.RequestInfo())
 
 
 def test_check_tx_aborts_if_chain_is_not_synced(b):
     b.models.store_abci_chain(0, "chain-XYZ", False)
 
     with pytest.raises(SystemExit):
-        App(b).check_tx("some bytes")
+        ApplicationLogic(validator=b).check_tx("some bytes")
 
 
 def test_begin_aborts_if_chain_is_not_synced(b):
     b.models.store_abci_chain(0, "chain-XYZ", False)
 
     with pytest.raises(SystemExit):
-        App(b).info(types.RequestBeginBlock())
+        ApplicationLogic(validator=b).info(types.RequestBeginBlock())
 
 
 def test_deliver_tx_aborts_if_chain_is_not_synced(b):
     b.models.store_abci_chain(0, "chain-XYZ", False)
 
     with pytest.raises(SystemExit):
-        App(b).deliver_tx("some bytes")
+        ApplicationLogic(validator=b).deliver_tx("some bytes")
 
 
 def test_end_block_aborts_if_chain_is_not_synced(b):
     b.models.store_abci_chain(0, "chain-XYZ", False)
 
     with pytest.raises(SystemExit):
-        App(b).info(types.RequestEndBlock())
+        ApplicationLogic(validator=b).info(types.RequestEndBlock())
 
 
 def test_commit_aborts_if_chain_is_not_synced(b):
     b.models.store_abci_chain(0, "chain-XYZ", False)
 
     with pytest.raises(SystemExit):
-        App(b).commit()
+        ApplicationLogic(validator=b).commit()

tests/tendermint/test_integration.py

@@ -21,11 +21,11 @@ from io import BytesIO
 
 @pytest.mark.bdb
 def test_app(b, eventqueue_fixture, init_chain_request):
-    from planetmint.abci.core import App
+    from planetmint.abci.application_logic import ApplicationLogic
-    from planetmint.abci.tendermint_utils import calculate_hash
+    from planetmint.abci.utils import calculate_hash
     from transactions.common.crypto import generate_key_pair
 
-    app = App(b, eventqueue_fixture)
+    app = ApplicationLogic(validator=b, events_queue=eventqueue_fixture)
     p = ProtocolHandler(app)
 
     data = p.process("info", types.Request(info=types.RequestInfo(version=__tm_supported_versions__[0])))

tests/tendermint/test_lib.py

@@ -29,7 +29,7 @@ from tests.utils import delete_unspent_outputs, get_utxoset_merkle_root, store_u
 def test_asset_is_separated_from_transaciton(b):
     import copy
     from transactions.common.crypto import generate_key_pair
-    from planetmint.backend.tarantool.connection import TarantoolDBConnection
+    from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 
     if isinstance(b.models.connection, TarantoolDBConnection):
         pytest.skip("This specific function is skipped because, assets are stored differently if using Tarantool")
@@ -102,7 +102,7 @@ def test_validation_error(b):
 @patch("requests.post")
 def test_write_and_post_transaction(mock_post, b, test_abci_rpc):
     from transactions.common.crypto import generate_key_pair
-    from planetmint.abci.tendermint_utils import encode_transaction
+    from planetmint.abci.utils import encode_transaction
 
     alice = generate_key_pair()
     tx = (
@@ -235,7 +235,7 @@ def test_store_zero_unspent_output(b):
 
 @pytest.mark.bdb
 def test_store_one_unspent_output(b, unspent_output_1, utxo_collection):
-    from planetmint.backend.tarantool.connection import TarantoolDBConnection
+    from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 
     res = store_unspent_outputs(b.models.connection, unspent_output_1)
     if not isinstance(b.models.connection, TarantoolDBConnection):

tests/tendermint/test_utils.py

@@ -11,7 +11,8 @@ from hashlib import sha3_256
 
 
 def test_encode_decode_transaction(b):
-    from planetmint.abci.tendermint_utils import encode_transaction, decode_transaction
+    from planetmint.abci.utils import decode_transaction
+    from planetmint.abci.utils import encode_transaction
 
     asset = {"value": "key"}
 
@@ -25,7 +26,7 @@ def test_encode_decode_transaction(b):
 
 
 def test_calculate_hash_no_key(b):
-    from planetmint.abci.tendermint_utils import calculate_hash
+    from planetmint.abci.utils import calculate_hash
 
     # pass an empty list
     assert calculate_hash([]) == ""
@@ -33,7 +34,7 @@ def test_calculate_hash_no_key(b):
 
 # TODO test for the case of an empty list of hashes, and possibly other cases.
 def test_merkleroot():
-    from planetmint.abci.tendermint_utils import merkleroot
+    from planetmint.abci.utils import merkleroot
 
     hashes = [sha3_256(i.encode()).digest() for i in "abc"]
     assert merkleroot(hashes) == ("78c7c394d3158c218916b7ae0ebdea502e0f4e85c08e3b371e3dfd824d389fa3")
@@ -49,14 +50,15 @@ SAMPLE_PUBLIC_KEY = {
     reason="ripemd160, the core of pulbic_key64_to_address is no longer supported by hashlib (from python 3.9.13 on)"
 )
 def test_convert_base64_public_key_to_address():
-    from planetmint.abci.tendermint_utils import public_key64_to_address
+    from planetmint.abci.utils import public_key64_to_address
 
     address = public_key64_to_address(SAMPLE_PUBLIC_KEY["pub_key"]["value"])
     assert address == SAMPLE_PUBLIC_KEY["address"]
 
 
 def test_public_key_encoding_decoding():
-    from planetmint.abci.tendermint_utils import public_key_from_base64, public_key_to_base64
+    from planetmint.abci.utils import public_key_to_base64
+    from planetmint.abci.utils import public_key_from_base64
 
     public_key = public_key_from_base64(SAMPLE_PUBLIC_KEY["pub_key"]["value"])
     base64_public_key = public_key_to_base64(public_key)

tests/upsert_validator/test_upsert_validator_vote.py

@@ -7,7 +7,7 @@ import pytest
 import codecs
 
 from planetmint.abci.rpc import MODE_LIST, MODE_COMMIT
-from planetmint.abci.tendermint_utils import public_key_to_base64
+from planetmint.abci.utils import public_key_to_base64
 
 from transactions.types.elections.validator_election import ValidatorElection
 from transactions.common.exceptions import AmountError

tests/upsert_validator/test_validator_election.py

@@ -7,7 +7,7 @@ import pytest
 
 from argparse import Namespace
 from unittest.mock import patch
-from planetmint.abci.tendermint_utils import public_key_to_base64
+from planetmint.abci.utils import public_key_to_base64
 from transactions.types.elections.validator_election import ValidatorElection
 from transactions.common.exceptions import (
     DuplicateTransaction,

tests/utils.py

@@ -13,14 +13,14 @@ from functools import singledispatch
 
 from planetmint import backend
 from planetmint.backend.localmongodb.connection import LocalMongoDBConnection
-from planetmint.backend.tarantool.connection import TarantoolDBConnection
+from planetmint.backend.tarantool.sync_io.connection import TarantoolDBConnection
 from planetmint.backend.schema import TABLES
 from transactions.common import crypto
 from transactions.common.transaction_mode_types import BROADCAST_TX_COMMIT
 from transactions.types.assets.create import Create
 from transactions.types.elections.vote import Vote
 from transactions.types.elections.validator_utils import election_id_to_public_key
-from planetmint.abci.tendermint_utils import key_to_base64, merkleroot
+from planetmint.abci.utils import merkleroot, key_to_base64
 from planetmint.abci.rpc import MODE_COMMIT, MODE_LIST
 
 

tests/web/test_content_type_middleware.py

@@ -38,7 +38,7 @@ def test_middleware_does_notstrip_content_type_from_other_methods():
     assert "CONTENT_TYPE" in mock.call_args[0][0]
 
 
-def test_get_outputs_endpoint_with_content_type(client, user_pk):
+def test_get_outputs_endpoint_with_content_type(client, user_pk, _bdb):
     res = client.get(
         OUTPUTS_ENDPOINT + "?public_key={}".format(user_pk), headers=[("Content-Type", "application/json")]
     )

tests/web/test_outputs.py

@@ -18,7 +18,7 @@ def test_get_outputs_endpoint(client, user_pk):
     m = MagicMock()
     m.txid = "a"
     m.output = 0
-    with patch("planetmint.model.models.Models.get_outputs_filtered") as gof:
+    with patch("planetmint.model.dataaccessor.DataAccessor.get_outputs_filtered") as gof:
         gof.return_value = [m, m]
         res = client.get(OUTPUTS_ENDPOINT + "?public_key={}".format(user_pk))
         assert res.json == [{"transaction_id": "a", "output_index": 0}, {"transaction_id": "a", "output_index": 0}]
@@ -30,7 +30,7 @@ def test_get_outputs_endpoint_unspent(client, user_pk):
     m = MagicMock()
     m.txid = "a"
     m.output = 0
-    with patch("planetmint.model.models.Models.get_outputs_filtered") as gof:
+    with patch("planetmint.model.dataaccessor.DataAccessor.get_outputs_filtered") as gof:
         gof.return_value = [m]
         params = "?spent=False&public_key={}".format(user_pk)
         res = client.get(OUTPUTS_ENDPOINT + params)
@@ -45,7 +45,7 @@ def test_get_outputs_endpoint_spent(client, user_pk):
     m = MagicMock()
     m.txid = "a"
     m.output = 0
-    with patch("planetmint.model.models.Models.get_outputs_filtered") as gof:
+    with patch("planetmint.model.dataaccessor.DataAccessor.get_outputs_filtered") as gof:
         gof.return_value = [m]
         params = "?spent=true&public_key={}".format(user_pk)
         res = client.get(OUTPUTS_ENDPOINT + params)

tests/web/test_transactions.py

@@ -40,7 +40,7 @@ def test_get_transaction_endpoint(client, posted_create_tx):
     assert res.status_code == 200
 
 
-def test_get_transaction_returns_404_if_not_found(client):
+def test_get_transaction_returns_404_if_not_found(client, b):
     res = client.get(TX_ENDPOINT + "123")
     assert res.status_code == 404
 
@@ -404,7 +404,7 @@ def test_transactions_get_list_good(client):
 
     asset_ids = ["1" * 64]
 
-    with patch("planetmint.model.models.Models.get_transactions_filtered", get_txs_patched):
+    with patch("planetmint.model.dataaccessor.DataAccessor.get_transactions_filtered", get_txs_patched):
         url = TX_ENDPOINT + "?asset_ids=" + ",".join(asset_ids)
         assert client.get(url).json == [
             ["asset_ids", asset_ids],
@@ -430,7 +430,7 @@ def test_transactions_get_list_bad(client):
         assert False
 
     with patch(
-        "planetmint.model.models.Models.get_transactions_filtered",
+        "planetmint.model.dataaccessor.DataAccessor.get_transactions_filtered",
         lambda *_, **__: should_not_be_called(),
     ):
         # Test asset id validated