Version 0.6.0

Gevent is supported.

Author: penguinolog

CHANGELOG.rst

@@ -1,5 +1,9 @@
 CHANGELOG
 =========
+Version 0.6.0
+-------------
+Gevent is supported.
+
 Version 0.5.0
 -------------
 CI/CD pipelines.

CI_REQUIREMENTS.txt

@@ -2,3 +2,4 @@ six
 typing >= 3.6
 futures>=3.1; python_version == "2.7"
 asyncio>=3.4; python_version == "3.3"
+gevent >= 1.2

README.rst

@@ -22,6 +22,7 @@ threaded is a set of decorators, which wrap functions in:
   * `concurrent.futures.ThreadPool`
   * `threading.Thread`
   * `asyncio.Task` in Python 3.
+  * `gevent.threadpool.ThreadPool` if gevent is installed.
 
 Why? Because copy-paste of `loop.create_task`, `threading.Thread` and `thread_pool.submit` is boring,
 especially if target functions is used by this way only.
@@ -46,15 +47,20 @@ Pros:
 
 Decorators:
 
-* `ThreadPooled` - native concurrent.futures.ThreadPool usage on Python 3 and it's backport on Python 2.7.
+* `ThreadPooled` - native ``concurrent.futures.ThreadPool`` usage on Python 3 and it's backport on Python 2.7.
 * `threadpooled` is alias for `ThreadPooled`.
 
-* `Threaded` - wrap in threading.Thread.
+* `Threaded` - wrap in ``threading.Thread``.
 * `threaded` is alias for `Threaded`.
 
-* `AsyncIOTask` - wrap in asyncio.Task. Uses the same API, as Python 3 `ThreadPooled`.
+* `AsyncIOTask` - wrap in ``asyncio.Task``. Uses the same API, as Python 3 `ThreadPooled`.
 * `asynciotask` is alias for `AsyncIOTask`.
 
+* `GThreadPooled` - wrap function in ``gevent.threadpool.ThreadPool``.
+* `gthreadpooled` is alias for `GThreadPooled`.
+
+.. note:: gevent is not in default package requirements.
+
 Usage
 =====
 
