Implement connect

Author: DifferentialOrange

tarantool/connection_pool.py

@@ -0,0 +1,162 @@
+# -*- coding: utf-8 -*-
+
+import json
+
+from tarantool.connection import Connection, ConnectionInterface
+from tarantool.const import (CLUSTER_DISCOVERY_DELAY, CONNECTION_TIMEOUT,
+                             RECONNECT_DELAY, RECONNECT_MAX_ATTEMPTS,
+                             SOCKET_TIMEOUT)
+from tarantool.error import ClusterConnectWarning, ConfigurationError, warn
+from tarantool.utils import ENCODING_DEFAULT
+
+try:
+    string_types = basestring
+except NameError:
+    string_types = str
+
+class Mode(Enum):
+    RW = 1
+    PREFER_RW = 2
+    PREFER_RO = 3
+
+def validate_address(address):
+    def format_error(address, err):
+        return None, 'Address %s: %s' % (str(address), err)
+
+    if not isinstance(address, dict):
+        return format_error(address, 'address must be a dict')
+
+    if 'port' not in address or address['port'] is None:
+        return format_error(address, 'port is not set or None')
+
+    if isinstance(address['port'], int):
+        # Looks like an inet address.
+
+        # Validate host.
+        if 'host' not in address or address['host'] is None:
+            return format_error(address,
+                                'host is mandatory for an inet address')
+        if not isinstance(address['host'], string_types):
+            return format_error(address,
+                                'host must be a string for an inet address')
+
+        # Validate port.
+        if not isinstance(address['port'], int):
+            return format_error(address,
+                                'port must be an int for an inet address')
+        if address['port'] < 1 or address['port'] > 65535:
+            return format_error(address, 'port must be in range [1, 65535] '
+                                         'for an inet address')
+
+        # Looks okay.
+        return True, None
+    elif isinstance(address['port'], string_types):
+        # Looks like a unix address.
+
+        # Expect no host.
+        if 'host' in address and address['host'] is not None:
+            return format_error(
+                address, 'host must be unset or None for a unix address')
+
+        # Validate port.
+        if not isinstance(address['port'], string_types):
+            return format_error(address,
+                                'port must be a string for a unix address')
+
+        # Looks okay.
+        return True, None
+
+    return format_error(address, 'port must be an int or a string')
+
+
+class ConnectionPool(ConnectionInterface):
+    def __init__(self,
+                 addrs,
+                 user=None,
+                 password=None,
+                 socket_timeout=SOCKET_TIMEOUT,
+                 reconnect_max_attempts=RECONNECT_MAX_ATTEMPTS,
+                 reconnect_delay=RECONNECT_DELAY,
+                 connect_now=True,
+                 encoding=ENCODING_DEFAULT,
+                 call_16=False,
+                 connection_timeout=CONNECTION_TIMEOUT,):
+        if not isinstance(addrs, list) or len(addrs) == 0:
+            raise ConfigurationError("addrs must be non-empty list")
+            addrs = []
+
+        # Verify addresses.
+        for addr in addrs:
+            ok, msg = validate_address(addr)
+            if not ok:
+                raise ConfigurationError(msg)
+        self.addrs = addrs
+
+        # Create connections
+        self.pool = {}
+
+        for addr in self.addrs:
+            key = json.dumps(addr)
+            self.pool[key] = Connection(
+                host=addr['host'],
+                port=addr['port'],
+                user=user,
+                password=password,
+                socket_timeout=socket_timeout,
+                reconnect_max_attempts=reconnect_max_attempts,
+                reconnect_delay=reconnect_delay,
+                connect_now=False,
+                encoding=encoding,
+                call_16=call_16,
+                connection_timeout=connection_timeout)
+
+        if connect_now:
+            self.connect()
+
+    def close(self):
+        raise NotImplementedError
+
+    def is_closed(self):
+        raise NotImplementedError
+
+    def connect(self):
+        for addr in self.addrs:
+            key = json.dumps(addr)
+            self.pool[key].connect()
+
+            if not self.pool[key].connected:
+                msg = 'Failed to connect to %s:%s' % str(addr['host'], addr['port'])
+                warn(msg, ClusterConnectWarning)
+
+    def call(self, func_name, *args):
+        raise NotImplementedError
+
+    def eval(self, expr, *args):
+        raise NotImplementedError
+
+    def replace(self, space_name, values):
+        raise NotImplementedError
+
+    def authenticate(self, user, password):
+        raise NotImplementedError
+
+    def insert(self, space_name, values):
+        raise NotImplementedError
+
+    def delete(self, space_name, key, **kwargs):
+        raise NotImplementedError
+
+    def upsert(self, space_name, tuple_value, op_list, **kwargs):
+        raise NotImplementedError
+
+    def update(self, space_name, key, op_list, **kwargs):
+        raise NotImplementedError
+
+    def ping(self, notime):
+        raise NotImplementedError
+
+    def select(self, space_name, key, **kwargs):
+        raise NotImplementedError
+
+    def execute(self, query, params):
+        raise NotImplementedError

tarantool/error.py

@@ -24,7 +24,6 @@
 import sys
 import warnings
 
-
 try:
     class Warning(StandardError):
         '''Exception raised for important warnings
@@ -223,6 +222,11 @@ class ClusterDiscoveryWarning(UserWarning):
     pass
 
 
+class ClusterConnectWarning(UserWarning):
+    '''Warning related to cluster pool connection'''
+    pass
+
+
 # always print this warnings
 warnings.filterwarnings("always", category=NetworkWarning)