Moved factory logic to from_url method on each client

Author: codepr

CHANGES.md

@@ -1,3 +1,10 @@
+### 1.1.7
+(Apr 26, 2019)
+
+- Moved factory logic for client creation to from_url method on client module
+- Added TasqFuture result from clients result, to return more structured results
+  with additional informations about execution.
+
 ### 1.1.6
 (Apr 22, 2019)
 

README.md

@@ -25,7 +25,7 @@ production enviroments.
 Starting a worker on a node, with debug flag set to true on configuration file
 
 ```
-$ tq worker
+$ tq redis
 Listening for jobs on 127.0.0.1:9000
 Response actor started
 ```

tasq/__init__.py

@@ -1,11 +1,12 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+from tasq.queue import TasqQueue
 from tasq.jobqueue import JobQueue
-from tasq.worker import ProcessQueueWorker#, ThreadQueueWorker
+from tasq.worker import ProcessQueueWorker
 from tasq.remote.client import ZMQTasqClient, RedisTasqClient
 from tasq.remote.supervisor import (ZMQActorSupervisor,
                                     ZMQQueueSupervisor,
                                     RedisActorSupervisor,
                                     Supervisors)
 
-__version__ = '1.0.5'
+__version__ = '1.1.7'

tasq/queue.py

@@ -18,27 +18,12 @@ def init_client(client, host, port, *args, **kwargs):
     return client(host, port, *args, **kwargs)
 
 