@@ -123,7 +129,7 @@ During application shutdown, pool can be stopped (while it will be recreated aut
 
 Threaded
 --------
-Classic threading.Thread. Useful for running until close and self-closing threads without return.
+Classic ``threading.Thread``. Useful for running until close and self-closing threads without return.
 
 Usage example:
 
@@ -168,7 +174,7 @@ it can be started automatically before return:
 
 AsyncIOTask
 -----------
-Wrap in asyncio.Task.
+Wrap in ``asyncio.Task``.
 
 .. note:: Python 3 only.
 
@@ -208,6 +214,28 @@ Usage with loop extraction from call arguments:
     loop = asyncio.get_event_loop()
     loop.run_until_complete(asyncio.wait_for(func(loop), timeout))
 
+GThreadPooled
+-------------
+Post function to ``gevent.threadpool.ThreadPool``.
+
+.. code-block:: python
+
+    threaded.GThreadPooled.configure(max_workers=3)
+
+.. note:: By if executor is not configured - it configures with default parameters: ``max_workers=(CPU_COUNT or 1) * 5``
+
+.. note:: Instead of standard ThreadPoolExecutor, gevent pool is not re-created during re-configuration.
+
+Basic usage example:
+
+.. code-block:: python
+
+    @threaded.GThreadPooled
+    def func():
+        pass
+
+    func().wait()
+
 Testing
 =======
 The main test mechanism for the package `threaded` is using `tox`.

doc/source/asynciotask.rst

@@ -1,4 +1,4 @@
-.. PrettyFormat, pretty_repr and pretty_str
+.. AsyncIOTask, asynciotask.
 
 API: Decorators: `AsyncIOTask`, `asynciotask`.
 ================================================

doc/source/gthreadpooled.rst

@@ -0,0 +1,55 @@
+.. GThreadPooled, gthreadpooled.
+
+API: Decorators: `GThreadPooled`, `gthreadpooled`.
+==================================================
+
+.. py:module:: pooled
+.. py:currentmodule:: pooled
+
+.. py:class:: GThreadPooled(func, )
+
+    Post function to gevent.threadpool.ThreadPool.
+
+    :param func: function to wrap
+    :type func: typing.Optional[typing.Callable[..., typing.Union[typing.Any, typing.Awaitable]]]
+
+    .. note:: Attributes is read-only
+
+    .. py:attribute:: executor
+
+        ``gevent.threadpool.ThreadPool`` instance. Class-wide.
+
+    .. py:attribute:: _func
+
+        ``typing.Optional[typing.Callable[..., typing.Union[typing.Any, typing.Awaitable]]]``
+        Wrapped function. Used for inheritance only.
+
+    .. py:classmethod:: configure(max_workers=None, hub=None)
+
+        Pool executor create and configure.
+
+        :param max_workers: Maximum workers
+        :type max_workers: typing.Optional[int]
+        :param hub: Event-loop hub
+        :type hub: typing.Optional[gevent.hub.Hub]
+
+        .. note:: max_workers=None means `(CPU_COUNT or 1) * 5`, it's default value.
+
+    .. py:classmethod:: shutdown
+
+        Shutdown executor.
+
+    .. py:method:: __call__(*args, **kwargs)
+
+        Decorator entry point.
+
+        :rtype: typing.Union[typing.Callable[..., gevent.event.AsyncResult], gevent.event.AsyncResult]
+
+
+.. py:function:: gthreadpooled(func, )
+
+    Post function to gevent.threadpool.ThreadPool.
+
+    :param func: function to wrap
+    :type func: typing.Optional[typing.Callable]
+    :rtype: ThreadPooled

doc/source/index.rst

@@ -13,6 +13,7 @@ Contents:
     threadpooled
     threaded
     asynciotask
+    gthreadpooled
 
 Indices and tables
 ==================

doc/source/threaded.rst

@@ -1,4 +1,4 @@
-.. logwrap function and LogWrap class description.
+.. Threaded class and threaded function.
 
 API: Decorators: `Threaded` class and `threaded` function.
 ==========================================================

doc/source/threadpooled.rst

@@ -1,4 +1,4 @@
-.. PrettyFormat, pretty_repr and pretty_str
+.. ThreadPooled, threadpooled.
 
 API: Decorators: `ThreadPooled`, `threadpooled`.
 ================================================
@@ -11,7 +11,7 @@ API: Decorators: `ThreadPooled`, `threadpooled`.
     Post function to ThreadPoolExecutor.
 
     :param func: function to wrap
-    :type func: typing.Optional[typing.Callable[..., typing.Union[typing.Callable, typing.Awaitable]]]
+    :type func: typing.Optional[typing.Callable[..., typing.Union[typing.Any, typing.Awaitable]]]
 
     :param loop_getter: Method to get event loop, if wrap in asyncio task
 
@@ -46,7 +46,7 @@ API: Decorators: `ThreadPooled`, `threadpooled`.
 
     .. py:attribute:: _func
 
-        ``typing.Optional[typing.Callable[..., typing.Union[typing.Callable, typing.Awaitable]]]``
+        ``typing.Optional[typing.Callable[..., typing.Union[typing.Any, typing.Awaitable]]]``
         Wrapped function. Used for inheritance only.
 
     .. py:classmethod:: configure(max_workers=None)
@@ -74,7 +74,7 @@ API: Decorators: `ThreadPooled`, `threadpooled`.
     Post function to ThreadPoolExecutor.
 
     :param func: function to wrap
-    :type func: typing.Optional[typing.Callable]
+    :type func: typing.Optional[typing.Callable[..., typing.Union[typing.Any, typing.Awaitable]]]
 
     .. note:: Next arguments is Python 3 only:
 

setup.py

@@ -134,6 +134,9 @@ def get_simple_vars_from_src(src):
         ':python_version == "3.3"': [
             'asyncio>=3.4',
         ],
+        'gevent': [
+            'gevent >= 1.2'
+        ],
     },
     install_requires=required,
 )

test/test_gevent_threadpooled.py

@@ -0,0 +1,93 @@
+#    Copyright 2017 Alexey Stepanov aka penguinolog
+##
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from __future__ import absolute_import
+from __future__ import unicode_literals
+
+import threading
+import unittest
+
+try:
+    import gevent
+    import gevent.threadpool
+except ImportError:
+    gevent = None
+
+import six
+
+import threaded
+
+if six.PY3:
+    from os import cpu_count
+else:
+    try:
+        from multiprocessing import cpu_count
+    except ImportError:
+        def cpu_count():
+            """Fake CPU count."""
+            return 1
+
+
+@unittest.skipIf(gevent is None, 'No gevent')
+class TestThreadPooled(unittest.TestCase):
+    def tearDown(self):
+        threaded.GThreadPooled.shutdown()
+
+    def test_thread_pooled_default(self):
+        @threaded.gthreadpooled
+        def test():
+            return threading.current_thread().name
+
+        pooled_name = test().wait()
+        self.assertNotEqual(pooled_name, threading.current_thread().name)
+
+    def test_thread_pooled_construct(self):
+        @threaded.gthreadpooled()
+        def test():
+            return threading.current_thread().name
+
+        pooled_name = test().wait()
+        self.assertNotEqual(pooled_name, threading.current_thread().name)
+
+    def test_thread_pooled_config(self):
+        thread_pooled = threaded.gthreadpooled()
+        thread_pooled.configure()
+
+        self.assertEqual(
+            thread_pooled.executor.maxsize,
+            (cpu_count() or 1) * 5
+        )
+
+        thread_pooled.configure(max_workers=2)
+
+        @thread_pooled
+        def test():
+            return threading.current_thread().name
+
+        pooled_name = test().wait()
+        self.assertNotEqual(pooled_name, threading.current_thread().name)
+        self.assertEqual(thread_pooled.executor.maxsize, 2)
+
+    def test_reconfigure(self):
+        """Gevent worker instance is re-used."""
+        thread_pooled = threaded.gthreadpooled()
+        executor = thread_pooled.executor
+
+        thread_pooled.configure(max_workers=executor.maxsize)
+        self.assertIs(executor, thread_pooled.executor)
+
+        old_size = executor.maxsize
+        thread_pooled.configure(max_workers=old_size + 1)
+        self.assertIs(executor, thread_pooled.executor)
+        self.assertEqual(executor.maxsize, old_size + 1)

threaded/__init__.py

@@ -28,13 +28,22 @@
         threaded,
         asynciotask
     )
+
+    try:  # pragma: no cover
+        from ._gthreadpooled3 import GThreadPooled, gthreadpooled
+    except ImportError:  # pragma: no cover
+        GThreadPooled = gthreadpooled = None
 else:  # pragma: no cover
     from ._threaded2 import (
         ThreadPooled,
         Threaded,
         threadpooled,
         threaded,
     )
+    try:  # pragma: no cover
+        from ._gthreadpooled2 import GThreadPooled, gthreadpooled
+    except ImportError:  # pragma: no cover
+        GThreadPooled = gthreadpooled = None
 # pylint: enable=no-name-in-module
 
 
@@ -48,6 +57,11 @@
         'AsyncIOTask',
         'asynciotask'
     )
+if GThreadPooled is not None:  # pragma: no cover
+    __all__ += (
+        'GThreadPooled',
+        'gthreadpooled'
+    )
 
-__version__ = '0.5.0'
+__version__ = '0.6.0'
 __author__ = "Alexey Stepanov <penguinolog@gmail.com>"

threaded/_base_gthreadpooled.py

@@ -0,0 +1,102 @@
+#    Copyright 2017 Alexey Stepanov aka penguinolog
+##
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""gevent.threadpool.ThreadPool usage."""
+
+from __future__ import absolute_import
+
+import gevent.threadpool
+import six
+
+from . import _base_threaded
+
+__all__ = (
+    'BaseGThreadPooled',
+)
+
+
+class BaseGThreadPooled(_base_threaded.APIPooled):
+    """Post function to gevent.threadpool.ThreadPool."""
+
+    __slots__ = ()
+
+    __executor = None
+
+    # pylint: disable=arguments-differ
+    @classmethod
+    def configure(
+        cls,
+        max_workers=None,
+        hub=None
+    ):
+        """Pool executor create and configure.
+
+        :param max_workers: Maximum workers
+        :type max_workers: typing.Optional[int]
+        :param hub: Event-loop hub
+        :type hub: typing.Optional[gevent.hub.Hub]
+        """
+        if max_workers is None:
+            max_workers = _base_threaded.cpu_count() * 5
+
+        if isinstance(cls.__executor, gevent.threadpool.ThreadPool):
+            if hub is None or hub == cls.__executor.hub:
+                if max_workers == cls.__executor.maxsize:
+                    return  # Nothing to change)
+                cls.__executor.maxsize = max_workers  # We can use internals
+                return
+            # Hub change. Very special case.
+            cls.__executor.kill()  # pragma: no cover
+
+        cls.__executor = gevent.threadpool.ThreadPool(
+            maxsize=max_workers,
+            hub=hub
+        )
+
+    # pylint: enable=arguments-differ
+
+    @classmethod
+    def shutdown(cls):
+        """Shutdown executor.
+
+        Due to not implemented method, set maxsize to 0 (do not accept new).
+        """
+        if cls.__executor is not None:
+            cls.__executor.kill()
+
+    @property
+    def executor(self):
+        """Executor instance.
+
+        :rtype: gevent.threadpool.ThreadPool
+        """
+        if not isinstance(self.__executor, gevent.threadpool.ThreadPool):
+            self.configure()
+        return self.__executor
+
+    def _get_function_wrapper(self, func):
+        """Here should be constructed and returned real decorator.
+
+        :param func: Wrapped function
+        :type func: typing.Callable
+        :return: wrapped function
+        :rtype: typing.Callable[..., gevent.event.AsyncResult]
+        """
+        # pylint: disable=missing-docstring
+        # noinspection PyMissingOrEmptyDocstring
+        @six.wraps(func)
+        def wrapper(*args, **kwargs):
+            return self.executor.spawn(func, *args, **kwargs)
+        # pylint: enable=missing-docstring
+        return wrapper

threaded/_base_threaded.py

@@ -16,6 +16,7 @@
 
 from __future__ import absolute_import
 
+import abc
 # noinspection PyCompatibility
 import concurrent.futures
 import threading
@@ -35,12 +36,47 @@ def cpu_count():
             return 1
 
 __all__ = (
+    'APIPooled',
     'BasePooled',
     'ThreadPoolExecutor',
+    'cpu_count'
 )
 
 
-class BasePooled(_class_decorator.BaseDecorator):
+class APIPooled(_class_decorator.BaseDecorator):
+    """API description for pooled."""
+
+    __slots__ = ()
+
+    __executor = None
+
+    @classmethod
+    @abc.abstractmethod
+    def configure(
+        cls,
+        max_workers=None,
+    ):
+        """Pool executor create and configure.
+
+        :param max_workers: Maximum workers
+        :type max_workers: typing.Optional[int]
+        """
+        raise NotImplementedError()  # pragma: no cover
+
+    @classmethod
+    @abc.abstractmethod
+    def shutdown(cls):
+        """Shutdown executor."""
+        raise NotImplementedError()  # pragma: no cover
+
+    @property
+    @abc.abstractmethod
+    def executor(self):
+        """Executor instance."""
+        raise NotImplementedError()  # pragma: no cover
+
+
+class BasePooled(APIPooled):
     """Base ThreadPooled class."""
 
     __slots__ = ()

threaded/_gthreadpooled2.py

@@ -0,0 +1,45 @@
+#    Copyright 2017 Alexey Stepanov aka penguinolog
+##
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Python 2 threaded implementation.
+
+Uses backport of concurrent.futures.
+"""
+
+from __future__ import absolute_import
+
+from . import _base_gthreadpooled
+
+__all__ = (
+    'GThreadPooled',
+    'gthreadpooled',
+)
+
+
+class GThreadPooled(_base_gthreadpooled.BaseGThreadPooled):
+    """Post function to gevent.threadpool.ThreadPool."""
+
+    __slots__ = ()
+
+
+# pylint: disable=unexpected-keyword-arg, no-value-for-parameter
+def gthreadpooled(func=None):
+    """Post function to gevent.threadpool.ThreadPool.
+
+    :param func: function to wrap
+    :type func: typing.Optional[typing.Callable]
+    :rtype: BaseGThreadPooled
+    """
+    return GThreadPooled(func=func)
+# pylint: enable=unexpected-keyword-arg, no-value-for-parameter

threaded/_gthreadpooled3.py

@@ -0,0 +1,78 @@
+#    Copyright 2017 Alexey Stepanov aka penguinolog
+##
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Python 3 threaded implementation.
+
+Asyncio is supported
+"""
+
+import functools
+import typing
+
+import gevent.event
+
+from . import _base_gthreadpooled
+from . import _py3_helpers
+
+__all__ = (
+    'GThreadPooled',
+    'gthreadpooled',
+)
+
+
+class GThreadPooled(_base_gthreadpooled.BaseGThreadPooled):
+    """Post function to ThreadPoolExecutor."""
+
+    __slots__ = ()
+
+    def _get_function_wrapper(
+        self,
+        func: typing.Callable
+    ) -> typing.Callable[
+        ...,
+        gevent.event.AsyncResult
+    ]:
+        """Here should be constructed and returned real decorator.
+
+        :param func: Wrapped function
+        :type func: typing.Callable
+        :return: wrapped coroutine or function
+        :rtype: typing.Callable[..., gevent.event.AsyncResult]
+        """
+        prepared = _py3_helpers.await_if_required(func)
+
+        # pylint: disable=missing-docstring
+        # noinspection PyMissingOrEmptyDocstring
+        @functools.wraps(prepared)
+        def wrapper(
+            *args, **kwargs
+        ) -> gevent.event.AsyncResult:
+            return self.executor.spawn(prepared, *args, **kwargs)
+
+        # pylint: enable=missing-docstring
+        return wrapper
+
+
+# pylint: disable=unexpected-keyword-arg, no-value-for-parameter
+def gthreadpooled(
+    func: typing.Optional[typing.Callable]=None
+) -> GThreadPooled:
+    """Post function to gevent.threadpool.ThreadPool.
+
+    :param func: function to wrap
+    :type func: typing.Optional[typing.Callable]
+    :rtype: GThreadPooled
+    """
+    return GThreadPooled(func=func)
+# pylint: enable=unexpected-keyword-arg, no-value-for-parameter

threaded/_py3_helpers.py

@@ -0,0 +1,50 @@
+#    Copyright 2017 Alexey Stepanov aka penguinolog
+##
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Python 3 related helpers."""
+
+# noinspection PyCompatibility
+import asyncio
+import functools
+import typing
+
+__all__ = (
+    'get_loop', 'await_if_required'
+)
+
+
+def get_loop(
+        self,
+        *args, **kwargs
+) -> typing.Optional[asyncio.AbstractEventLoop]:
+    """Get event loop in decorator class."""
+    if callable(self.loop_getter):
+        if self.loop_getter_need_context:
+            return self.loop_getter(*args, **kwargs)
+        return self.loop_getter()
+    return self.loop_getter
+
+
+def await_if_required(target: typing.Callable) -> typing.Callable:
+    """Await result if coroutine was returned."""
+    @functools.wraps(target)
+    def wrapper(*args, **kwargs):
+        """Decorator/wrapper."""
+        result = target(*args, **kwargs)
+        if asyncio.iscoroutine(result):
+            loop = asyncio.new_event_loop()
+            result = loop.run_until_complete(result)
+            loop.close()
+        return result
+    return wrapper

threaded/_threaded3.py

@@ -29,6 +29,7 @@
 
 from . import _base_threaded
 from . import _class_decorator
+from . import _py3_helpers
 
 __all__ = (
     'ThreadPooled',
@@ -40,32 +41,6 @@
 )
 
 
-def _get_loop(
-        self,
-        *args, **kwargs
-) -> typing.Optional[asyncio.AbstractEventLoop]:
-    """Get event loop in decorator class."""
-    if callable(self.loop_getter):
-        if self.loop_getter_need_context:
-            return self.loop_getter(*args, **kwargs)
-        return self.loop_getter()
-    return self.loop_getter
-
-
-def await_if_required(target: typing.Callable) -> typing.Callable:
-    """Await result if coroutine was returned."""
-    @functools.wraps(target)
-    def wrapper(*args, **kwargs):
-        """Decorator/wrapper."""
-        result = target(*args, **kwargs)
-        if asyncio.iscoroutine(result):
-            loop = asyncio.new_event_loop()
-            result = loop.run_until_complete(result)
-            loop.close()
-        return result
-    return wrapper
-
-
 class ThreadPooled(_base_threaded.BasePooled):
     """Post function to ThreadPoolExecutor."""
 
@@ -151,7 +126,7 @@ def _get_function_wrapper(
                     ]
                 ]
         """
-        prepared = await_if_required(func)
+        prepared = _py3_helpers.await_if_required(func)
 
         # pylint: disable=missing-docstring
         # noinspection PyMissingOrEmptyDocstring
@@ -162,7 +137,7 @@ def wrapper(
             concurrent.futures.Future,
             asyncio.Task
         ]:
-            loop = _get_loop(self, *args, **kwargs)
+            loop = _py3_helpers.get_loop(self, *args, **kwargs)
 
             if loop is None:
                 return self.executor.submit(prepared, *args, **kwargs)
@@ -210,7 +185,7 @@ def _get_function_wrapper(
         :return: wrapped function
         :rtype: typing.Callable[..., threading.Thread]
         """
-        prepared = await_if_required(func)
+        prepared = _py3_helpers.await_if_required(func)
         name = self.name
         if name is None:
             name = 'Threaded: ' + getattr(
@@ -310,7 +285,7 @@ def _get_function_wrapper(
         # noinspection PyCompatibility,PyMissingOrEmptyDocstring
         @functools.wraps(func)
         def wrapper(*args, **kwargs) -> asyncio.Task:
-            loop = _get_loop(self, *args, **kwargs)
+            loop = _py3_helpers.get_loop(self, *args, **kwargs)
             return loop.create_task(func(*args, **kwargs))
 
         # pylint: enable=missing-docstring