threaded
.. image:: https://github.com/python-useful-helpers/threaded/workflows/Python%20package/badge.svg
:target: https://github.com/python-useful-helpers/threaded/actions
.. image:: https://coveralls.io/repos/github/python-useful-helpers/threaded/badge.svg?branch=master
:target: https://coveralls.io/github/python-useful-helpers/threaded?branch=master
.. image:: https://readthedocs.org/projects/threaded/badge/?version=latest
:target: https://threaded.readthedocs.io/
:alt: Documentation Status
.. image:: https://img.shields.io/pypi/v/threaded.svg
:target: https://pypi.python.org/pypi/threaded
.. image:: https://img.shields.io/pypi/pyversions/threaded.svg
:target: https://pypi.python.org/pypi/threaded
.. image:: https://img.shields.io/pypi/status/threaded.svg
:target: https://pypi.python.org/pypi/threaded
.. image:: https://img.shields.io/github/license/python-useful-helpers/threaded.svg
:target: https://raw.githubusercontent.com/python-useful-helpers/threaded/master/LICENSE
.. image:: https://img.shields.io/badge/code%20style-black-000000.svg
:target: https://github.com/ambv/black
threaded is a set of decorators, which wrap functions in:
concurrent.futures.ThreadPool
threading.Thread
asyncio.Task
in Python 3.
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.
Pros:
Decorators:
-
ThreadPooled
- native concurrent.futures.ThreadPool
.
-
threadpooled
is alias for ThreadPooled
.
-
Threaded
- wrap in threading.Thread
.
-
threaded
is alias for Threaded
.
-
AsyncIOTask
- wrap in asyncio.Task
. Uses the same API, as ThreadPooled
.
-
asynciotask
is alias for AsyncIOTask
.
Usage
ThreadPooled
Mostly it is required decorator: submit function to ThreadPoolExecutor on call.
.. code-block:: python
threaded.ThreadPooled.configure(max_workers=3)
.. note::
By default, if executor is not configured - it configures with default parameters: ``max_workers=CPU_COUNT * 5``
.. code-block:: python
@threaded.ThreadPooled
def func():
pass
concurrent.futures.wait([func()])
Usage with asyncio:
.. note::
if `loop_getter` is not callable, `loop_getter_need_context` is ignored.
.. code-block:: python
loop = asyncio.get_event_loop()
@threaded.ThreadPooled(loop_getter=loop, loop_getter_need_context=False)
def func():
pass
loop.run_until_complete(asyncio.wait_for(func(), timeout))
Python 3.5+ usage with asyncio and loop extraction from call arguments:
.. code-block:: python
loop_getter = lambda tgt_loop: tgt_loop
@threaded.ThreadPooled(loop_getter=loop_getter, loop_getter_need_context=True) # loop_getter_need_context is required
def func(*args, **kwargs):
pass
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(func(loop), timeout))
During application shutdown, pool can be stopped (while it will be recreated automatically, if some component will request).
.. code-block:: python
threaded.ThreadPooled.shutdown()
Threaded
Classic threading.Thread
. Useful for running until close and self-closing threads without return.
Usage example:
.. code-block:: python
@threaded.Threaded
def func(*args, **kwargs):
pass
thread = func()
thread.start()
thread.join()
Without arguments, thread name will use pattern: 'Threaded: ' + func.__name__
.. note::
If func.__name__ is not accessible, str(hash(func)) will be used instead.
Override name can be don via corresponding argument:
.. code-block:: python
@threaded.Threaded(name='Function in thread')
def func(*args, **kwargs):
pass
Thread can be daemonized automatically:
.. code-block:: python
@threaded.Threaded(daemon=True)
def func(*args, **kwargs):
pass
Also, if no any addition manipulations expected before thread start,
it can be started automatically before return:
.. code-block:: python
@threaded.Threaded(started=True)
def func(*args, **kwargs):
pass
AsyncIOTask
Wrap in asyncio.Task
.
usage with asyncio:
.. code-block:: python
@threaded.AsyncIOTask
def func():
pass
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(func(), timeout))
Provide event loop directly:
.. note::
if `loop_getter` is not callable, `loop_getter_need_context` is ignored.
.. code-block:: python
loop = asyncio.get_event_loop()
@threaded.AsyncIOTask(loop_getter=loop)
def func():
pass
loop.run_until_complete(asyncio.wait_for(func(), timeout))
Usage with loop extraction from call arguments:
.. code-block:: python
loop_getter = lambda tgt_loop: tgt_loop
@threaded.AsyncIOTask(loop_getter=loop_getter, loop_getter_need_context=True)
def func(*args, **kwargs):
pass
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(func(loop), timeout))
Testing
The main test mechanism for the package threaded
is using tox
.
Available environments can be collected via tox -l
CI systems
For code checking several CI systems is used in parallel:
GitHub actions: <https://github.com/python-useful-helpers/threaded/actions>
_ is used for checking: PEP8, pylint, bandit, installation possibility and unit tests.coveralls: <https://coveralls.io/github/python-useful-helpers/threaded>
_ is used for coverage display.
CD system
GitHub actions: <https://github.com/python-useful-helpers/threaded/actions>
_ is used for package delivery on PyPI.