-Client = namedtuple('Client', ('handler', 'arguments'))
-
-
-defaults = {
-    'redis': Client(RedisTasqClient, {'host': '127.0.0.1',
-                                      'port': 6379,
-                                      'db': 0,
-                                      'name': os.getpid()}),
-    'amqp': Client(RabbitMQTasqClient, {'host': '127.0.0.1',
-                                        'port': 5672,
-                                        'name': os.getpid()}),
-    'unix': Client(ZMQTasqClient, {'host': '127.0.0.1',
-                                   'port': 9000,
-                                   'plport': 9001,
-                                   'unix_socket': True}),
-    'zmq': Client(ZMQTasqClient, {'host': '127.0.0.1',
-                                  'port': 9000,
-                                  'plport': 9001}),
-    'tcp': Client(ZMQTasqClient, {'host': '127.0.0.1',
-                                  'port': 9000,
-                                  'plport': 9001})
+backends = {
+    'redis': RedisTasqClient,
+    'amqp': RabbitMQTasqClient,
+    'unix': ZMQTasqClient,
+    'zmq': ZMQTasqClient,
+    'tcp': ZMQTasqClient
 }
 
 
@@ -49,8 +34,8 @@ class TasqQueue:
 
     The formats accepted for the backends are:
 
-    - redis://localhost:6379/0?name=test-queue-redis
-    - amqp://localhost:5672?name=test-queue-rabbitmq
+    - redis://localhost:6379/0?name=redis-queue
+    - amqp://localhost:5672?name=amqp-queue
     - zmq://localhost:9000?plport=9010
     - tcp://localhost:5555
 
@@ -74,28 +59,14 @@ class TasqQueue:
     def __init__(self, backend=u'zmq://localhost:9000',
                  store=None, sign_data=False):
 
-        url = urlparse(backend)
-        scheme = url.scheme or 'zmq'
-        assert url.scheme in {'redis', 'zmq', 'amqp', 'unix', 'tcp'}, \
-            f"Unsupported {url.scheme}"
-        args = {
-            'host': url.hostname,
-            'port': url.port,
-            'db': int(url.path.split('/')[-1]) if url.path else None,
-            'name': url.query.split('=')[-1] if url.query and scheme not in {'tcp', 'zmq'} else None,
-            'plport': int(url.query.split('=')[-1]) if url.query and scheme == 'zmq' else None,
-            'sign_data': sign_data
-        }
-
-        # Update defaults arguments
-        for k in defaults[scheme].arguments:
-            if k not in args or not args[k]:
-                args[k] = defaults[scheme].arguments[k]
-
-        # Remove useless args
-        args = {k: v for k, v in args.items() if v is not None}
-
-        self._backend = defaults[scheme].handler(**args)
+        if isinstance(backend, str):
+            url = urlparse(backend)
+            scheme = url.scheme or 'zmq'
+            assert url.scheme in {'redis', 'zmq', 'amqp', 'unix', 'tcp'}, \
+                f"Unsupported {url.scheme}"
+            self._backend = backends[scheme].from_url(backend, sign_data)
+        else:
+            self._backend = backend
         # Handle only redis as a backend store for now
         if store:
             urlstore = urlparse(store)

tasq/remote/client.py

@@ -5,8 +5,10 @@
 remote workers.
 """
 
-from __future__ import absolute_import, division, print_function, unicode_literals
+from __future__ import (absolute_import, division,
+                        print_function, unicode_literals)
 
+from urllib.parse import urlparse
 from abc import ABCMeta, abstractmethod
 from concurrent.futures import Future
 from threading import Thread
@@ -23,10 +25,7 @@
 
 
 class TasqClientNotConnected(Exception):
-
-    def __init__(self, msg=u''):
-        self.message = msg
-        super().__init__(self.message)
+    pass
 
 
 class TasqFuture(Future):
@@ -44,10 +43,22 @@ def exec_time(self):
 
 class BaseTasqClient(metaclass=ABCMeta):
 
-    """
-    Simple client class to schedule jobs to remote workers, currently
+    """Simple client class to schedule jobs to remote workers, currently
     supports a synchronous way of calling tasks awaiting for results and an
     asynchronous one which collect results in a dedicated dictionary
+
+    Attributes
+    ----------
+    :type host: str
+    :param host: The IP address to connect with
+
+    :type port: int
+    :param port: The port associated with the host param
+
+    :type sign_data: bool or False
+    :param sign_data: Boolean flag, sign bytes passing around through sockets
+                      if True
+
     """
 
     def __init__(self, host, port, sign_data=False):
@@ -100,11 +111,6 @@ def __exit__(self, exc_type, exc_value, exc_traceback):
             pass
         self.close()
 
-    def __repr__(self):
-        status = 'connected' if self.is_connected else 'disconnected'
-        return f"<BaseTasqClient worker=(tcp://{self.host}:{self.port}, " \
-               f"tcp://{self.host}:{self.port}) status={status}>"
-
     @abstractmethod
     def _make_client(self):
         pass
@@ -188,7 +194,11 @@ def schedule_blocking(self, func, *args, **kwargs):
         :type func: func
         :param func: A function to be executed on a worker by enqueing it
 
+        :rtype: tasq.remote.client.TasqFuture
         :return: The result of the func execution
+
+        :raise: tasq.remote.client.TasqClientNotConnected, in case of not
+                connected client
         """
         if not self.is_connected:
             raise TasqClientNotConnected('Client not connected to no worker')
@@ -203,8 +213,31 @@ class ZMQTasqClient(BaseTasqClient):
     """Simple client class to schedule jobs to remote workers, currently
     supports a synchronous way of calling tasks awaiting for results and an
     asynchronous one which collect results in a dedicated dictionary
+
+    Attributes
+    ----------
+    :type host: str
+    :param host: The IP address to connect with
+
+    :type port: int
+    :param port: The port associated with the host param for PUSH channel
+                 communication
+
+    :type plport: int or None
+    :param plport: The pull port to retrieve bytes from
+
+    :type sign_data: bool or False
+    :param sign_data: Boolean flag, sign bytes passing around through sockets
+                      if True
+
+    :type unix_socket: bool or False
+    :param unix_socket: Boolean flag to decide wether to use a UNIX socket or a
+                        TCP one
+
     """
 
+    __extraparams__ = {'plport'}
+
     def __init__(self, host, port, plport=None, sign_data=False, unix_socket=False):
         self._plport = plport or port + 1
         # Unix socket flag, if set to true, unix sockets for interprocess
@@ -245,12 +278,52 @@ def _gather_results(self):
             except KeyError:
                 self._log.error("Can't update result: key not found")
 
+    @classmethod
+    def from_url(cls, url, sign_data=False):
+        u = urlparse(url)
+        scheme = u.scheme or 'zmq'
+        assert scheme in ('zmq', 'unix', 'tcp'), f"Unsupported {scheme}"
+        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?')}
+        extras = {k: v for k, v in extras.items() if k in cls.__extraparams__}
+        conn_args = {
+            'host': u.hostname or '127.0.0.1',
+            'port': u.port or 9000,
+            'plport': int(extras.get('plport', 0)),
+            'sign_data': sign_data,
+            'unix_socket': scheme == 'unix'
+        }
+        return cls(**conn_args)
+
 
 class RedisTasqClient(BaseTasqClient):
 
-    """"""
+    """Simple Redis client class to schedule jobs to remote workers using
+    redis as the backend broker.
+
+    Attributes
+    ----------
+    :type host: str or 'localhost'
+    :param host: The IP address of the Redis instance to connect with
+
+    :type port: int or 6379
+    :param port: The port associated with the host param
+
+    :type db: int or 0
+    :param db: The database to use on redis for the queues
+
+    :type name: str or redis-queue
+    :param name: The name of the redis queue
+
+    :type sign_data: bool or False
+    :param sign_data: Boolean flag, sign bytes passing around through sockets
+                      if True
+
+    """
+
+    __extraparams__ = {'db', 'name'}
 
-    def __init__(self, host, port, db, name, sign_data=False):
+    def __init__(self, host='localhost', port=6379,
+                 db=0, name='redis-queue', sign_data=False):
         self._db = db
         self._name = name
         super().__init__(host, port, sign_data)
@@ -304,12 +377,49 @@ def disconnect(self):
         if self.is_connected:
             self._is_connected = False
 
+    @classmethod
+    def from_url(cls, url, sign_data=False):
+        u = urlparse(url)
+        scheme = u.scheme or 'redis'
+        assert scheme == 'redis', f"Unsupported {scheme}"
+        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?')}
+        extras = {k: v for k, v in extras.items() if k in cls.__extraparams__}
+        conn_args = {
+            'host': u.hostname or 'localhost',
+            'port': u.port or 6379,
+            'db': int(extras.get('db', 0)),
+            'name': extras.get('name', 'redis-queue'),
+            'sign_data': sign_data
+        }
+        return cls(**conn_args)
+
 
 class RabbitMQTasqClient(BaseTasqClient):
 
-    """"""
+    """Simple RabbitMQ client class to schedule jobs to remote workers using
+    RabbitMQ as the backend broker.
+
+    Attributes
+    ----------
+    :type host: str or 'localhost'
+    :param host: The IP address of the RabbitMQ instance to connect with
+
+    :type port: int or 5672
+    :param port: The port associated with the host param
+
+    :type name: str or amqp-queue
+    :param name: The name of the RabbitMQ queue
+
+    :type sign_data: bool or False
+    :param sign_data: Boolean flag, sign bytes passing around through sockets
+                      if True
 
-    def __init__(self, host, port, name, sign_data=False):
+    """
+
+    __extraparams__ = {'name'}
+
+    def __init__(self, host='localhost', port=5672,
+                 name='amqp-queue', sign_data=False):
         self._name = name
         super().__init__(host, port, sign_data)
 
@@ -357,6 +467,21 @@ def disconnect(self):
         if self.is_connected:
             self._is_connected = False
 
+    @classmethod
+    def from_url(cls, url, sign_data=False):
+        u = urlparse(url)
+        scheme = u.scheme or 'amqp'
+        assert scheme == 'amqp', f"Unsupported {scheme}"
+        extras = {t.split('=')[0]: t.split('=')[1] for t in u.query.split('?')}
+        extras = {k: v for k, v in extras.items() if k in cls.__extraparams__}
+        conn_args = {
+            'host': u.hostname or 'localhost',
+            'port': u.port or 5672,
+            'name': extras.get('name', 'amqp-queue'),
+            'sign_data': sign_data
+        }
+        return cls(**conn_args)
+
 
 class TasqClientPool: